from __future__ import annotations import json from pathlib import Path from types import SimpleNamespace from urllib.parse import parse_qs, urlparse from uuid import uuid4 import pytest from fastapi.testclient import TestClient from geoalchemy2.shape import from_shape from shapely.geometry import MultiPolygon, Polygon from app.core.config import Settings from app.core.errors import AppError from app.db.session import get_db from app.main import app from app.models import Area, Dataset, Job, Project from app.schemas.bathymetry import BathymetryProfileAcquireRequest from app.services.bathymetry_profile_acquisition_service import BathymetryProfileAcquisitionService from app.services.dataset_service import DatasetService ROOT = Path(__file__).resolve().parents[2] class FakeQuery: def __init__(self, result=None): self.result = result def filter(self, *_args): return self def order_by(self, *_args): return self def first(self): return self.result def all(self): return self.result if isinstance(self.result, list) else [] class FakeSession: def __init__(self, rows=None, query_result=None): self.rows = rows or {} self.query_result = query_result self.added = [] def get(self, model, row_id): row = self.rows.get((model, row_id)) if row is not None: return row return next((item for item in self.added if isinstance(item, model) and item.id == row_id), None) def add(self, row): self.added.append(row) def commit(self): return None def rollback(self): return None def refresh(self, row): return row def query(self, _model): return FakeQuery(self.query_result) class JsonResponse: def __init__(self, payload): self.content = json.dumps(payload).encode("utf-8") def __enter__(self): return self def __exit__(self, *_args): return False def read(self, size=-1): return self.content if size < 0 else self.content[:size] def request(*, area_id=None, force_refresh=True) -> BathymetryProfileAcquireRequest: return BathymetryProfileAcquireRequest( bbox={ "min_x": 5.0, "min_y": 51.0, "max_x": 6.0, "max_y": 52.0, "crs": "EPSG:4326", }, area_id=area_id, force_refresh=force_refresh, ) def profile(object_id, vhag, x, y, *, depth=None, document=None, measured_at=951868800000): return { "attributes": { "OBJECTID": object_id, "vhag": vhag, "atlaspunt": str(object_id), "opg_kruinb": 4.5 if depth is not None else None, "opg_vloerb": 1.2 if depth is not None else None, "d_opmeti": measured_at, "hyperlink": document, "bron": 4, "kunstwerkid": f"structure-{object_id}", "opg_diepte": depth, }, "geometry": {"x": x, "y": y}, } def provider_opener(*, count=3): profiles = [ profile( 1, 8506, 5.2, 51.2, depth=1.8, document="http://vha.waterinfo.be/download/dwarsprofielen/Molse_Nete/8506_DP_1.pdf", ), profile(2, 8634, 5.8, 51.8, depth=2.4), profile(3, 8506, 5.2, 51.8), ] def opener(raw_request, timeout): assert timeout == 120 url = raw_request.full_url query = parse_qs(urlparse(url).query) if query.get("returnCountOnly") == ["true"]: return JsonResponse({"count": count}) if "MapServer/1/query" in url: return JsonResponse( { "features": [ { "attributes": { "wlasvl.vhag": 8506, "VHAG_TABEL.naam": "Molse Nete", "VHAG_TABEL.namen": "Molse Nete - Mol Neet", } }, { "attributes": { "wlasvl.vhag": 8634, "VHAG_TABEL.naam": "Scheppelijke Nete", "VHAG_TABEL.namen": "Scheppelijke Nete - Stevensloop", } }, ] } ) return JsonResponse({"features": profiles[:count]}) return opener def test_bathymetry_source_registry_is_honest_and_nationally_extensible() -> None: sources = BathymetryProfileAcquisitionService.list_sources() by_key = {item["key"]: item for item in sources} assert set(by_key) == { "vha_inland_profiles", "mdk_bcp_bathymetry", "spw_walloon_waterway_bathymetry", "port_antwerp_bathymetry", } assert by_key["vha_inland_profiles"]["integration_status"] == "operational" assert by_key["vha_inland_profiles"]["acquisition_supported"] is True assert by_key["mdk_bcp_bathymetry"]["vertical_reference"] == "LAT" assert by_key["mdk_bcp_bathymetry"]["acquisition_supported"] is False assert by_key["spw_walloon_waterway_bathymetry"]["vertical_reference"] == "mDNG" assert by_key["spw_walloon_waterway_bathymetry"]["license_note"].startswith("CC BY 4.0") def test_bathymetry_normalization_exactly_clips_area_and_preserves_evidence() -> None: l_shape = Polygon( [ (5.0, 51.0), (6.0, 51.0), (6.0, 51.4), (5.4, 51.4), (5.4, 52.0), (5.0, 52.0), (5.0, 51.0), ] ) raw, _provenance = BathymetryProfileAcquisitionService._fetch_profiles( (5.0, 51.0, 6.0, 52.0), Settings(_env_file=None), provider_opener(), ) names, _urls, _hashes = BathymetryProfileAcquisitionService._fetch_watercourse_names( {8506, 8634}, Settings(_env_file=None), provider_opener(), ) collection, summary = BathymetryProfileAcquisitionService._normalize_features(raw, l_shape, names) assert summary == { "profile_count": 2, "document_count": 1, "structured_depth_count": 1, "structured_width_count": 1, "watercourse_count": 1, "measurement_date_min": "2000-03-01", "measurement_date_max": "2000-03-01", } assert {feature["id"] for feature in collection["features"]} == {"1", "3"} first = collection["features"][0]["properties"] assert first["watercourse_name"] == "Molse Nete" assert first["recorded_depth_m"] == 1.8 assert first["source_document_url"].startswith("https://vha.waterinfo.be/") assert first["vertical_reference"] == "document-specific" def test_bathymetry_acquisition_persists_reference_dataset_through_dataset_service(monkeypatch) -> None: project_id, area_id, dataset_id = uuid4(), uuid4(), uuid4() l_shape = MultiPolygon( [ Polygon( [ (5.0, 51.0), (6.0, 51.0), (6.0, 51.4), (5.4, 51.4), (5.4, 52.0), (5.0, 52.0), (5.0, 51.0), ] ) ] ) project = Project(id=project_id, name="Mol") area = Area( id=area_id, project_id=project_id, name="Gemeente Mol - officieel", geometry=from_shape(l_shape, srid=4326), ) db = FakeSession({(Project, project_id): project, (Area, area_id): area}) captured = {} def persist(_db, **kwargs): captured.update(kwargs) dataset = Dataset( id=dataset_id, project_id=project_id, area_id=area_id, name=kwargs["filename"], dataset_type="vector", source=kwargs["source"], dataset_role=kwargs["dataset_role"], source_name=kwargs["source_name"], reference_layer_name=kwargs["reference_layer_name"], source_metadata=kwargs["source_metadata"], provenance_metadata=kwargs["provenance_metadata"], status="ready", ) db.rows[(Dataset, dataset_id)] = dataset return SimpleNamespace(id=dataset_id) monkeypatch.setattr(DatasetService, "import_vector_bytes", persist) result = BathymetryProfileAcquisitionService.acquire( db, project_id, request(area_id=area_id), settings=Settings(_env_file=None), opener=provider_opener(), ) assert result["output_dataset_id"] == str(dataset_id) assert result["profile_count"] == 2 assert captured["dataset_role"] == "reference" assert captured["source_name"] == BathymetryProfileAcquisitionService.PROVIDER assert captured["reference_layer_name"] == "bathymetry_profiles" assert captured["source_metadata"]["theme"] == "bathymetry" assert captured["source_metadata"]["volume_supported"] is False assert captured["source_metadata"]["municipality"] == "Mol" assert captured["source_metadata"]["regional_partitions_complete"] is False payload = json.loads(captured["content"]) assert payload["features"][0]["properties"]["municipality"] == "Mol" assert captured["provenance_metadata"]["water_volume_available"] is False assert len(payload["features"]) == 2 def test_bathymetry_acquisition_rejects_unbounded_feature_volume() -> None: settings = Settings(_env_file=None, BATHYMETRY_PROFILES_MAX_FEATURES=2) with pytest.raises(AppError) as exc_info: BathymetryProfileAcquisitionService._fetch_profiles( (5.0, 51.0, 6.0, 52.0), settings, provider_opener(count=3), ) assert exc_info.value.code == "BATHYMETRY_SCOPE_TOO_LARGE" assert exc_info.value.details["candidate_count"] == 3 def test_bathymetry_source_api_uses_canonical_envelope() -> None: project_id = uuid4() db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}) app.dependency_overrides[get_db] = lambda: db try: response = TestClient(app).get(f"/api/v1/projects/{project_id}/datasets/bathymetry/sources") finally: app.dependency_overrides.clear() assert response.status_code == 200 assert set(response.json()) == {"data"} assert response.json()["data"]["total"] == 4 assert response.json()["data"]["items"][0]["integration_status"] == "operational" def test_bathymetry_acquisition_route_stays_inside_existing_job_envelope(monkeypatch) -> None: project_id, dataset_id = uuid4(), uuid4() db = FakeSession({(Project, project_id): Project(id=project_id, name="Mol")}) monkeypatch.setattr( BathymetryProfileAcquisitionService, "acquire", lambda *_args, **_kwargs: { "output_dataset_id": str(dataset_id), "provider": BathymetryProfileAcquisitionService.PROVIDER, "profile_count": 2, }, ) app.dependency_overrides[get_db] = lambda: db try: response = TestClient(app).post( f"/api/v1/projects/{project_id}/datasets/bathymetry/profiles/acquire", json=request().model_dump(mode="json"), ) finally: app.dependency_overrides.clear() assert response.status_code == 200 assert set(response.json()) == {"data"} assert response.json()["data"]["job_type"] == "vector.bathymetry_profiles.acquire" assert response.json()["data"]["output_dataset_id"] == str(dataset_id) assert any(isinstance(item, Job) for item in db.added) def test_bathymetry_contract_and_expansion_roadmap_are_documented() -> None: api_contracts = (ROOT / "docs" / "API_CONTRACTS.md").read_text(encoding="utf-8") roadmap = ROOT / "docs" / "BATHYMETRY_EXPANSION_ROADMAP.md" assert "bathymetry/profiles/acquire" in api_contracts assert roadmap.exists() contents = roadmap.read_text(encoding="utf-8") assert "LAT" in contents and "mDNG" in contents and "territoriale zee" in contents readiness = (ROOT / "scripts" / "run_readiness_check.sh").read_text(encoding="utf-8") dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8") assert "py_compile scripts/provision_mol_bathymetry_profiles.py" in readiness assert "COPY scripts/provision_mol_bathymetry_profiles.py" in dockerfile operator = (ROOT / "scripts" / "provision_mol_bathymetry_profiles.py").read_text(encoding="utf-8") map_workspace = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8") assert 'DEFAULT_PROJECT_NAME = "Kempen Regional Workbench"' in operator assert "regional_partitions_complete" in map_workspace assert "historische profielen" in map_workspace