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>
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
|
|
|
|
|
|
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 = (root / "frontend" / "src" / "hooks" / "useProjectWorkspace.ts").read_text(encoding="utf-8")
|
|
coverage_hook = (root / "frontend" / "src" / "hooks" / "useCoverageResolver.ts").read_text(encoding="utf-8")
|
|
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
|