JensandClaude Opus 5
39c12822bc
MapWorkspace.tsx opened with ~590 lines of theme catalogue, dataset matching and label formatting above a 3.200-line component. None of it is React, all of it is independently testable, and both render paths read from it, so it belongs beside the pure helpers that already live in mapWorkspaceUtils. The contract tests that read MapWorkspace.tsx would have gone red for a move that changes no behaviour at all — 24 of them. That is the brittleness the frontend_contract helper exists to remove, so it gains read_map_workspace(): the workspace is one feature spread over several modules, and a contract belongs to the feature rather than to whichever file currently holds it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
368 lines
13 KiB
Python
368 lines
13 KiB
Python
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
|
|
from tests.frontend_contract import read_map_workspace
|
|
|
|
|
|
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"
|
|
# Bounded MDK acquisition now exists but stays fail-closed until the
|
|
# operator enables it explicitly with a live-validated coverage id.
|
|
assert by_key["mdk_bcp_bathymetry"]["acquisition_supported"] is True
|
|
assert by_key["mdk_bcp_bathymetry"]["configured"] 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 = read_map_workspace()
|
|
assert 'DEFAULT_PROJECT_NAME = "Kempen Regional Workbench"' in operator
|
|
assert "regional_partitions_complete" in map_workspace
|
|
assert "historische profielen" in map_workspace
|