270 lines
10 KiB
Python
270 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
|
|
|
|
|
|
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 = (root / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
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 = (root / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
assert "function themeIdForDataset(" in workspace
|
|
assert "useState<DataThemeId>(() =>" in workspace
|
|
assert "return themeIdForDataset(selectedDataset) ?? 'buildings'" in workspace
|
|
assert "const matchingThemeId = themeIdForDataset(selectedMapDataset)" in workspace
|