153 lines
6.4 KiB
Python
153 lines
6.4 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from shapely.geometry import Polygon, shape
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def load_module(name: str, filename: str):
|
|
scripts_path = str(SCRIPTS)
|
|
if scripts_path not in sys.path:
|
|
sys.path.insert(0, scripts_path)
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / filename)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_kempen_scope_matches_official_28_municipality_policy_region() -> None:
|
|
scopes = load_module("geographic_scopes_test", "geographic_scopes.py")
|
|
scope = scopes.KEMPEN_TRANSPORT_REGION_SCOPE
|
|
|
|
assert scope.key == "kempen-transport-region"
|
|
assert scope.project_name == "Kempen Regional Workbench"
|
|
assert scope.scope_type == "transport_region"
|
|
assert len(scope.members) == 28
|
|
assert len(set(scope.nis_codes)) == 28
|
|
assert ("Mol", "13025") in {(member.name, member.nis_code) for member in scope.members}
|
|
assert ("Nijlen", "12026") in {(member.name, member.nis_code) for member in scope.members}
|
|
assert "vervoerregio-kempen" in scope.authority_url
|
|
assert "geen claim" in scope.limitation_message
|
|
|
|
|
|
def test_scope_union_preserves_member_identity_and_policy_limitation() -> None:
|
|
scopes = load_module("geographic_scopes_union_test", "geographic_scopes.py")
|
|
provisioner = load_module("provision_geographic_scope_test", "provision_geographic_scope.py")
|
|
scope = scopes.GeographicScope(
|
|
key="test-region",
|
|
display_name="Testregio",
|
|
project_name="Test Regional Workbench",
|
|
project_region="Test",
|
|
area_name="Testregio - operationele grens",
|
|
authority_name="Test authority",
|
|
authority_url="https://example.test/scope",
|
|
scope_type="policy_region",
|
|
limitation_message="Operationele testgrens; geen landschappelijke claim.",
|
|
members=(scopes.ScopeMember("Alpha", "10001"), scopes.ScopeMember("Beta", "10002")),
|
|
)
|
|
source_features = [
|
|
{
|
|
"type": "Feature",
|
|
"id": "alpha",
|
|
"geometry": Polygon([(4.0, 51.0), (4.1, 51.0), (4.1, 51.1), (4.0, 51.1)]).__geo_interface__,
|
|
"properties": {"NAAM": "Alpha", "NISCODE": "10001"},
|
|
},
|
|
{
|
|
"type": "Feature",
|
|
"id": "beta",
|
|
"geometry": Polygon([(4.1, 51.0), (4.2, 51.0), (4.2, 51.1), (4.1, 51.1)]).__geo_interface__,
|
|
"properties": {"NAAM": "Beta", "NISCODE": "10002"},
|
|
},
|
|
]
|
|
|
|
boundary, members, summary = provisioner.build_scope_payloads(
|
|
scope,
|
|
source_features,
|
|
source_url="https://example.test/vrbg",
|
|
generated_at="2026-07-14T00:00:00+00:00",
|
|
)
|
|
|
|
assert len(boundary["features"]) == 1
|
|
assert len(members["features"]) == 2
|
|
assert shape(boundary["features"][0]["geometry"]).is_valid
|
|
assert boundary["features"][0]["properties"]["member_nis_codes"] == ["10001", "10002"]
|
|
assert boundary["features"][0]["properties"]["scope_limitation"] == scope.limitation_message
|
|
assert [feature["properties"]["municipality"] for feature in members["features"]] == ["Alpha", "Beta"]
|
|
assert summary["member_count"] == 2
|
|
assert summary["area_km2"] > 0
|
|
|
|
|
|
def test_scope_api_pagination_respects_canonical_limit() -> None:
|
|
provisioner = load_module("provision_geographic_scope_paging_test", "provision_geographic_scope.py")
|
|
|
|
class Response:
|
|
ok = True
|
|
status_code = 200
|
|
text = ""
|
|
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def json(self):
|
|
return {"data": self.payload}
|
|
|
|
class Session:
|
|
def __init__(self) -> None:
|
|
self.offsets = []
|
|
|
|
def get(self, url, *, params, timeout):
|
|
del url, timeout
|
|
self.offsets.append(params["offset"])
|
|
offset = params["offset"]
|
|
page = [{"id": index} for index in range(offset, min(offset + 200, 401))]
|
|
return Response({"items": page, "total": 401})
|
|
|
|
session = Session()
|
|
items = provisioner.list_paginated_items(session, "http://backend/api/v1/projects", timeout=30)
|
|
|
|
assert len(items) == 401
|
|
assert session.offsets == [0, 200, 400]
|
|
|
|
|
|
def test_kempen_scope_operator_is_packaged_and_exposed_in_map_flow() -> None:
|
|
dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
|
|
readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
|
|
workspace = (ROOT / "frontend/src/components/map/MapWorkspace.tsx").read_text(encoding="utf-8")
|
|
app = (ROOT / "frontend/src/App.tsx").read_text(encoding="utf-8")
|
|
|
|
assert "COPY scripts/geographic_scopes.py" in dockerfile
|
|
assert "COPY scripts/provision_geographic_scope.py" in dockerfile
|
|
assert "py_compile scripts/provision_geographic_scope.py" in readiness
|
|
assert "Immutable scope dataset" in (ROOT / "scripts/provision_geographic_scope.py").read_text(encoding="utf-8")
|
|
assert "Kempen (28 gemeenten)" in workspace
|
|
assert 'aria-label="Regio"' not in workspace
|
|
assert 'aria-label="Ingeladen regiobereik"' in workspace
|
|
assert "Snel naar een gemeente (optioneel)" in workspace
|
|
assert "projects={projects}" in app
|
|
map_props = app.split("<MapWorkspace", maxsplit=1)[1].split("/>", maxsplit=1)[0]
|
|
assert "onSelectProject={selectProject}" not in map_props
|
|
|
|
|
|
def test_project_switches_reset_scoped_state_and_prefer_the_regional_context() -> None:
|
|
bootstrap = (ROOT / "frontend/src/hooks/useWorkbenchBootstrap.ts").read_text(encoding="utf-8")
|
|
project_workspace = (ROOT / "frontend/src/hooks/useProjectWorkspace.ts").read_text(encoding="utf-8")
|
|
map_state = (ROOT / "frontend/src/hooks/useMapWorkspaceState.ts").read_text(encoding="utf-8")
|
|
dataset_workflow = (ROOT / "frontend/src/hooks/useDatasetWorkflow.ts").read_text(encoding="utf-8")
|
|
|
|
selected_project_branch = bootstrap.split("if (!selectedProjectId)", maxsplit=1)[1]
|
|
assert "resetProjectData()" in selected_project_branch
|
|
assert "resetDatasetForProject()" in selected_project_branch
|
|
assert "projectDataRequestSequence" in project_workspace
|
|
assert "REGIONAL_WORKSPACE_PROJECT_NAME" in project_workspace
|
|
assert "vervoerregio|operationele grens" in map_state
|
|
assert "datasets.find(isOperationalScopeBoundaryDataset)" in dataset_workflow
|