364 lines
13 KiB
Python
364 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
|
|
|
|
|
|
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
|