JensandClaude Opus 5
6572e4ad5f
93 test files read a single frontend source and asserted identifiers in it. The
MapWorkspace split showed what that costs: 24 tests went red for a move that
changed no behaviour at all. A contract belongs to the feature — a container,
its hooks, its domain layer — not to whichever file currently holds it.
232 read sites now resolve through read_feature(). The distinction that makes
this safe is direction: a *positive* contract ("this is wired") may widen,
because the identifier must still exist somewhere in the feature; a *negative*
one ("this component performs no transport") is a statement about one file, and
widening it would quietly weaken the check. The 73 single-file reads that
remain are exactly those, and a guard now enforces the rule for new tests.
Verified rather than assumed: of the 732 migrated positive assertions, 644 still
match exactly one module — as specific as before — and the other 86 already
spanned a container and its hook by nature. Two apparent misses are an artefact
of the checking regex reading an escaped newline literally.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
250 lines
9.0 KiB
Python
250 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import gzip
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from shapely.geometry import box, mapping, shape
|
|
from tests.frontend_contract import read_feature
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
|
|
def load_script():
|
|
path = SCRIPTS / "provision_regional_historical_landuse.py"
|
|
spec = importlib.util.spec_from_file_location("test_provision_regional_historical_landuse", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class FakeResponse:
|
|
status_code = 200
|
|
ok = True
|
|
text = ""
|
|
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
self.content = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
|
|
class FakeSession:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.calls = []
|
|
|
|
def post(self, url, **kwargs):
|
|
self.calls.append((url, kwargs))
|
|
if not self.responses:
|
|
raise AssertionError("Unexpected source/API request")
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def source_feature(feature_id: str, geometry, landuse_class: str = "bebouwing"):
|
|
return {
|
|
"type": "Feature",
|
|
"id": feature_id,
|
|
"geometry": mapping(geometry),
|
|
"properties": {"KLASSE": landuse_class},
|
|
}
|
|
|
|
|
|
def test_member_boundaries_require_every_approved_municipality(tmp_path: Path) -> None:
|
|
module = load_script()
|
|
scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
|
|
features = []
|
|
for index, member in enumerate(scope.members):
|
|
features.append(
|
|
{
|
|
"type": "Feature",
|
|
"geometry": mapping(box(index, 0, index + 0.9, 0.9)),
|
|
"properties": {"nis_code": member.nis_code, "municipality": member.name},
|
|
}
|
|
)
|
|
path = tmp_path / "members.geojson"
|
|
path.write_text(json.dumps({"type": "FeatureCollection", "features": features}), encoding="utf-8")
|
|
|
|
boundaries = module.load_member_boundaries(path, scope)
|
|
|
|
assert list(boundaries) == list(scope.nis_codes)
|
|
assert len(boundaries) == 28
|
|
assert boundaries["13025"][0].name == "Mol"
|
|
|
|
|
|
def test_partition_retains_exact_source_response_and_clips_to_municipality(tmp_path: Path) -> None:
|
|
module = load_script()
|
|
definition = next(item for item in module.THEMES if item.key == "buildings")
|
|
member = module.ScopeMember("Mol", "13025")
|
|
boundary = box(5.0, 51.0, 5.1, 51.1)
|
|
payload = {
|
|
"type": "FeatureCollection",
|
|
"features": [source_feature("Lgbrk1778.1", box(4.95, 51.02, 5.05, 51.08))],
|
|
}
|
|
session = FakeSession([FakeResponse(payload)])
|
|
|
|
manifest = module.prepare_partition(
|
|
session,
|
|
output_root=tmp_path,
|
|
year=1778,
|
|
definition=definition,
|
|
scope_key="kempen-transport-region",
|
|
member=member,
|
|
boundary=boundary,
|
|
page_size=50,
|
|
max_features=100,
|
|
simplify_tolerance_degrees=0.0,
|
|
timeout=30,
|
|
force=False,
|
|
)
|
|
|
|
output = json.loads(Path(manifest["output_path"]).read_text(encoding="utf-8"))
|
|
feature = output["features"][0]
|
|
raw_path = Path(manifest["output_path"]).parent / manifest["raw_pages"][0]["artifact_path"]
|
|
assert manifest["feature_count"] == 1
|
|
assert manifest["source_feature_count"] == 1
|
|
assert feature["id"] == "Lgbrk1778.1:13025"
|
|
assert feature["properties"]["original_source_feature_id"] == "Lgbrk1778.1"
|
|
assert feature["properties"]["coverage_scope"] == "kempen-transport-region"
|
|
assert shape(feature["geometry"]).bounds == (5.0, 51.02, 5.05, 51.08)
|
|
assert gzip.decompress(raw_path.read_bytes()) == FakeResponse(payload).content
|
|
|
|
cached = module.prepare_partition(
|
|
FakeSession([]),
|
|
output_root=tmp_path,
|
|
year=1778,
|
|
definition=definition,
|
|
scope_key="kempen-transport-region",
|
|
member=member,
|
|
boundary=boundary,
|
|
page_size=50,
|
|
max_features=100,
|
|
simplify_tolerance_degrees=0.0,
|
|
timeout=30,
|
|
force=False,
|
|
)
|
|
assert cached["output_sha256"] == manifest["output_sha256"]
|
|
|
|
|
|
def test_regional_snapshot_assembles_unique_partition_features(tmp_path: Path) -> None:
|
|
module = load_script()
|
|
scope = module.GeographicScope(
|
|
key="test-region",
|
|
display_name="Test region",
|
|
project_name="Test",
|
|
project_region="Test",
|
|
area_name="Test area",
|
|
authority_name="Authority",
|
|
authority_url="https://example.test",
|
|
scope_type="test",
|
|
limitation_message="Test only",
|
|
members=(module.ScopeMember("Left", "10001"), module.ScopeMember("Right", "10002")),
|
|
)
|
|
definition = next(item for item in module.THEMES if item.key == "water")
|
|
partitions = []
|
|
for index, member in enumerate(scope.members):
|
|
path, _manifest_path, _raw_dir = module.partition_paths(tmp_path, 1873, definition.key, member.nis_code)
|
|
feature = source_feature(f"water.{index}:{member.nis_code}", box(index, 0, index + 0.5, 0.5), "water")
|
|
module.atomic_write_json(path, {"type": "FeatureCollection", "features": [feature]})
|
|
partitions.append(
|
|
{
|
|
"municipality": member.name,
|
|
"nis_code": member.nis_code,
|
|
"source_feature_count": 1,
|
|
"feature_count": 1,
|
|
"raw_pages": [{"artifact_path": "unused"}],
|
|
"output_path": str(path),
|
|
"output_sha256": module.sha256_file(path),
|
|
}
|
|
)
|
|
|
|
output_path, manifest = module.assemble_snapshot(
|
|
output_root=tmp_path,
|
|
scope=scope,
|
|
year=1873,
|
|
definition=definition,
|
|
partitions=partitions,
|
|
max_total_features=10,
|
|
)
|
|
payload = json.loads(output_path.read_text(encoding="utf-8"))
|
|
|
|
assert manifest["coverage_complete"] is True
|
|
assert manifest["feature_count"] == 2
|
|
assert manifest["empty_partitions"] == []
|
|
assert len({feature["id"] for feature in payload["features"]}) == 2
|
|
|
|
|
|
def test_upload_contract_is_regional_temporal_and_partition_audited(tmp_path: Path) -> None:
|
|
module = load_script()
|
|
scope = module.GEOGRAPHIC_SCOPES["kempen-transport-region"]
|
|
definition = next(item for item in module.THEMES if item.key == "roads")
|
|
path = tmp_path / "roads.geojson"
|
|
path.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
|
|
response_payload = {"data": {"id": "dataset-id", "feature_count": 42}}
|
|
session = FakeSession([FakeResponse(response_payload)])
|
|
manifest = {
|
|
"coverage_complete": True,
|
|
"empty_partitions": [],
|
|
"partitions": [{} for _ in scope.members],
|
|
"partition_identity_sha256": "partition-hash",
|
|
"output_sha256": "output-hash",
|
|
"generated_at": "2026-07-15T00:00:00+00:00",
|
|
}
|
|
|
|
result = module.upload_snapshot(
|
|
session,
|
|
base_url="http://backend:8000",
|
|
project_id="project-id",
|
|
area_id="area-id",
|
|
scope=scope,
|
|
year=1969,
|
|
definition=definition,
|
|
path=path,
|
|
manifest=manifest,
|
|
simplify_tolerance_degrees=0.00001,
|
|
timeout=30,
|
|
)
|
|
data = session.calls[0][1]["data"]
|
|
source_metadata = json.loads(data["source_metadata_json"])
|
|
provenance = json.loads(data["provenance_metadata_json"])
|
|
|
|
assert result["id"] == "dataset-id"
|
|
assert data["area_id"] == "area-id"
|
|
assert data["temporal_series_key"].endswith(":roads:kempen-transport-region")
|
|
assert data["observed_at"] == "1969-01-01T00:00:00Z"
|
|
assert source_metadata["member_count"] == 28
|
|
assert source_metadata["partitioned_source_audit"] is True
|
|
assert source_metadata["geometry_clipped_to_area"] is True
|
|
assert source_metadata["identity_stable"] is False
|
|
assert source_metadata["semantic_metrics"] is False
|
|
assert source_metadata["selection_aggregation"]["metric_key"] == "roads_area"
|
|
assert source_metadata["selection_aggregation"]["label"] == "Oppervlakte historische wegen"
|
|
assert provenance["partition_count"] == 28
|
|
assert provenance["raw_source_responses_retained"] is True
|
|
assert provenance["geometry_clipped_to_area"] is True
|
|
|
|
|
|
def test_regional_historical_operator_is_packaged_and_release_checked() -> 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")
|
|
app = read_feature("shell")
|
|
workspace = read_feature("map_workspace")
|
|
|
|
assert "COPY scripts/provision_regional_historical_landuse.py" in dockerfile
|
|
assert "py_compile scripts/provision_regional_historical_landuse.py" in readiness
|
|
assert "onSetContextSourceLabel={setMapContextSourceLabel}" in app
|
|
assert "analysisMode === 'evolution' ? activeTemporalSeriesGroup?.label" in workspace
|