Add regional DOV soil operator

backend/tests/test_sprint217_regional_dov_soil_map.py

@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+import gzip
+import importlib.util
+import json
+from pathlib import Path
+import sys
+
+from shapely.geometry import box, mapping, shape
+from shapely.ops import transform as transform_geometry
+
+from app.models import Dataset
+from app.services.vector_feature_service import VectorFeatureService
+
+
+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_soil_map.py"
+    spec = importlib.util.spec_from_file_location("test_provision_regional_soil_map", 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
+
+    def json(self):
+        return self.payload
+
+
+class FakeApiSession:
+    def __init__(self, payload):
+        self.payload = payload
+        self.calls = []
+
+    def post(self, url, **kwargs):
+        self.calls.append((url, kwargs))
+        return FakeResponse({"data": self.payload})
+
+
+def source_feature(feature_id: str, geometry) -> dict:
+    return {
+        "type": "Feature",
+        "id": feature_id,
+        "geometry": mapping(geometry),
+        "properties": {
+            "gid": 1,
+            "id_kaartvlak": 10,
+            "Bodemtype": "Zeg",
+            "Unibodemtype": "Zeg",
+            "Bodemserie": "Zeg",
+            "Beknopte_omschrijving_bodemserie": "Natte zandbodem",
+            "Gegeneraliseerde_legende": "Nat zand",
+            "Textuurklasse_code": "Z",
+            "Textuurklasse": "zand",
+            "Drainageklasse_code": "e",
+            "Drainageklasse": "nat",
+            "Profielontwikkelingsgroep_code": "g",
+            "Profielontwikkelingsgroep": "humus B horizont",
+            "Eenduidige_legende_titel": "bodemserie Zeg",
+        },
+    }
+
+
+def test_normalized_regional_features_keep_member_identity_and_unique_ids() -> None:
+    module = load_script()
+    boundary = box(5.0, 51.0, 5.1, 51.1)
+    boundary_lambert72 = transform_geometry(module.soil.TO_LAMBERT72.transform, boundary)
+    feature = source_feature("bodemtypes.1", box(4.98, 51.02, 5.05, 51.08))
+
+    left, _ = module.soil.normalize_feature(
+        feature,
+        boundary_lambert72,
+        municipality="Mol",
+        nis_code="13025",
+        coverage_scope="test-region",
+        feature_id_suffix="13025",
+    )
+    right, _ = module.soil.normalize_feature(
+        feature,
+        boundary_lambert72,
+        municipality="Balen",
+        nis_code="13003",
+        coverage_scope="test-region",
+        feature_id_suffix="13003",
+    )
+
+    assert left is not None and right is not None
+    assert left["id"] == "bodemtypes.1:13025"
+    assert right["id"] == "bodemtypes.1:13003"
+    assert left["properties"]["municipality"] == "Mol"
+    assert left["properties"]["coverage_scope"] == "test-region"
+    assert shape(left["geometry"]).within(boundary.buffer(1e-7))
+
+
+def test_partition_retains_gzipped_source_and_is_checksum_reusable(tmp_path: Path, monkeypatch) -> 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="Test",
+        authority_url="https://example.test",
+        scope_type="test",
+        limitation_message="Test",
+        members=(module.ScopeMember("Mol", "13025"),),
+    )
+    boundary = box(5.0, 51.0, 5.1, 51.1)
+    payload = {
+        "type": "FeatureCollection",
+        "features": [source_feature("bodemtypes.1", box(4.98, 51.02, 5.05, 51.08))],
+    }
+    raw_bytes = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+    monkeypatch.setattr(
+        module.soil,
+        "iter_wfs_pages",
+        lambda *_args, **_kwargs: iter([(payload, "https://example.test/page", raw_bytes)]),
+    )
+
+    manifest = module.prepare_partition(
+        object(),
+        output_root=tmp_path,
+        scope=scope,
+        member=scope.members[0],
+        boundary_wgs84=boundary,
+        page_limit=100,
+        max_features=1000,
+        timeout=30,
+        force=False,
+    )
+    output = json.loads(Path(manifest["output_path"]).read_text(encoding="utf-8"))
+    raw_path = Path(manifest["manifest_path"]).parent / manifest["raw_pages"][0]["artifact_path"]
+
+    assert manifest["feature_count"] == 1
+    assert output["features"][0]["id"] == "bodemtypes.1:13025"
+    assert gzip.decompress(raw_path.read_bytes()) == raw_bytes
+
+    cached = module.prepare_partition(
+        object(),
+        output_root=tmp_path,
+        scope=scope,
+        member=scope.members[0],
+        boundary_wgs84=boundary,
+        page_limit=100,
+        max_features=1000,
+        timeout=30,
+        force=False,
+    )
+    assert cached["output_sha256"] == manifest["output_sha256"]
+
+
+def test_snapshot_assembles_all_partitions_with_governed_area_metrics(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="Test",
+        authority_url="https://example.test",
+        scope_type="test",
+        limitation_message="Test",
+        members=(module.ScopeMember("Left", "10001"), module.ScopeMember("Right", "10002")),
+    )
+    partitions = []
+    for index, member in enumerate(scope.members):
+        output_path, manifest_path, _raw_dir = module.partition_paths(tmp_path / scope.key, member.nis_code)
+        feature = source_feature(f"bodemtypes.{index}", box(5.0 + index * 0.1, 51.0, 5.05 + index * 0.1, 51.05))
+        feature["id"] = f"bodemtypes.{index}:{member.nis_code}"
+        feature["properties"].update(
+            {
+                "clipped_area_ha": 1.0 + index,
+                "soil_generalized_legend": "Nat zand",
+                "soil_texture_class": "zand",
+                "soil_drainage_class": "nat",
+            }
+        )
+        module.soil.write_json_atomic(output_path, {"type": "FeatureCollection", "features": [feature]})
+        partitions.append(
+            {
+                "municipality": member.name,
+                "nis_code": member.nis_code,
+                "feature_count": 1,
+                "raw_source_feature_count": 1,
+                "page_count": 1,
+                "output_path": str(output_path),
+                "output_sha256": module.soil.sha256_file(output_path),
+                "manifest_path": str(manifest_path),
+            }
+        )
+
+    output_path, _manifest_path, manifest = module.assemble_snapshot(
+        output_root=tmp_path,
+        scope=scope,
+        partitions=partitions,
+        member_boundaries_sha256="boundaries-hash",
+        max_total_features=10,
+    )
+    output = json.loads(output_path.read_text(encoding="utf-8"))
+
+    assert manifest["coverage_complete"] is True
+    assert manifest["feature_count"] == 2
+    assert manifest["area_by_generalized_legend_ha"]["Nat zand"] == 3.0
+    assert manifest["area_by_texture_ha"]["zand"] == 3.0
+    assert len({feature["id"] for feature in output["features"]}) == 2
+
+
+def test_upload_contract_is_regional_historical_and_canonical(tmp_path: Path) -> None:
+    module = load_script()
+    scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
+    path = tmp_path / "soil.geojson"
+    path.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
+    manifest_path = tmp_path / "manifest.json"
+    manifest = {
+        "coverage_complete": True,
+        "feature_count": 42,
+        "output_sha256": "output-hash",
+        "partition_identity_sha256": "partition-hash",
+        "partitions": [{} for _ in scope.members],
+        "generated_at": "2026-07-16T00:00:00+00:00",
+        "limitations": ["historical"],
+    }
+    session = FakeApiSession({"id": "dataset-id", "feature_count": 42})
+
+    result = module.upload_snapshot(
+        session,
+        base_url="http://backend:8000",
+        project_id="project-id",
+        area_id="area-id",
+        scope=scope,
+        path=path,
+        manifest_path=manifest_path,
+        manifest=manifest,
+        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"] == "dov:digital-soil-map:kempen-transport-region"
+    assert data["valid_from"] == module.soil.VALID_FROM
+    assert data["valid_to"] == module.soil.VALID_TO
+    assert source_metadata["coverage_scope"] == "kempen-transport-region"
+    assert source_metadata["member_count"] == 28
+    assert source_metadata["authority_level"] == "authoritative_historical_baseline"
+    assert provenance["operator_tool"] == "provision_regional_soil_map.py"
+    assert provenance["raw_source_responses_retained"] is True
+
+
+def test_regional_operator_is_packaged_release_checked_and_query_optimized() -> 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")
+    service = (ROOT / "backend/app/services/vector_feature_service.py").read_text(encoding="utf-8")
+
+    assert "COPY scripts/provision_regional_soil_map.py" in dockerfile
+    assert "py_compile scripts/provision_regional_soil_map.py" in readiness
+    assert '"provision_regional_soil_map.py"' in service
+
+    dataset = Dataset(
+        name="regional-soil.geojson",
+        dataset_type="vector",
+        status="ready",
+        source_metadata={"partitioned_source_audit": True, "geometry_clipped_to_area": True},
+        provenance_metadata={"operator_tool": "provision_regional_soil_map.py"},
+    )
+    assert VectorFeatureService.preclipped_partition_filter(
+        dataset, "Gemeente Mol - officiele grens"
+    ) == ("municipality", "Mol")
+    assert VectorFeatureService.preclipped_partition_filter(dataset, "Vervoerregio Kempen") is None