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>
271 lines
10 KiB
Python
271 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.core.errors import AppError
|
|
from app.models import Dataset
|
|
from app.services.grb_refresh_plan_service import GrbRefreshPlanService
|
|
from tests.frontend_contract import read_map_workspace
|
|
|
|
|
|
NOW = datetime(2026, 7, 16, 16, 0, tzinfo=timezone.utc)
|
|
PROJECT_ID = uuid.uuid4()
|
|
|
|
|
|
class _Query:
|
|
def __init__(self, rows: list[Dataset]) -> None:
|
|
self.rows = rows
|
|
|
|
def filter(self, *_args):
|
|
return self
|
|
|
|
def all(self) -> list[Dataset]:
|
|
return self.rows
|
|
|
|
|
|
class _Db:
|
|
def __init__(self, rows: list[Dataset]) -> None:
|
|
self.rows = rows
|
|
|
|
def get(self, _model, identifier):
|
|
return SimpleNamespace(id=identifier)
|
|
|
|
def query(self, _model):
|
|
return _Query(self.rows)
|
|
|
|
|
|
def _dataset(theme: str, version: str = "2026-07-14", count: int = 100) -> Dataset:
|
|
return Dataset(
|
|
id=uuid.uuid4(),
|
|
project_id=PROJECT_ID,
|
|
name=f"grb_{theme}.geojson",
|
|
dataset_type="vector",
|
|
source="operator_official_import",
|
|
source_name="grb",
|
|
dataset_role="reference",
|
|
reference_layer_name=theme,
|
|
source_version=version,
|
|
temporal_series_key=f"grb:{theme}:kempen-transport-region",
|
|
observed_at=datetime.fromisoformat(f"{version}T00:00:00+00:00"),
|
|
imported_at=NOW,
|
|
metadata_json={"feature_count": count},
|
|
size_bytes=1000,
|
|
status="ready",
|
|
)
|
|
|
|
|
|
def _catalog(status: str = "available", version: str | None = "Toestand 2026-07-15"):
|
|
return SimpleNamespace(
|
|
items=[
|
|
SimpleNamespace(
|
|
source_name="grb",
|
|
status=status,
|
|
reachable=status == "available",
|
|
remote_version=version,
|
|
checked_at=NOW,
|
|
)
|
|
]
|
|
)
|
|
|
|
|
|
def test_refresh_plan_marks_all_older_immutable_snapshots_as_update_available(monkeypatch) -> None:
|
|
rows = [_dataset(theme, count=(index + 1) * 100) for index, theme in enumerate(("buildings", "roads", "water", "parcels"))]
|
|
monkeypatch.setattr(
|
|
"app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
|
|
lambda *_args, **_kwargs: _catalog(),
|
|
)
|
|
|
|
plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW)
|
|
|
|
assert plan.remote_edition_date.isoformat() == "2026-07-15"
|
|
assert plan.summary.update_available_count == 4
|
|
assert plan.summary.new_dataset_count_if_applied == 4
|
|
assert plan.summary.retained_dataset_count == 4
|
|
assert plan.summary.current_feature_count == 1000
|
|
assert plan.summary.current_size_bytes == 4000
|
|
assert all(item.status == "update_available" for item in plan.layers)
|
|
assert all(item.retained_after_refresh for item in plan.layers)
|
|
assert plan.automatic_import is False
|
|
assert plan.destructive_replacement is False
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("rows", "catalog", "expected"),
|
|
[
|
|
([_dataset("buildings", "2026-07-15")], _catalog(), "current"),
|
|
([], _catalog(), "not_loaded"),
|
|
([_dataset("buildings")], _catalog("unavailable"), "remote_unavailable"),
|
|
([_dataset("buildings")], _catalog("available", "Onbekende toestand"), "review_required"),
|
|
],
|
|
)
|
|
def test_refresh_plan_status_matrix(monkeypatch, rows, catalog, expected) -> None:
|
|
monkeypatch.setattr(
|
|
"app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
|
|
lambda *_args, **_kwargs: catalog,
|
|
)
|
|
plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW)
|
|
buildings = next(item for item in plan.layers if item.theme == "buildings")
|
|
assert buildings.status == expected
|
|
|
|
|
|
def test_refresh_plan_rejects_an_unapproved_scope(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
"app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
|
|
lambda *_args, **_kwargs: _catalog(),
|
|
)
|
|
with pytest.raises(AppError) as raised:
|
|
GrbRefreshPlanService.build(_Db([]), PROJECT_ID, scope="mol", now=NOW)
|
|
assert raised.value.code == "GRB_REFRESH_SCOPE_UNSUPPORTED"
|
|
|
|
|
|
def test_refresh_plan_route_uses_the_canonical_envelope(monkeypatch) -> None:
|
|
from app.api.routes import datasets as dataset_routes
|
|
|
|
monkeypatch.setattr(
|
|
"app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
|
|
lambda *_args, **_kwargs: _catalog(),
|
|
)
|
|
response = dataset_routes.plan_grb_dataset_refresh(
|
|
project_id=PROJECT_ID,
|
|
scope="kempen-transport-region",
|
|
refresh_catalog=False,
|
|
db=_Db([]),
|
|
)
|
|
assert list(response) == ["data"]
|
|
assert response["data"]["project_id"] == PROJECT_ID
|
|
assert response["data"]["automatic_import"] is False
|
|
|
|
|
|
def _load_operator_module():
|
|
root = Path(__file__).resolve().parents[2]
|
|
scripts = root / "scripts"
|
|
if str(scripts) not in sys.path:
|
|
sys.path.insert(0, str(scripts))
|
|
spec = importlib.util.spec_from_file_location("manage_grb_refresh_s223", scripts / "manage_grb_refresh.py")
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_operator_requires_exact_edition_and_plan_hash() -> None:
|
|
module = _load_operator_module()
|
|
assert module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-15") == "2026-07-15"
|
|
with pytest.raises(RuntimeError, match="confirm-edition"):
|
|
module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-14")
|
|
|
|
payload = {"status": "staged", "layers": []}
|
|
first = module.canonical_plan_sha256(payload)
|
|
assert first == module.canonical_plan_sha256({**payload, "plan_sha256": first})
|
|
assert first != module.canonical_plan_sha256({"status": "staged", "layers": [{"theme": "roads"}]})
|
|
|
|
|
|
def test_operator_validates_every_staged_artifact_and_partition(tmp_path) -> None:
|
|
module = _load_operator_module()
|
|
manifest_dir = tmp_path / "kempen-transport-region" / "buildings" / "2026-07-15"
|
|
partition_dir = manifest_dir / "partitions"
|
|
partition_dir.mkdir(parents=True)
|
|
artifact = manifest_dir / "buildings.geojson"
|
|
partition = partition_dir / "13025_mol.geojson"
|
|
artifact.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
|
|
partition.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
|
|
manifest = {
|
|
"status": "complete",
|
|
"scope": "kempen-transport-region",
|
|
"theme": "buildings",
|
|
"observed_at": "2026-07-15",
|
|
"reference_truncated": False,
|
|
"member_count": 1,
|
|
"feature_count": 1,
|
|
"artifact_filename": artifact.name,
|
|
"artifact_sha256": module.sha256_file(artifact),
|
|
"artifact_size_bytes": artifact.stat().st_size,
|
|
"partitions": [{"filename": partition.name, "sha256": module.sha256_file(partition)}],
|
|
}
|
|
path = manifest_dir / "regional_buildings_manifest.json"
|
|
path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
|
assert module.validate_manifest(
|
|
path,
|
|
output_root=tmp_path,
|
|
scope="kempen-transport-region",
|
|
theme="buildings",
|
|
edition="2026-07-15",
|
|
)["feature_count"] == 1
|
|
|
|
partition.write_text("changed", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="partition checksum"):
|
|
module.validate_manifest(
|
|
path,
|
|
output_root=tmp_path,
|
|
scope="kempen-transport-region",
|
|
theme="buildings",
|
|
edition="2026-07-15",
|
|
)
|
|
|
|
|
|
def test_operator_builds_only_allowlisted_local_subprocess_commands(tmp_path) -> None:
|
|
module = _load_operator_module()
|
|
args = SimpleNamespace(
|
|
scope="kempen-transport-region",
|
|
api_url="http://127.0.0.1:8000/api/v1",
|
|
output_root=tmp_path,
|
|
request_timeout=180,
|
|
api_timeout=180,
|
|
batch_size=1000,
|
|
page_limit=1000,
|
|
max_features_per_member=100000,
|
|
max_total_features=1500000,
|
|
)
|
|
commands = module.build_operator_commands(
|
|
args,
|
|
["buildings", "roads", "water", "parcels"],
|
|
"2026-07-15",
|
|
fetch_only=True,
|
|
)
|
|
flattened = [argument for _label, command in commands for argument in command]
|
|
assert len(commands) == 2
|
|
assert "--fetch-only" in flattened
|
|
assert "provision_regional_grb_buildings.py" in " ".join(flattened)
|
|
assert "provision_regional_grb_context.py" in " ".join(flattened)
|
|
assert "--force" not in flattened
|
|
|
|
|
|
def test_refresh_api_and_frontend_remain_explicit_only() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
service = (root / "backend" / "app" / "services" / "grb_refresh_plan_service.py").read_text(encoding="utf-8")
|
|
hook = (root / "frontend" / "src" / "hooks" / "useSourceFreshness.ts").read_text(encoding="utf-8")
|
|
operator = (root / "scripts" / "manage_grb_refresh.py").read_text(encoding="utf-8")
|
|
assert "DatasetService" not in service
|
|
assert "VectorFeatureService" not in service
|
|
assert "void probeCatalogs(" not in hook
|
|
assert 'choices=("plan", "stage", "apply")' in operator
|
|
assert "--confirm-plan-sha256" in operator
|
|
|
|
|
|
def test_map_theme_ranking_prefers_newer_observation_over_feature_count() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
workspace = read_map_workspace()
|
|
observed_sort = workspace.index("const observedAtDifference")
|
|
feature_tiebreaker = workspace.index("right.feature_count", observed_sort)
|
|
assert observed_sort < feature_tiebreaker
|
|
assert "new Date(right.observed_at ?? 0).getTime()" in workspace
|
|
assert "if (observedAtDifference !== 0) return observedAtDifference" in workspace
|
|
|
|
|
|
def test_map_workspace_restores_theme_from_selected_dataset() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
workspace = read_map_workspace()
|
|
assert "function themeIdForDataset(" in workspace
|
|
assert "useState<DataThemeId>(() =>" in workspace
|
|
assert "return themeIdForDataset(selectedDataset) ?? 'buildings'" in workspace
|
|
assert "const matchingThemeId = themeIdForDataset(selectedMapDataset)" in workspace
|