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>
491 lines
17 KiB
Python
491 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import ValidationError
|
|
|
|
from app.main import app
|
|
from app.models import Dataset, Export, SourceRegistry, SourceSnapshot
|
|
from app.schemas.export import ExportCreateResponse, MapResultExportRequest
|
|
from app.schemas.project import ProjectRead
|
|
from app.services.export_service import ExportService
|
|
from app.services.project_service import ProjectService
|
|
from app.services.storage_service import StorageService
|
|
from app.services.source_registry_service import SourceRegistryService
|
|
from app.services.temporal_analysis_service import TemporalAnalysisService
|
|
from app.services.thematic_raster_analysis_service import ThematicRasterAnalysisService
|
|
from tests.frontend_contract import assert_calls, assert_mentions, assert_wired, read_map_workspace
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, rows=None):
|
|
self.rows = rows or {}
|
|
self.added = []
|
|
|
|
def get(self, model, row_id):
|
|
return self.rows.get((model, row_id))
|
|
|
|
def add(self, row):
|
|
self.added.append(row)
|
|
|
|
def commit(self):
|
|
return None
|
|
|
|
def refresh(self, row):
|
|
return row
|
|
|
|
|
|
def bbox_payload() -> dict:
|
|
return {
|
|
"min_x": 5.10,
|
|
"min_y": 51.17,
|
|
"max_x": 5.11,
|
|
"max_y": 51.18,
|
|
"crs": "EPSG:4326",
|
|
}
|
|
|
|
|
|
def governed_dataset(
|
|
*,
|
|
project_id,
|
|
dataset_id,
|
|
name: str,
|
|
dataset_type: str,
|
|
source_key: str,
|
|
dataset_role: str = "source",
|
|
) -> Dataset:
|
|
"""Build an in-memory stand-in for a passed governed dataset.
|
|
|
|
Map-result export is an operational consumption boundary. These tests
|
|
must therefore model the same source registry/snapshot, checksum and
|
|
passed-contract evidence supplied by a real adapter rather than relying
|
|
on an old transient Dataset fixture.
|
|
"""
|
|
|
|
source = SourceRegistry(
|
|
id=uuid4(),
|
|
**SourceRegistryService.definition_for(source_key).as_model_values(),
|
|
)
|
|
checksum = sha256(f"{dataset_id}:{source_key}:{dataset_type}".encode("utf-8")).hexdigest()
|
|
snapshot = SourceSnapshot(
|
|
id=uuid4(),
|
|
source_registry_id=source.id,
|
|
snapshot_key=f"test:{source_key}:{checksum}",
|
|
checksum_sha256=checksum,
|
|
fetched_at=datetime.now(UTC),
|
|
crs="EPSG:31370",
|
|
units=source.default_units,
|
|
spatial_resolution_json={"x": 1.0, "y": 1.0, "unit": "m"},
|
|
temporal_coverage_json={"status": "test-fixture"},
|
|
geographic_coverage_json={"zone": "Flanders"},
|
|
observed_schema_json={"dataset_type": dataset_type},
|
|
freshness_status="current",
|
|
ingest_status="ingested",
|
|
known_limitations_json=["In-memory governed fixture used only by this export test."],
|
|
snapshot_metadata_json={"fixture_mode": True},
|
|
)
|
|
return Dataset(
|
|
id=dataset_id,
|
|
project_id=project_id,
|
|
name=name,
|
|
dataset_type=dataset_type,
|
|
source="governed test fixture",
|
|
dataset_role=dataset_role,
|
|
source_name=source.source_key,
|
|
source_registry_id=source.id,
|
|
source_snapshot_id=snapshot.id,
|
|
source_registry=source,
|
|
source_snapshot=snapshot,
|
|
data_contract_key=("geointel.raster.geotiff" if dataset_type == "raster" else "geointel.vector.geojson"),
|
|
data_contract_version="1.0.0",
|
|
validation_status="passed",
|
|
provenance_status="complete",
|
|
lineage_status="not_applicable",
|
|
quarantine_status="not_quarantined",
|
|
checksum_sha256=checksum,
|
|
metadata_json={"fixture_mode": True},
|
|
source_metadata={"fixture_mode": True, "source_registry_key": source.source_key},
|
|
provenance_metadata={"fixture_mode": True, "source_snapshot_id": str(snapshot.id)},
|
|
status="ready",
|
|
)
|
|
|
|
|
|
def test_map_result_export_request_requires_a_complete_target() -> None:
|
|
with pytest.raises(ValidationError):
|
|
MapResultExportRequest(project_id=uuid4(), mode="current", bbox=bbox_payload())
|
|
with pytest.raises(ValidationError):
|
|
MapResultExportRequest(project_id=uuid4(), mode="evolution", bbox=bbox_payload())
|
|
with pytest.raises(ValidationError):
|
|
MapResultExportRequest(
|
|
project_id=uuid4(),
|
|
mode="current",
|
|
dataset_id=uuid4(),
|
|
bbox=bbox_payload(),
|
|
partitioned=True,
|
|
)
|
|
|
|
|
|
def test_current_vector_map_result_uses_authoritative_selection_export(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
area_id = uuid4()
|
|
dataset = governed_dataset(
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
name="buildings.geojson",
|
|
dataset_type="vector",
|
|
source_key="grb",
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
expected = ExportCreateResponse(
|
|
export_id=uuid4(),
|
|
path="storage/exports/buildings-selection.geojson",
|
|
status="ready",
|
|
export_type="vector_selection_geojson",
|
|
)
|
|
captured: dict = {}
|
|
|
|
def fake_vector_export(*_args, **kwargs):
|
|
captured.update(kwargs)
|
|
return expected
|
|
|
|
monkeypatch.setattr(ExportService, "export_vector_selection_geojson", fake_vector_export)
|
|
response = ExportService.export_map_result(
|
|
db,
|
|
MapResultExportRequest(
|
|
project_id=project_id,
|
|
mode="current",
|
|
dataset_id=dataset_id,
|
|
area_id=area_id,
|
|
bbox=bbox_payload(),
|
|
theme_id="buildings",
|
|
),
|
|
)
|
|
|
|
assert response is expected
|
|
assert captured["area_id"] == area_id
|
|
assert captured["limit"] == 1000
|
|
|
|
|
|
def test_partitioned_vector_map_result_uses_governed_partition_export(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
dataset = governed_dataset(
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
name="vha-municipality.geojson",
|
|
dataset_type="vector",
|
|
source_key="vmm_vha_bathymetry_profiles",
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
expected = ExportCreateResponse(
|
|
export_id=uuid4(),
|
|
path="storage/exports/bathymetry-profile-selection.geojson",
|
|
status="ready",
|
|
export_type="partitioned_vector_selection_geojson",
|
|
)
|
|
captured: dict = {}
|
|
|
|
def fake_partition_export(*args, **kwargs):
|
|
captured["dataset"] = args[1]
|
|
captured.update(kwargs)
|
|
return expected
|
|
|
|
monkeypatch.setattr(
|
|
ExportService,
|
|
"export_partitioned_vector_selection_geojson",
|
|
fake_partition_export,
|
|
)
|
|
response = ExportService.export_map_result(
|
|
db,
|
|
MapResultExportRequest(
|
|
project_id=project_id,
|
|
mode="current",
|
|
dataset_id=dataset_id,
|
|
bbox=bbox_payload(),
|
|
partitioned=True,
|
|
partition_scope_key="flanders",
|
|
theme_id="bathymetry",
|
|
),
|
|
)
|
|
|
|
assert response is expected
|
|
assert captured["dataset"] is dataset
|
|
assert captured["partition_scope_key"] == "flanders"
|
|
assert captured["limit"] == 1000
|
|
|
|
|
|
def test_raster_map_result_is_recomputed_and_persisted(tmp_path, monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
dataset = governed_dataset(
|
|
project_id=project_id,
|
|
dataset_id=dataset_id,
|
|
name="space-occupation.tif",
|
|
dataset_type="raster",
|
|
source_key="department_omgeving_thematic_raster",
|
|
)
|
|
db = FakeSession({(Dataset, dataset_id): dataset})
|
|
export_path = tmp_path / "space-occupation-analysis.json"
|
|
captured: dict = {}
|
|
|
|
def fake_analyze(_db, captured_project_id, captured_dataset_id, payload):
|
|
captured.update(
|
|
project_id=captured_project_id,
|
|
dataset_id=captured_dataset_id,
|
|
area_id=payload.area_id,
|
|
)
|
|
return {
|
|
"selection_bbox": bbox_payload(),
|
|
"summary": {"metric_label": "Ruimtebeslag", "metric_value": 12.5, "metric_unit": "ha"},
|
|
}
|
|
|
|
monkeypatch.setattr(ThematicRasterAnalysisService, "analyze", fake_analyze)
|
|
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(export_path))
|
|
response = ExportService.export_map_result(
|
|
db,
|
|
MapResultExportRequest(
|
|
project_id=project_id,
|
|
mode="current",
|
|
dataset_id=dataset_id,
|
|
bbox=bbox_payload(),
|
|
theme_id="space_occupation",
|
|
),
|
|
)
|
|
|
|
persisted = [row for row in db.added if isinstance(row, Export)]
|
|
assert response.export_type == "map_analysis_json"
|
|
assert len(persisted) == 1
|
|
assert persisted[0].metadata_json["server_recomputed"] is True
|
|
assert persisted[0].metadata_json["theme_id"] == "space_occupation"
|
|
assert captured == {"project_id": project_id, "dataset_id": dataset_id, "area_id": None}
|
|
assert json.loads(export_path.read_text(encoding="utf-8"))["result"]["summary"]["metric_value"] == 12.5
|
|
|
|
|
|
def test_evolution_map_result_is_recomputed_and_persisted(tmp_path, monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
earlier_id = uuid4()
|
|
later_id = uuid4()
|
|
earlier_dataset = governed_dataset(
|
|
project_id=project_id,
|
|
dataset_id=earlier_id,
|
|
name="forest-earlier.geojson",
|
|
dataset_type="vector",
|
|
source_key="inbo_bwk_natura2000",
|
|
)
|
|
later_dataset = governed_dataset(
|
|
project_id=project_id,
|
|
dataset_id=later_id,
|
|
name="forest-later.geojson",
|
|
dataset_type="vector",
|
|
source_key="inbo_bwk_natura2000",
|
|
)
|
|
db = FakeSession(
|
|
{
|
|
(Dataset, earlier_id): earlier_dataset,
|
|
(Dataset, later_id): later_dataset,
|
|
}
|
|
)
|
|
export_path = tmp_path / "forest-evolution.json"
|
|
captured: dict = {}
|
|
|
|
class Comparison:
|
|
def model_dump(self, *, mode):
|
|
assert mode == "json"
|
|
return {"temporal_series_key": "forest", "metric": {"absolute_change": -2.0}}
|
|
|
|
def fake_compare(_db, *, project_id, payload):
|
|
captured.update(project_id=project_id, payload=payload)
|
|
return Comparison()
|
|
|
|
monkeypatch.setattr(TemporalAnalysisService, "compare", fake_compare)
|
|
monkeypatch.setattr(StorageService, "dataset_export_path", lambda *_args: str(export_path))
|
|
response = ExportService.export_map_result(
|
|
db,
|
|
MapResultExportRequest(
|
|
project_id=project_id,
|
|
mode="evolution",
|
|
earlier_dataset_id=earlier_id,
|
|
later_dataset_id=later_id,
|
|
bbox=bbox_payload(),
|
|
theme_id="forest",
|
|
),
|
|
)
|
|
|
|
assert response.export_type == "map_evolution_json"
|
|
assert captured["project_id"] == project_id
|
|
assert captured["payload"].earlier_dataset_id == earlier_id
|
|
assert json.loads(export_path.read_text(encoding="utf-8"))["metric"]["absolute_change"] == -2.0
|
|
|
|
|
|
def test_map_result_export_endpoint_uses_canonical_envelope(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
dataset_id = uuid4()
|
|
export_id = uuid4()
|
|
monkeypatch.setattr(
|
|
ExportService,
|
|
"export_map_result",
|
|
lambda *_args: ExportCreateResponse(
|
|
export_id=export_id,
|
|
path="storage/exports/map-analysis.json",
|
|
status="ready",
|
|
export_type="map_analysis_json",
|
|
),
|
|
)
|
|
|
|
response = TestClient(app).post(
|
|
"/api/v1/exports/map-result",
|
|
json={
|
|
"project_id": str(project_id),
|
|
"mode": "current",
|
|
"dataset_id": str(dataset_id),
|
|
"bbox": bbox_payload(),
|
|
"theme_id": "space_occupation",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"data": {
|
|
"export_id": str(export_id),
|
|
"path": "storage/exports/map-analysis.json",
|
|
"status": "ready",
|
|
"export_type": "map_analysis_json",
|
|
"metadata_json": None,
|
|
}
|
|
}
|
|
|
|
|
|
def test_frontend_persists_map_result_before_opening_downloads() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
workspace = read_map_workspace()
|
|
hook = (root / "frontend/src/hooks/useExportWorkflow.ts").read_text(encoding="utf-8")
|
|
api = (root / "frontend/src/services/api/exports.ts").read_text(encoding="utf-8")
|
|
|
|
assert "persistActiveResultAndOpenDownloads" in workspace
|
|
assert "onPersistMapResult(payload)" in workspace
|
|
assert "Bewaar in downloads" in workspace
|
|
assert "persistMapResult" in hook
|
|
assert "exportsApi.exportMapResult(payload)" in hook
|
|
assert "/api/v1/exports/map-result" in api
|
|
|
|
|
|
def test_project_list_supports_exact_canonical_workspace_lookup(monkeypatch) -> None:
|
|
project_id = uuid4()
|
|
captured: dict = {}
|
|
|
|
def fake_list(_db, *, limit, offset, name, project_status):
|
|
captured.update(limit=limit, offset=offset, name=name, project_status=project_status)
|
|
return [
|
|
ProjectRead(
|
|
id=project_id,
|
|
name="Kempen Regional Workbench",
|
|
region="Kempen",
|
|
status="active",
|
|
)
|
|
], 1
|
|
|
|
monkeypatch.setattr(ProjectService, "list_projects", fake_list)
|
|
response = TestClient(app).get(
|
|
"/api/v1/projects",
|
|
params={"name": "Kempen Regional Workbench", "limit": 1},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["items"][0]["id"] == str(project_id)
|
|
assert captured == {
|
|
"limit": 1,
|
|
"offset": 0,
|
|
"name": "Kempen Regional Workbench",
|
|
"project_status": "active",
|
|
}
|
|
|
|
|
|
def test_frontend_fetches_canonical_workspace_outside_default_project_page() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
workflow = (root / "frontend/src/hooks/useProjectWorkspace.ts").read_text(encoding="utf-8")
|
|
api = (root / "frontend/src/services/api/projects.ts").read_text(encoding="utf-8")
|
|
|
|
assert "projectsApi.list({ name: REGIONAL_WORKSPACE_PROJECT_NAME, limit: 1 })" in workflow
|
|
assert "[...canonicalResponse.items, ...response.items]" in workflow
|
|
assert "new URLSearchParams()" in api
|
|
|
|
|
|
def test_theme_failures_name_the_source_and_reason() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
hook = (root / "frontend/src/hooks/useMapThemeSelectionInsights.ts").read_text(encoding="utf-8")
|
|
|
|
assert "queries[index]?.dataset?.name" in hook
|
|
assert "queries[index]?.acquisition?.displayName" in hook
|
|
assert "reason: formatError(item.reason" in hook
|
|
assert "failure.dataset}: ${failure.reason}" in hook
|
|
|
|
|
|
def test_workspace_navigation_resets_the_actual_scroll_container() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
app = (root / "frontend/src/App.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "const workbenchMainRef = useRef<HTMLElement | null>(null)" in app
|
|
assert "workbenchMainRef.current?.scrollTo({ top: 0, left: 0 })" in app
|
|
assert "<main" in app
|
|
assert "ref={workbenchMainRef}" in app
|
|
|
|
|
|
def test_quality_scores_have_plain_language_interpretation() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
quality = (root / "frontend/src/components/quality/QualityResultsPanel.tsx").read_text(encoding="utf-8")
|
|
detection = (root / "frontend/src/components/detection/DetectionLab.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "Laatste score (0-1)" in quality
|
|
assert "Bruikbaar na controle" in quality
|
|
assert "Verkennend, controle vereist" in quality
|
|
assert "detectionQualityInterpretation" in detection
|
|
# The detection panel must translate an F1 into plain language.
|
|
assert_wired(detection, "detectionQualityInterpretation")
|
|
assert_mentions(detection, "kwaliteitsmeting")
|
|
|
|
|
|
def test_map_and_detection_workspaces_avoid_page_length_driven_layouts() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
styles = (root / "frontend/src/styles/app.css").read_text(encoding="utf-8")
|
|
premium = (root / "frontend/src/styles/premium.css").read_text(encoding="utf-8")
|
|
|
|
assert "height: clamp(34rem, calc(100dvh - 10rem), 58rem);" in styles
|
|
assert ".geo-theme-list" in styles and "overflow-y: auto;" in styles
|
|
assert "@media (max-width: 1500px)" in styles
|
|
assert ".workspace-grid-ai {\n grid-template-columns: minmax(0, 1fr);" in premium
|
|
assert "max-height: none;" in premium
|
|
|
|
|
|
def test_detection_lab_only_receives_operational_imagery_rasters() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
app_source = (root / "frontend/src/App.tsx").read_text(encoding="utf-8")
|
|
capability_source = (root / "frontend/src/lib/datasetCapabilities.ts").read_text(encoding="utf-8")
|
|
|
|
assert "department_omgeving_thematic_raster" in capability_source
|
|
assert "digitaal_vlaanderen_dhmv" in capability_source
|
|
assert "vmm_flood_hazard" in capability_source
|
|
assert "dataset.dataset_type !== 'raster' || dataset.status !== 'ready'" in capability_source
|
|
assert "const detectionRasterDatasets = useMemo(" in app_source
|
|
assert "rasterDatasets: detectionRasterDatasets" in app_source
|
|
assert "rasterDatasets={detectionRasterDatasets}" in app_source
|
|
assert "!isDetectionImageryDataset(selectedDataset)" in app_source
|
|
|
|
|
|
def test_download_workspace_surfaces_map_results_in_plain_dutch() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
exports = (root / "frontend/src/components/exports/ExportCenter.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "Gebiedsanalyse (JSON)" in exports
|
|
assert "Historische vergelijking (JSON)" in exports
|
|
assert "Kaartselectie (GeoJSON)" in exports
|
|
assert "Klaar om te delen" in exports
|
|
assert "Downloads vernieuwen" in exports
|
|
assert "JSON bekijken" in exports
|