JensandClaude Opus 5
6572e4ad5f
93 test files read a single frontend source and asserted identifiers in it. The
MapWorkspace split showed what that costs: 24 tests went red for a move that
changed no behaviour at all. A contract belongs to the feature — a container,
its hooks, its domain layer — not to whichever file currently holds it.
232 read sites now resolve through read_feature(). The distinction that makes
this safe is direction: a *positive* contract ("this is wired") may widen,
because the identifier must still exist somewhere in the feature; a *negative*
one ("this component performs no transport") is a statement about one file, and
widening it would quietly weaken the check. The 73 single-file reads that
remain are exactly those, and a guard now enforces the rule for new tests.
Verified rather than assumed: of the 732 migrated positive assertions, 644 still
match exactly one module — as specific as before — and the other 86 already
spanned a container and its hook by nature. Two apparent misses are an artefact
of the checking regex reading an escaped newline literally.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
489 lines
18 KiB
Python
489 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
from shapely.geometry import box
|
|
|
|
from app.core.errors import AppError
|
|
from app.main import app
|
|
from app.models import Area, Dataset, Project
|
|
from app.schemas.coverage import CoverageBBox
|
|
from app.services.coverage_registry_service import CoverageRegistryService, THEMES, ZONES
|
|
from app.services.vector_feature_service import VectorFeatureService
|
|
from tests.frontend_contract import read_map_workspace, read_feature
|
|
|
|
|
|
class FakeQuery:
|
|
def __init__(self, rows):
|
|
self.rows = rows
|
|
|
|
def filter(self, *_args, **_kwargs):
|
|
return self
|
|
|
|
def all(self):
|
|
return list(self.rows)
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, *, project, areas, datasets):
|
|
self.project = project
|
|
self.areas = areas
|
|
self.datasets = datasets
|
|
|
|
def get(self, model, object_id):
|
|
if model is Project and str(self.project.id) == str(object_id):
|
|
return self.project
|
|
return None
|
|
|
|
def query(self, model):
|
|
if model is Area:
|
|
return FakeQuery(self.areas)
|
|
if model is Dataset:
|
|
return FakeQuery(self.datasets)
|
|
raise AssertionError(f"Unexpected query model: {model}")
|
|
|
|
|
|
def scope_area(name: str, geometry):
|
|
return SimpleNamespace(name=name, geometry=geometry)
|
|
|
|
|
|
def governed_materialization(
|
|
*,
|
|
source_name: str,
|
|
reference_layer_name: str | None,
|
|
source_metadata: dict[str, object],
|
|
dataset_id=None,
|
|
) -> SimpleNamespace:
|
|
"""Build a complete authoritative materialization for coverage tests.
|
|
|
|
Coverage is a production-facing statement. These fixtures must therefore
|
|
carry the same registry, immutable snapshot, checksum and freshness state
|
|
that a materialized official dataset needs in production.
|
|
"""
|
|
|
|
source_registry_id = uuid4()
|
|
source_snapshot_id = uuid4()
|
|
checksum_sha256 = "a" * 64
|
|
return SimpleNamespace(
|
|
id=dataset_id or uuid4(),
|
|
status="ready",
|
|
source=source_name,
|
|
source_name=source_name,
|
|
reference_layer_name=reference_layer_name,
|
|
source_metadata=dict(source_metadata),
|
|
checksum_sha256=checksum_sha256,
|
|
source_registry_id=source_registry_id,
|
|
source_snapshot_id=source_snapshot_id,
|
|
data_contract_key="geointel.vector.geojson",
|
|
data_contract_version="1.0.0",
|
|
validation_status="passed",
|
|
provenance_status="complete",
|
|
lineage_status="not_applicable",
|
|
quarantine_status="not_quarantined",
|
|
source_registry=SimpleNamespace(
|
|
source_key=source_name,
|
|
classification="authoritative",
|
|
authority_scope_json={"scope": "coverage test"},
|
|
usage_policy_json={},
|
|
),
|
|
source_snapshot=SimpleNamespace(
|
|
source_registry_id=source_registry_id,
|
|
checksum_sha256=checksum_sha256,
|
|
freshness_status="current",
|
|
ingest_status="ingested",
|
|
),
|
|
)
|
|
|
|
|
|
def test_coverage_catalog_uses_normalized_contracts_and_does_not_change_provider_registry() -> None:
|
|
catalog = CoverageRegistryService.catalog()
|
|
|
|
assert set(catalog.themes) == set(THEMES)
|
|
assert set(catalog.zones) == set(ZONES)
|
|
assert catalog.statuses == ["unsupported", "not_configured", "partial", "operational"]
|
|
assert {source.source_name for source in catalog.sources} >= {
|
|
"ngi_adminvector",
|
|
"statbel",
|
|
"digitaal_vlaanderen",
|
|
"spw_geoportail",
|
|
"urbis",
|
|
"rbins_marine_reporting_units",
|
|
"rbins_msp_2026",
|
|
"mdk_bathymetry",
|
|
"vmm_vha_bathymetry_profiles",
|
|
}
|
|
assert next(source for source in catalog.sources if source.source_name == "ngi_adminvector").license_note == "CC BY 4.0"
|
|
assert next(source for source in catalog.sources if source.source_name == "mdk_bathymetry").integration_status == "not_configured"
|
|
assert next(
|
|
source for source in catalog.sources if source.source_name == "vmm_vha_bathymetry_profiles"
|
|
).integration_status == "operational"
|
|
|
|
response = TestClient(app).get("/api/v1/external/coverage/catalog")
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["themes"] == list(THEMES)
|
|
|
|
|
|
def test_national_and_maritime_reference_layers_are_selection_analyzable() -> None:
|
|
cases = (
|
|
("ngi_adminvector", "belgium_municipalities", "administrative"),
|
|
("rbins_marine_reporting_units", "marine_legal_scopes", "marine_environment"),
|
|
("rbins_msp_2026", "marine_spatial_plan_2026", "maritime_planning"),
|
|
)
|
|
for source_name, layer_name, expected_theme in cases:
|
|
dataset = Dataset(
|
|
id=uuid4(),
|
|
project_id=uuid4(),
|
|
name=f"{layer_name}.geojson",
|
|
dataset_type="vector",
|
|
source="operator_official_import",
|
|
source_name=source_name,
|
|
reference_layer_name=layer_name,
|
|
source_metadata={"authority_level": "authoritative"},
|
|
status="ready",
|
|
)
|
|
|
|
assert VectorFeatureService._dataset_theme(dataset) == expected_theme
|
|
assert VectorFeatureService.supports_selection_summary(dataset) is True
|
|
|
|
|
|
def test_national_scope_operator_assigns_explicit_map_themes() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
operator = (root / "scripts" / "provision_belgium_north_sea_scope.py").read_text(encoding="utf-8")
|
|
map_workspace = read_map_workspace()
|
|
|
|
assert '"belgium_municipalities": "administrative"' in operator
|
|
assert '"marine_legal_scopes": "marine_environment"' in operator
|
|
assert '"marine_spatial_plan_2026": "maritime_planning"' in operator
|
|
assert "id: 'administrative'" in map_workspace
|
|
assert "id: 'maritime_planning'" in map_workspace
|
|
assert "id: 'marine_environment'" in map_workspace
|
|
|
|
|
|
def test_coverage_resolver_only_reports_operational_for_materialized_ready_dataset() -> None:
|
|
project_id = uuid4()
|
|
project = SimpleNamespace(id=project_id)
|
|
areas = [
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Flanders", box(2.5, 50.7, 5.9, 51.5)),
|
|
]
|
|
bbox = CoverageBBox(minx=4.9, miny=50.9, maxx=5.0, maxy=51.0)
|
|
|
|
without_materialized = CoverageRegistryService.resolve(
|
|
FakeSession(project=project, areas=areas, datasets=[]),
|
|
project_id,
|
|
bbox,
|
|
["admin"],
|
|
)
|
|
assert without_materialized.intersected_zones == ["flanders"]
|
|
assert without_materialized.items[0].status == "partial"
|
|
assert without_materialized.items[0].materialized_dataset_ids == []
|
|
|
|
dataset_id = uuid4()
|
|
materialized = governed_materialization(
|
|
dataset_id=dataset_id,
|
|
source_name="ngi_adminvector",
|
|
reference_layer_name="belgium_regions",
|
|
source_metadata={"coverage_zones": ["belgium", "flanders", "wallonia", "brussels"]},
|
|
)
|
|
with_materialized = CoverageRegistryService.resolve(
|
|
FakeSession(project=project, areas=areas, datasets=[materialized]),
|
|
project_id,
|
|
bbox,
|
|
["admin"],
|
|
)
|
|
admin_item = next(item for item in with_materialized.items if item.zone == "flanders")
|
|
assert admin_item.status == "operational"
|
|
assert admin_item.materialized_dataset_ids == [dataset_id]
|
|
|
|
|
|
def test_statbel_population_materialization_does_not_masquerade_as_admin_data() -> None:
|
|
project_id = uuid4()
|
|
statbel_id = uuid4()
|
|
statbel = governed_materialization(
|
|
dataset_id=statbel_id,
|
|
source_name="statbel",
|
|
reference_layer_name="population",
|
|
source_metadata={"coverage_zones": ["belgium", "flanders", "wallonia", "brussels"]},
|
|
)
|
|
result = CoverageRegistryService.resolve(
|
|
FakeSession(
|
|
project=SimpleNamespace(id=project_id),
|
|
areas=[
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Flanders", box(2.5, 50.7, 5.9, 51.5)),
|
|
],
|
|
datasets=[statbel],
|
|
),
|
|
project_id,
|
|
CoverageBBox(minx=4.9, miny=50.9, maxx=5.0, maxy=51.0),
|
|
["admin", "population"],
|
|
)
|
|
|
|
admin = next(item for item in result.items if item.theme == "admin")
|
|
population = next(item for item in result.items if item.theme == "population")
|
|
assert admin.materialized_dataset_ids == []
|
|
assert population.status == "operational"
|
|
assert population.materialized_dataset_ids == [statbel_id]
|
|
|
|
|
|
def test_bounded_api_materialization_only_covers_its_persisted_bbox() -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
dataset = governed_materialization(
|
|
dataset_id=dataset_id,
|
|
source_name="spw_picc",
|
|
reference_layer_name="buildings",
|
|
source_metadata={
|
|
"coverage_zones": ["wallonia"],
|
|
"bbox_epsg4326": [4.55, 50.58, 4.56, 50.59],
|
|
},
|
|
)
|
|
session = FakeSession(
|
|
project=SimpleNamespace(id=project_id),
|
|
areas=[
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Wallonia", box(2.5, 49.5, 6.4, 50.8)),
|
|
],
|
|
datasets=[dataset],
|
|
)
|
|
|
|
inside = CoverageRegistryService.resolve(
|
|
session,
|
|
project_id,
|
|
CoverageBBox(minx=4.551, miny=50.581, maxx=4.559, maxy=50.589),
|
|
["buildings"],
|
|
)
|
|
outside = CoverageRegistryService.resolve(
|
|
session,
|
|
project_id,
|
|
CoverageBBox(minx=4.7, miny=50.6, maxx=4.71, maxy=50.61),
|
|
["buildings"],
|
|
)
|
|
|
|
assert inside.items[0].status == "operational"
|
|
assert inside.items[0].materialized_dataset_ids == [dataset_id]
|
|
assert outside.items[0].status == "partial"
|
|
assert outside.items[0].materialized_dataset_ids == []
|
|
|
|
|
|
def test_bounded_partition_union_can_be_operational() -> None:
|
|
project_id = uuid4()
|
|
left_id = uuid4()
|
|
right_id = uuid4()
|
|
datasets = [
|
|
governed_materialization(
|
|
dataset_id=left_id,
|
|
source_name="spw_picc",
|
|
reference_layer_name="buildings",
|
|
source_metadata={
|
|
"coverage_zones": ["wallonia"],
|
|
"bbox_epsg4326": [4.50, 50.50, 4.60, 50.60],
|
|
},
|
|
),
|
|
governed_materialization(
|
|
dataset_id=right_id,
|
|
source_name="spw_picc",
|
|
reference_layer_name="buildings",
|
|
source_metadata={
|
|
"coverage_zones": ["wallonia"],
|
|
"bbox_epsg4326": [4.60, 50.50, 4.70, 50.60],
|
|
},
|
|
),
|
|
]
|
|
session = FakeSession(
|
|
project=SimpleNamespace(id=project_id),
|
|
areas=[scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)), scope_area("Wallonia", box(2.5, 49.5, 6.4, 50.8))],
|
|
datasets=datasets,
|
|
)
|
|
|
|
result = CoverageRegistryService.resolve(session, project_id, CoverageBBox(minx=4.51, miny=50.51, maxx=4.69, maxy=50.59), ["buildings"])
|
|
|
|
assert result.items[0].status == "operational"
|
|
assert result.items[0].materialized_dataset_ids == [left_id, right_id]
|
|
|
|
|
|
def test_spw_bathymetry_materialization_is_source_specific() -> None:
|
|
project_id = uuid4()
|
|
scope = [
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Wallonia", box(2.5, 49.5, 6.4, 50.8)),
|
|
]
|
|
selection = CoverageBBox(minx=4.851, miny=50.451, maxx=4.869, maxy=50.469)
|
|
spw_picc = governed_materialization(
|
|
source_name="spw_picc",
|
|
reference_layer_name="buildings",
|
|
source_metadata={
|
|
"coverage_zones": ["wallonia"],
|
|
"bbox_epsg4326": [4.85, 50.45, 4.87, 50.47],
|
|
},
|
|
)
|
|
without_bathymetry = CoverageRegistryService.resolve(
|
|
FakeSession(project=SimpleNamespace(id=project_id), areas=scope, datasets=[spw_picc]),
|
|
project_id,
|
|
selection,
|
|
["bathymetry"],
|
|
)
|
|
assert without_bathymetry.items[0].status == "partial"
|
|
assert without_bathymetry.items[0].materialized_dataset_ids == []
|
|
|
|
bathymetry_id = uuid4()
|
|
bathymetry = governed_materialization(
|
|
dataset_id=bathymetry_id,
|
|
source_name="spw_bathymetry",
|
|
reference_layer_name=None,
|
|
source_metadata={
|
|
"coverage_zones": ["wallonia"],
|
|
"bbox_epsg4326": [4.85, 50.45, 4.87, 50.47],
|
|
},
|
|
)
|
|
with_bathymetry = CoverageRegistryService.resolve(
|
|
FakeSession(project=SimpleNamespace(id=project_id), areas=scope, datasets=[spw_picc, bathymetry]),
|
|
project_id,
|
|
selection,
|
|
["bathymetry"],
|
|
)
|
|
assert with_bathymetry.items[0].status == "operational"
|
|
assert with_bathymetry.items[0].materialized_dataset_ids == [bathymetry_id]
|
|
|
|
|
|
def test_vha_bathymetry_profiles_are_operational_only_inside_the_persisted_selection() -> None:
|
|
project_id = uuid4()
|
|
scope = [
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Flanders", box(2.5, 50.7, 5.9, 51.5)),
|
|
]
|
|
selection = CoverageBBox(minx=5.101, miny=51.171, maxx=5.109, maxy=51.179)
|
|
without_profiles = CoverageRegistryService.resolve(
|
|
FakeSession(project=SimpleNamespace(id=project_id), areas=scope, datasets=[]),
|
|
project_id,
|
|
selection,
|
|
["bathymetry"],
|
|
)
|
|
assert without_profiles.items[0].status == "partial"
|
|
assert without_profiles.items[0].source_names == ["vmm_vha_bathymetry_profiles"]
|
|
|
|
profile_id = uuid4()
|
|
profiles = governed_materialization(
|
|
dataset_id=profile_id,
|
|
source_name="vmm_vha_bathymetry_profiles",
|
|
reference_layer_name="bathymetry_profile_points",
|
|
source_metadata={
|
|
"coverage_zones": ["flanders"],
|
|
"bbox_epsg4326": [5.1, 51.17, 5.11, 51.18],
|
|
},
|
|
)
|
|
with_profiles = CoverageRegistryService.resolve(
|
|
FakeSession(project=SimpleNamespace(id=project_id), areas=scope, datasets=[profiles]),
|
|
project_id,
|
|
selection,
|
|
["bathymetry"],
|
|
)
|
|
assert with_profiles.items[0].status == "operational"
|
|
assert with_profiles.items[0].materialized_dataset_ids == [profile_id]
|
|
|
|
|
|
def test_mixed_land_and_north_sea_selection_remains_split() -> None:
|
|
project_id = uuid4()
|
|
project = SimpleNamespace(id=project_id)
|
|
areas = [
|
|
scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5)),
|
|
scope_area("Flanders", box(2.5, 50.7, 5.9, 51.5)),
|
|
scope_area("Belgian part of the North Sea", box(2.2, 51.1, 3.4, 51.9)),
|
|
scope_area("Belgian territorial sea (0-12 nautical miles)", box(2.7, 51.1, 3.4, 51.5)),
|
|
scope_area("Belgian exclusive economic zone beyond territorial sea", box(2.2, 51.5, 3.1, 51.9)),
|
|
scope_area("Belgian continental shelf beyond territorial sea", box(2.2, 51.5, 3.1, 51.9)),
|
|
]
|
|
|
|
result = CoverageRegistryService.resolve(
|
|
FakeSession(project=project, areas=areas, datasets=[]),
|
|
project_id,
|
|
CoverageBBox(minx=2.65, miny=51.05, maxx=2.85, maxy=51.2),
|
|
["admin", "bathymetry"],
|
|
)
|
|
|
|
assert result.intersected_zones == ["flanders", "territorial_sea"]
|
|
assert len(result.items) == 4
|
|
assert any("crosses coverage zones" in warning for warning in result.warnings)
|
|
bathymetry = next(item for item in result.items if item.zone == "territorial_sea" and item.theme == "bathymetry")
|
|
assert bathymetry.status == "not_configured"
|
|
assert bathymetry.materialized_dataset_ids == []
|
|
|
|
|
|
def test_flemish_materialization_is_theme_specific() -> None:
|
|
project_id = uuid4()
|
|
project = SimpleNamespace(id=project_id)
|
|
orthophoto_id = uuid4()
|
|
orthophoto = governed_materialization(
|
|
dataset_id=orthophoto_id,
|
|
source_name="digitaal_vlaanderen_orthophoto",
|
|
reference_layer_name="orthophoto",
|
|
source_metadata={"coverage_zones": ["flanders"]},
|
|
)
|
|
result = CoverageRegistryService.resolve(
|
|
FakeSession(
|
|
project=project,
|
|
areas=[scope_area("Flanders", box(2.5, 50.7, 5.9, 51.5))],
|
|
datasets=[orthophoto],
|
|
),
|
|
project_id,
|
|
CoverageBBox(minx=4.9, miny=50.9, maxx=5.0, maxy=51.0),
|
|
["orthophoto", "roads"],
|
|
)
|
|
|
|
assert next(item for item in result.items if item.theme == "orthophoto").status == "operational"
|
|
assert next(item for item in result.items if item.theme == "roads").status == "partial"
|
|
|
|
|
|
def test_outside_scope_and_unknown_theme_are_explicit() -> None:
|
|
project_id = uuid4()
|
|
project = SimpleNamespace(id=project_id)
|
|
db = FakeSession(
|
|
project=project,
|
|
areas=[scope_area("Belgium land", box(2.5, 49.5, 6.4, 51.5))],
|
|
datasets=[],
|
|
)
|
|
result = CoverageRegistryService.resolve(
|
|
db,
|
|
project_id,
|
|
CoverageBBox(minx=7.0, miny=52.0, maxx=7.1, maxy=52.1),
|
|
["admin"],
|
|
)
|
|
assert result.intersected_zones == []
|
|
assert result.outside_supported_scope is True
|
|
assert result.items == []
|
|
|
|
try:
|
|
CoverageRegistryService.resolve(
|
|
db,
|
|
project_id,
|
|
CoverageBBox(minx=4.0, miny=50.0, maxx=4.1, maxy=50.1),
|
|
["invented_metric"],
|
|
)
|
|
except AppError as exc:
|
|
assert exc.code == "COVERAGE_THEME_UNSUPPORTED"
|
|
assert exc.status_code == 422
|
|
assert exc.details["unsupported_themes"] == ["invented_metric"]
|
|
else:
|
|
raise AssertionError("Unknown coverage theme was accepted")
|
|
|
|
|
|
def test_frontend_prefers_materialized_national_workspace_and_resolves_drawn_bbox() -> None:
|
|
root = Path(__file__).parents[2]
|
|
focus = (root / "frontend" / "src" / "config" / "primaryFocus.ts").read_text(encoding="utf-8")
|
|
workspace_hook = read_feature("shell")
|
|
coverage_hook = read_feature("map_workspace")
|
|
map_workspace = read_map_workspace()
|
|
|
|
assert "Belgium and North Sea Workbench" in focus
|
|
assert "nationalProject" in workspace_hook
|
|
assert "return nationalProject.id" in workspace_hook
|
|
assert "NATIONAL_WORKSPACE_REGION" in workspace_hook
|
|
assert "externalApi.resolveCoverage" in coverage_hook
|
|
assert "coverage.outside_supported_scope" in map_workspace
|
|
assert "coverageStatusLabel" in map_workspace
|
|
assert "activeThemeSupportsCurrentSelection" in map_workspace
|
|
assert "activeThemeAvailable && !regionalPartitionedThemeActive" in map_workspace
|