from __future__ import annotations import argparse from datetime import datetime, timezone import importlib.util import json from pathlib import Path from types import SimpleNamespace import sys import numpy as np import pytest from rasterio.io import MemoryFile from rasterio.transform import from_origin 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(name: str): path = SCRIPTS / name module_name = f"test_{path.stem}_sprint231" spec = importlib.util.spec_from_file_location(module_name, path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) return module MANAGER = load_script("manage_orthophoto_release.py") def arguments(tmp_path: Path, **overrides) -> argparse.Namespace: values = { "action": "plan", "project_id": "00000000-0000-0000-0000-000000000001", "scope": "kempen-transport-region", "api_url": "http://127.0.0.1:8000/api/v1", "bbox": [5.110, 51.180, 5.113, 51.182], "area_id": None, "confirm_edition": None, "confirm_plan_sha256": None, "confirm_review_sha256": None, "approve": False, "reviewer": None, "review_note": "", "establish_official_baseline": False, "confirm_local_version": None, "plan_path": None, "review_path": None, "evidence_root": tmp_path / "operator-evidence" / "orthophoto-release", "refresh_catalog": False, "api_timeout": 180, "wms_timeout": 60, "import_timeout": 600, "max_response_mb": 32, } values.update(overrides) return argparse.Namespace(**values) def report(*, status: str = "update_available", local: str | None = "2024.03") -> dict: return { "schema_version": 1, "status": "passed", "generated_at": "2026-07-17T10:00:00Z", "project_id": "00000000-0000-0000-0000-000000000001", "scope": "kempen-transport-region", "product": { "key": "most_recent", "display_name": "Orthofoto meest recent", "temporal_granularity": "snapshot", "native_resolution_m": 0.15, "supports_detection": True, "color_mode": "rgb", "catalog_url": MANAGER.preflight.CATALOG_URL, }, "release": { "status": status, "remote_edition": "2025.04", "remote_year": 2025, "local_source_version": local, "comparison_status": "different" if status == "update_available" else "not_comparable", "metadata_identifier": MANAGER.preflight.METADATA_IDENTIFIER, "metadata_url": "https://metadata.vlaanderen.be/srv/dut/catalog.search#/metadata/f5304d6d", "remote_title": "Orthofoto meest recent, 2025.04", "remote_modified_at": "2026-04-27T00:00:00Z", "remote_published_at": "2025-12-11T00:00:00Z", "catalog_checked_at": "2026-07-17T10:00:00Z", "capabilities_url": MANAGER.WMS_BASE_URL + "?SERVICE=WMS&REQUEST=GetCapabilities", "capabilities_sha256": "a" * 64, }, "capabilities": { "service_version": "1.3.0", "capabilities_sha256": "a" * 64, "layers": ["Ortho", "Vliegdagcontour"], "vliegdagcontour_queryable": True, "feature_info_format": "application/geo+json", "extent_epsg31370": [0.0, 0.0, 300000.0, 300000.0], "metadata_identifier": MANAGER.preflight.METADATA_IDENTIFIER, }, "coverage_domain": { "coverage_id": "Ortho", "crs": "EPSG:31370", "extent_epsg31370": [0.0, 0.0, 300000.0, 300000.0], "native_resolution_m": 0.15, "band_count": 3, "native_format": "image/tiff", "coverage_description_sha256": "b" * 64, "selected_area_fully_inside_domain": True, "pixel_data_requested": False, }, "selection": { "bbox_epsg4326": [5.110, 51.180, 5.113, 51.182], "bbox_epsg31370": [200000.0, 210000.0, 200200.0, 210160.0], "width_m": 200.0, "height_m": 160.0, }, "flight_day_coverage": { "status": "passed", "mode": "official_queryable_flight_day_grid", "sample_count": 4, "grid_columns": 2, "grid_rows": 2, "covered_sample_count": 4, "sample_coverage_ratio": 1.0, "flight_dates": ["5/4/2025"], "flight_years": [2025], "feature_ids": ["123"], "sample_evidence_sha256": "c" * 64, "claim_boundary": "Bounded point evidence, not a polygon-union proof.", }, "flight_year_matches_release": True, "staging_permitted": status in {"not_loaded", "update_available"}, "next_action": "governed_pixel_stage", "pixel_requests_performed": 0, "datasets_mutated": 0, "automatic_staging": False, "automatic_import": False, } def raw_rgb_tiff(width: int = 200, height: int = 160) -> bytes: pixels = np.zeros((3, height, width), dtype=np.uint8) pixels[0] = 80 pixels[1] = np.arange(width, dtype=np.uint8)[None, :] pixels[2] = np.arange(height, dtype=np.uint8)[:, None] with MemoryFile() as memory: with memory.open( driver="GTiff", width=width, height=height, count=3, dtype="uint8", crs="EPSG:31370", transform=from_origin(200000.0, 210160.0, 1.0, 1.0), ) as dataset: dataset.write(pixels) return memory.read() class Response: def __init__(self, body: bytes, url: str, content_type: str = "image/tiff") -> None: self.body = body self.url = url self.headers = {"Content-Type": content_type, "Content-Length": str(len(body))} def __enter__(self): return self def __exit__(self, *_args): return None def geturl(self) -> str: return self.url def read(self, amount: int | None = None) -> bytes: return self.body if amount is None else self.body[:amount] def staged_plan(tmp_path: Path) -> tuple[argparse.Namespace, Path, dict, dict]: args = arguments( tmp_path, action="stage", confirm_edition="2025.04", ) release_report = report() request = MANAGER.map_request(release_report) staged = MANAGER.stage_artifacts( args, release_report, request, opener=lambda _request, timeout: Response(raw_rgb_tiff(), request["url"]), ) plan = MANAGER.build_staged_plan( args, release_report, MANAGER.authorize_stage(args, release_report), staged, ) path = MANAGER.default_plan_path(args, "2025.04") MANAGER.write_json(path, plan) return args, path, plan, staged def test_orthophoto_catalog_prefers_latest_official_edition_over_rolling_marker() -> None: from app.services.source_catalog_probe_service import _latest_local_version now = datetime(2026, 7, 17, tzinfo=timezone.utc) datasets = [ SimpleNamespace( source_name=MANAGER.SOURCE_NAME, source="", source_version="2025.04", imported_at=now, observed_at=now, id="official-2025", ), SimpleNamespace( source_name=MANAGER.SOURCE_NAME, source="", source_version="most_recent_at_2026-07-18", imported_at=datetime(2026, 7, 18, tzinfo=timezone.utc), observed_at=now, id="rolling", ), SimpleNamespace( source_name=MANAGER.SOURCE_NAME, source="", source_version="2026.02", imported_at=datetime(2026, 7, 16, tzinfo=timezone.utc), observed_at=now, id="official-2026", ), ] assert _latest_local_version(MANAGER.SOURCE_NAME, datasets) == "2026.02" def test_stage_requires_exact_official_or_explicit_legacy_baseline_confirmation(tmp_path: Path) -> None: args = arguments(tmp_path) normal = report() assert MANAGER.authorize_stage(args, normal)["mode"] == "normal_release" legacy = report(status="blocked_local_version", local="most_recent_at_2026-07-15") with pytest.raises(RuntimeError, match="establish-official-baseline"): MANAGER.authorize_stage(args, legacy) args.establish_official_baseline = True args.confirm_local_version = "most_recent_at_2026-07-15" assert MANAGER.authorize_stage(args, legacy)["mode"] == "explicit_legacy_baseline_transition" args.confirm_local_version = "different" with pytest.raises(RuntimeError, match="exact"): MANAGER.authorize_stage(args, legacy) def test_stage_fetches_one_allowlisted_map_and_writes_reviewable_rgb_evidence(tmp_path: Path) -> None: args = arguments(tmp_path) release_report = report() request = MANAGER.map_request(release_report) calls = [] def opener(http_request, timeout): calls.append((http_request.full_url, timeout)) return Response(raw_rgb_tiff(), request["url"]) staged = MANAGER.stage_artifacts(args, release_report, request, opener=opener) manifest = MANAGER.validate_staged_artifacts(args, Path(staged["manifest_path"])) assert calls == [(request["url"], 60)] assert manifest["pixel_request_count"] == 1 assert manifest["datasets_mutated"] == 0 assert manifest["normalized_geotiff"]["crs"] == "EPSG:31370" assert manifest["normalized_geotiff"]["band_count"] == 3 assert Path(manifest["review_preview"]["path"]).read_bytes().startswith(b"\x89PNG") with pytest.raises(RuntimeError, match="will not be overwritten"): MANAGER.stage_artifacts(args, release_report, request, opener=opener) assert len(calls) == 1 def test_getmap_redirect_and_response_limits_fail_closed(tmp_path: Path) -> None: request = MANAGER.map_request(report()) with pytest.raises(RuntimeError, match="allowlist"): MANAGER.fetch_map( request, timeout=5, max_bytes=10_000_000, opener=lambda *_args, **_kwargs: Response(raw_rgb_tiff(), "https://example.com/image.tif"), ) body = raw_rgb_tiff() with pytest.raises(RuntimeError, match="release limit"): MANAGER.fetch_map( request, timeout=5, max_bytes=len(body) - 1, opener=lambda *_args, **_kwargs: Response(body, request["url"]), ) def test_plan_and_artifact_tampering_are_rejected(tmp_path: Path) -> None: args, path, plan, staged = staged_plan(tmp_path) args.confirm_plan_sha256 = plan["plan_sha256"] _, loaded, _ = MANAGER.load_staged_plan(args, "2025.04") assert loaded == plan Path(staged["normalized_geotiff"]["path"]).write_bytes(b"tampered") with pytest.raises(RuntimeError, match="no longer match"): MANAGER.load_staged_plan(args, "2025.04") outside = tmp_path / "outside.json" with pytest.raises(RuntimeError, match="outside the governed"): MANAGER.governed_path(args, outside) def test_review_requires_named_approval_exact_plan_and_unchanged_preflight(tmp_path: Path) -> None: args, path, plan, staged = staged_plan(tmp_path) manifest = MANAGER.validate_staged_artifacts(args, Path(staged["manifest_path"])) with pytest.raises(RuntimeError, match="named --reviewer"): MANAGER.build_review_evidence(args, path, plan, manifest) args.approve = True args.reviewer = "Jens" review = MANAGER.build_review_evidence(args, path, plan, manifest) assert review["review_preview_sha256"] == manifest["review_preview"]["sha256"] assert review["review_sha256"] == MANAGER.canonical_sha256(review, "review_sha256") changed = report() changed["capabilities"]["capabilities_sha256"] = "d" * 64 with pytest.raises(RuntimeError, match="evidence changed"): MANAGER.require_preflight_unchanged(plan, changed, require_local=True) def test_approved_upload_uses_canonical_dataset_route_and_complete_provenance( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: args, path, plan, staged = staged_plan(tmp_path) manifest = MANAGER.validate_staged_artifacts(args, Path(staged["manifest_path"])) args.approve = True args.reviewer = "Jens" review = MANAGER.build_review_evidence(args, path, plan, manifest) captured = {} class UploadResponse: def __enter__(self): return self def __exit__(self, *_args): return None def read(self): return json.dumps( { "data": { "id": "dataset-1", "status": "ready", "source_name": MANAGER.SOURCE_NAME, "source_version": "2025.04", "checksum_sha256": manifest["normalized_geotiff"]["sha256"], } } ).encode("utf-8") def fake_urlopen(request, timeout): captured["url"] = request.full_url captured["body"] = request.data captured["timeout"] = timeout return UploadResponse() monkeypatch.setattr(MANAGER, "urlopen", fake_urlopen) dataset = MANAGER.upload_approved_dataset(args, plan, manifest, review) assert dataset["id"] == "dataset-1" assert captured["url"].endswith(f"/projects/{args.project_id}/datasets/upload") assert b'name="source_version"\r\n\r\n2025.04' in captured["body"] assert b'name="temporal_series_key"' in captured["body"] assert plan["plan_sha256"].encode("ascii") in captured["body"] assert review["review_sha256"].encode("ascii") in captured["body"] plan_path_json = json.dumps(str(MANAGER.default_plan_path(args, "2025.04")))[1:-1] review_path_json = json.dumps(str(MANAGER.default_review_path(args, "2025.04")))[1:-1] assert plan_path_json.encode("utf-8") in captured["body"] assert review_path_json.encode("utf-8") in captured["body"] def test_apply_target_must_be_loopback_api() -> None: assert MANAGER.internal_api_url("http://127.0.0.1:8000/api/v1").endswith("/api/v1") with pytest.raises(RuntimeError, match="local /api/v1"): MANAGER.internal_api_url("http://192.168.10.150:1202/api/v1") with pytest.raises(RuntimeError, match="local /api/v1"): MANAGER.internal_api_url("http://127.0.0.1:8000/not-api") def test_official_flight_dates_are_persisted_without_inventing_a_catalog_date() -> None: dates = MANAGER.parse_flight_dates(["5/4/2025", "2025-04-06", "05-04-2025"]) assert [value.date().isoformat() for value in dates] == ["2025-04-05", "2025-04-06"] with pytest.raises(RuntimeError, match="not safely parseable"): MANAGER.parse_flight_dates(["spring 2025"]) def test_applied_evidence_is_immutable_on_idempotent_retry(tmp_path: Path) -> None: args, plan_path, plan, staged = staged_plan(tmp_path) manifest = MANAGER.validate_staged_artifacts(args, Path(staged["manifest_path"])) args.approve = True args.reviewer = "Jens" review = MANAGER.build_review_evidence(args, plan_path, plan, manifest) review_path = MANAGER.default_review_path(args, "2025.04") dataset = { "id": "dataset-1", "checksum_sha256": manifest["normalized_geotiff"]["sha256"], } final_report = report(status="current", local="2025.04") final_report["release"]["comparison_status"] = "same" path, first = MANAGER.applied_evidence( args, plan_path, plan, review_path, review, dataset, final_report, reused=False, ) original_bytes = path.read_bytes() _, second = MANAGER.applied_evidence( args, plan_path, plan, review_path, review, dataset, final_report, reused=True, ) assert second == first assert path.read_bytes() == original_bytes assert second["dataset_status"] == "imported" def test_release_manager_is_packaged_and_compiled_by_readiness() -> 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") assert "COPY scripts/manage_orthophoto_release.py /app/scripts/manage_orthophoto_release.py" in dockerfile assert "-m py_compile scripts/manage_orthophoto_release.py" in readiness