geointel/backend/tests/test_sprint214_dov_soil_map.py

from __future__ import annotations

import importlib.util
from pathlib import Path
from uuid import uuid4

import pytest
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
from tests.frontend_contract import read_map_workspace


ROOT = Path(__file__).resolve().parents[2]


def load_operator():
    path = ROOT / "scripts" / "provision_mol_soil_map.py"
    spec = importlib.util.spec_from_file_location("dov_soil_map_operator", path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class FakeResponse:
    def __init__(self, payload: dict, url: str):
        self._payload = payload
        self.url = url
        self.content = b'{"type":"FeatureCollection"}'

    def raise_for_status(self) -> None:
        return None

    def json(self) -> dict:
        return self._payload


class FakeSession:
    def __init__(self, pages: list[dict]):
        self.pages = pages
        self.calls: list[dict] = []

    def get(self, _url: str, *, params: dict, timeout: int):
        self.calls.append({"params": dict(params), "timeout": timeout})
        return FakeResponse(self.pages[len(self.calls) - 1], f"https://example.test/page/{len(self.calls)}")


def soil_feature(module, feature_id: str = "bodemtypes.1") -> dict:
    return {
        "type": "Feature",
        "id": feature_id,
        "geometry": mapping(box(5.0, 51.0, 5.02, 51.02)),
        "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_wfs_pagination_is_bounded_complete_and_deterministic() -> None:
    module = load_operator()
    feature = soil_feature(module)
    pages = [
        {
            "type": "FeatureCollection",
            "numberMatched": 3,
            "numberReturned": 2,
            "features": [feature, {**feature, "id": "bodemtypes.2"}],
        },
        {
            "type": "FeatureCollection",
            "numberMatched": 3,
            "numberReturned": 1,
            "features": [{**feature, "id": "bodemtypes.3"}],
        },
    ]
    session = FakeSession(pages)

    result = list(
        module.iter_wfs_pages(
            session,
            (196000.0, 205000.0, 211000.0, 224000.0),
            page_limit=2,
            timeout=30,
        )
    )

    assert len(result) == 2
    assert [call["params"]["startIndex"] for call in session.calls] == ["0", "2"]
    assert all(call["params"]["typeNames"] == "bodemkaart:bodemtypes" for call in session.calls)
    assert all(call["params"]["bbox"].endswith("EPSG:31370") for call in session.calls)
    assert all(call["params"]["sortBy"] == "gid" for call in session.calls)


def test_soil_feature_is_exactly_clipped_and_keeps_governed_properties() -> None:
    module = load_operator()
    boundary_wgs84 = box(5.005, 51.005, 5.015, 51.015)
    boundary_lambert72 = transform_geometry(module.TO_LAMBERT72.transform, boundary_wgs84)

    normalized, was_clipped = module.normalize_feature(soil_feature(module), boundary_lambert72)

    assert normalized is not None and was_clipped is True
    persisted_geometry = shape(normalized["geometry"])
    assert persisted_geometry.within(boundary_wgs84.buffer(1e-7))
    properties = normalized["properties"]
    assert properties["source_name"] == "dov_soil_map"
    assert properties["soil_texture_class"] == "zand"
    assert properties["soil_drainage_class"] == "nat"
    assert properties["survey_period"] == "1949-1971"
    assert properties["clipped_area_ha"] > 0
    assert "may differ today" in properties["historical_drainage_limitation"]


def test_soil_map_uses_existing_semantic_selection_architecture() -> None:
    dataset = Dataset(
        id=uuid4(),
        project_id=uuid4(),
        name="dov_soil_map_mol.geojson",
        dataset_type="vector",
        source="operator_official_import",
        source_name="dov_soil_map",
        reference_layer_name="soil",
        source_metadata={
            "theme": "soil",
            "selection_aggregation": {
                "method": "intersection_area",
                "label": "Bodemkaartoppervlakte",
                "unit": "ha",
            },
        },
        status="ready",
    )

    assert VectorFeatureService._dataset_theme(dataset) == "soil"
    assert VectorFeatureService.supports_selection_summary(dataset) is True
    assert VectorFeatureService.can_use_full_area_fast_path(dataset, None) is False


def test_soil_operator_contract_has_no_direct_persistence_and_is_packaged() -> None:
    operator = (ROOT / "scripts" / "provision_mol_soil_map.py").read_text(encoding="utf-8")
    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")
    map_workspace = read_map_workspace()

    assert "/datasets/upload" in operator
    assert "vector_features" in operator
    assert "does not write directly" in " ".join(operator.split())
    assert "SessionLocal" not in operator and "INSERT INTO" not in operator
    assert "COPY scripts/provision_mol_soil_map.py" in dockerfile
    assert "py_compile scripts/provision_mol_soil_map.py" in readiness
    assert "id: 'soil'" in map_workspace
    assert "dataset.source_name === 'dov_soil_map'" in map_workspace


def test_incomplete_wfs_pagination_fails_closed() -> None:
    module = load_operator()
    session = FakeSession(
        [
            {
                "type": "FeatureCollection",
                "numberMatched": 2,
                "numberReturned": 0,
                "features": [],
            }
        ]
    )

    with pytest.raises(RuntimeError, match="returned 0 of 2"):
        list(module.iter_wfs_pages(session, (0.0, 0.0, 1.0, 1.0), page_limit=100, timeout=30))