263 lines
10 KiB
Python
263 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
import pytest
|
|
from shapely.geometry import LineString, Polygon, shape
|
|
from shapely.ops import unary_union
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def load_operator():
|
|
scripts_path = str(SCRIPTS)
|
|
if scripts_path not in sys.path:
|
|
sys.path.insert(0, scripts_path)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"provision_regional_grb_context_test",
|
|
SCRIPTS / "provision_regional_grb_context.py",
|
|
)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def source_feature(feature_id: str, geometry) -> dict:
|
|
return {
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": geometry.__geo_interface__,
|
|
"properties": {"UIDN": feature_id},
|
|
}
|
|
|
|
|
|
def test_layer_registry_matches_verified_official_grb_collections() -> None:
|
|
operator = load_operator()
|
|
|
|
assert [item.key for item in operator.LAYERS] == ["roads", "water", "parcels"]
|
|
assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["roads"].collections] == [
|
|
("Wegsegment", 1)
|
|
]
|
|
assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["water"].collections] == [
|
|
("WTZ", 2),
|
|
("WLAS", 1),
|
|
("WGR", 1),
|
|
]
|
|
assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["parcels"].collections] == [
|
|
("ADP", 2)
|
|
]
|
|
assert [item.key for item in operator.selected_definitions("parcels,roads")] == ["roads", "parcels"]
|
|
with pytest.raises(ValueError, match="Unsupported layers"):
|
|
operator.selected_definitions("buildings")
|
|
|
|
|
|
def test_layer_selection_accepts_space_and_comma_cli_forms() -> None:
|
|
operator = load_operator()
|
|
|
|
assert [item.key for item in operator.selected_definitions(["roads", "water", "parcels"])] == [
|
|
"roads",
|
|
"water",
|
|
"parcels",
|
|
]
|
|
assert [item.key for item in operator.selected_definitions("roads,parcels")] == ["roads", "parcels"]
|
|
with pytest.raises(ValueError, match="Unsupported layers"):
|
|
operator.selected_definitions(["roads", "imaginary"])
|
|
|
|
|
|
def test_line_owner_uses_intersection_length_and_deterministic_tie_break() -> None:
|
|
operator = load_operator()
|
|
scopes = __import__("geographic_scopes")
|
|
alpha = scopes.ScopeMember("Alpha", "10001")
|
|
beta = scopes.ScopeMember("Beta", "10002")
|
|
alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)])
|
|
members = [(alpha, alpha_boundary), (beta, beta_boundary)]
|
|
|
|
assert operator.assign_owner_nis(
|
|
LineString([(0.7, 0.5), (1.1, 0.5)]),
|
|
members,
|
|
expected_dimension=1,
|
|
) == "10001"
|
|
assert operator.assign_owner_nis(
|
|
LineString([(0.8, 0.5), (1.2, 0.5)]),
|
|
members,
|
|
expected_dimension=1,
|
|
) == "10001"
|
|
|
|
|
|
def test_mixed_water_partition_preserves_dimensions_and_source_identity() -> None:
|
|
operator = load_operator()
|
|
scopes = __import__("geographic_scopes")
|
|
member = scopes.ScopeMember("Alpha", "10001")
|
|
scope = scopes.GeographicScope(
|
|
key="test-region",
|
|
display_name="Test region",
|
|
project_name="Test project",
|
|
project_region="Test",
|
|
area_name="Test boundary",
|
|
authority_name="Test",
|
|
authority_url="https://example.test/scope",
|
|
scope_type="policy_region",
|
|
limitation_message="Test only.",
|
|
members=(member,),
|
|
)
|
|
boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
definition = operator.LAYER_BY_KEY["water"]
|
|
pages = [
|
|
(
|
|
definition.collections[0],
|
|
{"type": "FeatureCollection", "features": [source_feature("WTZ.1", Polygon([(0.1, 0.1), (0.3, 0.1), (0.3, 0.3), (0.1, 0.3)]))]},
|
|
"https://example.test/wtz",
|
|
),
|
|
(
|
|
definition.collections[1],
|
|
{"type": "FeatureCollection", "features": [source_feature("WLAS.1", LineString([(-0.2, 0.5), (0.5, 0.5)]))]},
|
|
"https://example.test/wlas",
|
|
),
|
|
(
|
|
definition.collections[2],
|
|
{"type": "FeatureCollection", "features": [source_feature("WGR.1", LineString([(0.4, 0.7), (0.8, 0.7)]))]},
|
|
"https://example.test/wgr",
|
|
),
|
|
]
|
|
|
|
features, summary = operator.build_partition_features(
|
|
pages,
|
|
definition=definition,
|
|
member=member,
|
|
members=[(member, boundary)],
|
|
regional_boundary=boundary,
|
|
scope=scope,
|
|
max_features=10,
|
|
)
|
|
|
|
assert [item["id"] for item in features] == ["WTZ:WTZ.1", "WLAS:WLAS.1", "WGR:WGR.1"]
|
|
assert [shape(item["geometry"]).geom_type for item in features] == ["Polygon", "LineString", "LineString"]
|
|
assert shape(features[1]["geometry"]).bounds == (0.0, 0.5, 0.5, 0.5)
|
|
assert features[1]["properties"]["clipped_to_regional_scope"] is True
|
|
assert summary["features_by_collection"] == {"WTZ": 1, "WLAS": 1, "WGR": 1}
|
|
assert summary["reference_truncated"] is False
|
|
|
|
|
|
def test_context_partition_rejects_missing_official_source_identity() -> None:
|
|
operator = load_operator()
|
|
scopes = __import__("geographic_scopes")
|
|
member = scopes.ScopeMember("Alpha", "10001")
|
|
scope = scopes.GeographicScope(
|
|
key="test-region",
|
|
display_name="Test region",
|
|
project_name="Test project",
|
|
project_region="Test",
|
|
area_name="Test boundary",
|
|
authority_name="Test",
|
|
authority_url="https://example.test/scope",
|
|
scope_type="policy_region",
|
|
limitation_message="Test only.",
|
|
members=(member,),
|
|
)
|
|
boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
definition = operator.LAYER_BY_KEY["roads"]
|
|
missing_identity = source_feature("", LineString([(0.1, 0.1), (0.2, 0.2)]))
|
|
|
|
with pytest.raises(RuntimeError, match="missing an official source identity"):
|
|
operator.build_partition_features(
|
|
[(definition.collections[0], {"type": "FeatureCollection", "features": [missing_identity]}, "https://example.test/roads")],
|
|
definition=definition,
|
|
member=member,
|
|
members=[(member, boundary)],
|
|
regional_boundary=boundary,
|
|
scope=scope,
|
|
max_features=10,
|
|
)
|
|
|
|
|
|
def test_polygon_partition_assignment_does_not_duplicate_cross_boundary_parcel() -> None:
|
|
operator = load_operator()
|
|
scopes = __import__("geographic_scopes")
|
|
alpha = scopes.ScopeMember("Alpha", "10001")
|
|
beta = scopes.ScopeMember("Beta", "10002")
|
|
scope = scopes.GeographicScope(
|
|
key="test-region",
|
|
display_name="Test region",
|
|
project_name="Test project",
|
|
project_region="Test",
|
|
area_name="Test boundary",
|
|
authority_name="Test",
|
|
authority_url="https://example.test/scope",
|
|
scope_type="policy_region",
|
|
limitation_message="Test only.",
|
|
members=(alpha, beta),
|
|
)
|
|
alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
|
|
beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)])
|
|
members = [(alpha, alpha_boundary), (beta, beta_boundary)]
|
|
region = unary_union([alpha_boundary, beta_boundary])
|
|
definition = operator.LAYER_BY_KEY["parcels"]
|
|
crossing = source_feature("ADP.1", Polygon([(0.7, 0.2), (1.1, 0.2), (1.1, 0.6), (0.7, 0.6)]))
|
|
page = (definition.collections[0], {"type": "FeatureCollection", "features": [crossing]}, "https://example.test/adp")
|
|
|
|
alpha_features, _ = operator.build_partition_features(
|
|
[page], definition=definition, member=alpha, members=members, regional_boundary=region, scope=scope, max_features=10
|
|
)
|
|
with pytest.raises(RuntimeError, match="No GRB parcels"):
|
|
operator.build_partition_features(
|
|
[page], definition=definition, member=beta, members=members, regional_boundary=region, scope=scope, max_features=10
|
|
)
|
|
|
|
assert [item["id"] for item in alpha_features] == ["ADP:ADP.1"]
|
|
|
|
|
|
def test_combined_context_artifact_rejects_duplicate_source_identity(tmp_path: Path) -> None:
|
|
operator = load_operator()
|
|
scope = __import__("geographic_scopes").KEMPEN_TRANSPORT_REGION_SCOPE
|
|
definition = operator.LAYER_BY_KEY["roads"]
|
|
feature = source_feature("Wegsegment:Wegsegment.1", LineString([(4.9, 51.1), (4.91, 51.11)]))
|
|
first = tmp_path / "first.geojson"
|
|
second = tmp_path / "second.geojson"
|
|
first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature]}), encoding="utf-8")
|
|
second.write_text(json.dumps({"type": "FeatureCollection", "features": [source_feature("Wegsegment:Wegsegment.2", LineString([(5.0, 51.2), (5.01, 51.21)]))]}), encoding="utf-8")
|
|
combined = tmp_path / "combined.geojson"
|
|
|
|
summary = operator.write_combined_artifact(
|
|
combined,
|
|
definition=definition,
|
|
scope=scope,
|
|
observed_date=operator.date(2026, 7, 14),
|
|
partition_paths=[first, second],
|
|
expected_feature_count=2,
|
|
)
|
|
assert summary["feature_count"] == 2
|
|
|
|
second.write_text(first.read_text(encoding="utf-8"), encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="Duplicate regional source feature"):
|
|
operator.write_combined_artifact(
|
|
combined,
|
|
definition=definition,
|
|
scope=scope,
|
|
observed_date=operator.date(2026, 7, 14),
|
|
partition_paths=[first, second],
|
|
expected_feature_count=2,
|
|
)
|
|
|
|
|
|
def test_context_operator_is_packaged_and_uses_existing_service_boundary() -> None:
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
|
|
operator = (SCRIPTS / "provision_regional_grb_context.py").read_text(encoding="utf-8")
|
|
|
|
assert "provision_regional_grb_context.py" in dockerfile
|
|
assert "py_compile scripts/provision_regional_grb_context.py" in readiness
|
|
assert "DatasetService.import_partitioned_vector_artifact" in operator
|
|
assert '"identity_stable": True' in operator
|
|
assert '"identity_scheme": "grb_ogc_feature_id"' in operator
|
|
assert "insert into vector_features" not in operator.lower()
|
|
assert "db.add(VectorFeature" not in operator
|