from __future__ import annotations import gzip import importlib.util import json from pathlib import Path import sys from shapely.geometry import box, mapping, shape ROOT = Path(__file__).resolve().parents[2] SCRIPTS = ROOT / "scripts" if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) def load_script(): path = SCRIPTS / "provision_regional_historical_landuse.py" spec = importlib.util.spec_from_file_location("test_provision_regional_historical_landuse", path) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class FakeResponse: status_code = 200 ok = True text = "" def __init__(self, payload): self.payload = payload self.content = json.dumps(payload, separators=(",", ":")).encode("utf-8") def raise_for_status(self): return None def json(self): return self.payload class FakeSession: def __init__(self, responses): self.responses = list(responses) self.calls = [] def post(self, url, **kwargs): self.calls.append((url, kwargs)) if not self.responses: raise AssertionError("Unexpected source/API request") return self.responses.pop(0) def source_feature(feature_id: str, geometry, landuse_class: str = "bebouwing"): return { "type": "Feature", "id": feature_id, "geometry": mapping(geometry), "properties": {"KLASSE": landuse_class}, } def test_member_boundaries_require_every_approved_municipality(tmp_path: Path) -> None: module = load_script() scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"] features = [] for index, member in enumerate(scope.members): features.append( { "type": "Feature", "geometry": mapping(box(index, 0, index + 0.9, 0.9)), "properties": {"nis_code": member.nis_code, "municipality": member.name}, } ) path = tmp_path / "members.geojson" path.write_text(json.dumps({"type": "FeatureCollection", "features": features}), encoding="utf-8") boundaries = module.load_member_boundaries(path, scope) assert list(boundaries) == list(scope.nis_codes) assert len(boundaries) == 28 assert boundaries["13025"][0].name == "Mol" def test_partition_retains_exact_source_response_and_clips_to_municipality(tmp_path: Path) -> None: module = load_script() definition = next(item for item in module.THEMES if item.key == "buildings") member = module.ScopeMember("Mol", "13025") boundary = box(5.0, 51.0, 5.1, 51.1) payload = { "type": "FeatureCollection", "features": [source_feature("Lgbrk1778.1", box(4.95, 51.02, 5.05, 51.08))], } session = FakeSession([FakeResponse(payload)]) manifest = module.prepare_partition( session, output_root=tmp_path, year=1778, definition=definition, scope_key="kempen-transport-region", member=member, boundary=boundary, page_size=50, max_features=100, simplify_tolerance_degrees=0.0, timeout=30, force=False, ) output = json.loads(Path(manifest["output_path"]).read_text(encoding="utf-8")) feature = output["features"][0] raw_path = Path(manifest["output_path"]).parent / manifest["raw_pages"][0]["artifact_path"] assert manifest["feature_count"] == 1 assert manifest["source_feature_count"] == 1 assert feature["id"] == "Lgbrk1778.1:13025" assert feature["properties"]["original_source_feature_id"] == "Lgbrk1778.1" assert feature["properties"]["coverage_scope"] == "kempen-transport-region" assert shape(feature["geometry"]).bounds == (5.0, 51.02, 5.05, 51.08) assert gzip.decompress(raw_path.read_bytes()) == FakeResponse(payload).content cached = module.prepare_partition( FakeSession([]), output_root=tmp_path, year=1778, definition=definition, scope_key="kempen-transport-region", member=member, boundary=boundary, page_size=50, max_features=100, simplify_tolerance_degrees=0.0, timeout=30, force=False, ) assert cached["output_sha256"] == manifest["output_sha256"] def test_regional_snapshot_assembles_unique_partition_features(tmp_path: Path) -> None: module = load_script() scope = module.GeographicScope( key="test-region", display_name="Test region", project_name="Test", project_region="Test", area_name="Test area", authority_name="Authority", authority_url="https://example.test", scope_type="test", limitation_message="Test only", members=(module.ScopeMember("Left", "10001"), module.ScopeMember("Right", "10002")), ) definition = next(item for item in module.THEMES if item.key == "water") partitions = [] for index, member in enumerate(scope.members): path, _manifest_path, _raw_dir = module.partition_paths(tmp_path, 1873, definition.key, member.nis_code) feature = source_feature(f"water.{index}:{member.nis_code}", box(index, 0, index + 0.5, 0.5), "water") module.atomic_write_json(path, {"type": "FeatureCollection", "features": [feature]}) partitions.append( { "municipality": member.name, "nis_code": member.nis_code, "source_feature_count": 1, "feature_count": 1, "raw_pages": [{"artifact_path": "unused"}], "output_path": str(path), "output_sha256": module.sha256_file(path), } ) output_path, manifest = module.assemble_snapshot( output_root=tmp_path, scope=scope, year=1873, definition=definition, partitions=partitions, max_total_features=10, ) payload = json.loads(output_path.read_text(encoding="utf-8")) assert manifest["coverage_complete"] is True assert manifest["feature_count"] == 2 assert manifest["empty_partitions"] == [] assert len({feature["id"] for feature in payload["features"]}) == 2 def test_upload_contract_is_regional_temporal_and_partition_audited(tmp_path: Path) -> None: module = load_script() scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"] definition = next(item for item in module.THEMES if item.key == "roads") path = tmp_path / "roads.geojson" path.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8") response_payload = {"data": {"id": "dataset-id", "feature_count": 42}} session = FakeSession([FakeResponse(response_payload)]) manifest = { "coverage_complete": True, "empty_partitions": [], "partitions": [{} for _ in scope.members], "partition_identity_sha256": "partition-hash", "output_sha256": "output-hash", "generated_at": "2026-07-15T00:00:00+00:00", } result = module.upload_snapshot( session, base_url="http://backend:8000", project_id="project-id", area_id="area-id", scope=scope, year=1969, definition=definition, path=path, manifest=manifest, simplify_tolerance_degrees=0.00001, timeout=30, ) data = session.calls[0][1]["data"] source_metadata = json.loads(data["source_metadata_json"]) provenance = json.loads(data["provenance_metadata_json"]) assert result["id"] == "dataset-id" assert data["area_id"] == "area-id" assert data["temporal_series_key"].endswith(":roads:kempen-transport-region") assert data["observed_at"] == "1969-01-01T00:00:00Z" assert source_metadata["member_count"] == 28 assert source_metadata["partitioned_source_audit"] is True assert source_metadata["geometry_clipped_to_area"] is True assert source_metadata["identity_stable"] is False assert source_metadata["semantic_metrics"] is False assert source_metadata["selection_aggregation"]["metric_key"] == "roads_area" assert source_metadata["selection_aggregation"]["label"] == "Oppervlakte historische wegen" assert provenance["partition_count"] == 28 assert provenance["raw_source_responses_retained"] is True assert provenance["geometry_clipped_to_area"] is True def test_regional_historical_operator_is_packaged_and_release_checked() -> 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") app = (ROOT / "frontend/src/App.tsx").read_text(encoding="utf-8") workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8") assert "COPY scripts/provision_regional_historical_landuse.py" in dockerfile assert "py_compile scripts/provision_regional_historical_landuse.py" in readiness assert "onSetContextSourceLabel={setMapContextSourceLabel}" in app assert "analysisMode === 'evolution' ? activeTemporalSeriesGroup?.label" in workspace