Add governed orthophoto release preflight

backend/README.md

@@ -1300,6 +1300,26 @@ Settings: `ORTHOPHOTO_ENABLED`, `ORTHOPHOTO_WMS_URL`,
 `ORTHOPHOTO_CACHE_TTL_HOURS`. Keep the official HTTPS URL and 1 m profile
 unless a separately verified deployment/model profile requires a change.
 
+Before a future `most_recent` source release is allowed into a governed pixel
+stage, run the metadata-only preflight for the exact intended rectangle:
+
+```bash
+docker exec geointel python /app/scripts/orthophoto_release_preflight.py \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --bbox 5.110 51.180 5.117 51.185 \
+  --refresh-catalog
+```
+
+The command reads canonical API envelopes, exact official WMS capabilities,
+WCS `DescribeCoverage` and at most 64 queryable flight-day points. It never
+requests raster pixels or mutates application/storage state. Only
+`staging_permitted=true` may feed a future separate staging command. `current`,
+remote-older, mixed/incorrect flight years and legacy local values such as
+`most_recent_at_2026-07-15` remain non-stageable. The report's point grid is
+flight-date evidence; complete selected-area coverage comes from containment
+inside the official 15 cm WCS raster domain.
+
 ## Governed DHMV terrain acquisition
 
 `GET /api/v1/projects/{project_id}/datasets/dhmv/products` exposes the fixed

backend/tests/test_docker_runtime_config.py

@@ -92,6 +92,7 @@ def test_all_in_one_dockerfile_copies_operator_scripts_for_runtime_use() -> None
         "run_split_background_promotion_workflow.sh",
         "activate_promoted_yolo_candidate.py",
         "manage_grb_refresh.py",
+        "orthophoto_release_preflight.py",
     }
     for script_name in required_runtime_scripts:
         assert f"COPY scripts/{script_name} /app/scripts/{script_name}" in dockerfile

backend/tests/test_sprint230_orthophoto_release_preflight.py

@@ -0,0 +1,365 @@
+from __future__ import annotations
+
+import argparse
+from email.message import Message
+from hashlib import sha256
+import importlib.util
+import json
+from pathlib import Path
+import sys
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+
+
+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 / "orthophoto_release_preflight.py"
+    module_name = "test_orthophoto_release_preflight_sprint230"
+    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
+
+
+PREFLIGHT = load_script()
+PROJECT_ID = "00000000-0000-0000-0000-000000000001"
+
+
+def arguments(**overrides) -> argparse.Namespace:
+    values = {
+        "project_id": PROJECT_ID,
+        "scope": PREFLIGHT.DEFAULT_SCOPE,
+        "api_url": "http://127.0.0.1:8000/api/v1",
+        "bbox": [5.110, 51.180, 5.117, 51.185],
+        "refresh_catalog": True,
+        "api_timeout": 30,
+        "wms_timeout": 10,
+    }
+    values.update(overrides)
+    return argparse.Namespace(**values)
+
+
+def capabilities(*, queryable: bool = True, feature_info: bool = True) -> bytes:
+    info_format = "<Format>application/geo+json</Format>" if feature_info else "<Format>text/plain</Format>"
+    queryable_value = "1" if queryable else "0"
+    metadata_url = (
+        "https://metadata.vlaanderen.be/srv/dut/csw?request=GetRecordById&amp;service=CSW&amp;"
+        f"id={PREFLIGHT.METADATA_IDENTIFIER}"
+    )
+    return f"""
+    <WMS_Capabilities version="1.3.0" xmlns="http://www.opengis.net/wms"
+      xmlns:xlink="http://www.w3.org/1999/xlink">
+      <Capability>
+        <Request><GetFeatureInfo>{info_format}</GetFeatureInfo></Request>
+        <Layer>
+          <CRS>EPSG:31370</CRS>
+          <BoundingBox CRS="EPSG:31370" minx="22000" miny="150000" maxx="259000" maxy="245000" />
+          <Layer queryable="0"><Name>Ortho</Name><MetadataURL><OnlineResource xlink:href="{metadata_url}" /></MetadataURL></Layer>
+          <Layer queryable="{queryable_value}"><Name>Vliegdagcontour</Name><MetadataURL><OnlineResource xlink:href="{metadata_url}" /></MetadataURL></Layer>
+        </Layer>
+      </Capability>
+    </WMS_Capabilities>
+    """.encode()
+
+
+def coverage_description(*, coverage_id: str = "Ortho", resolution: float = 0.15) -> bytes:
+    return f"""
+    <wcs:CoverageDescriptions xmlns:wcs="http://www.opengis.net/wcs/2.0"
+      xmlns:gml="http://www.opengis.net/gml/3.2" xmlns:gmlcov="http://www.opengis.net/gmlcov/1.0"
+      xmlns:swe="http://www.opengis.net/swe/2.0">
+      <wcs:CoverageDescription>
+        <gml:boundedBy><gml:Envelope srsName="http://www.opengis.net/def/crs/EPSG/0/31370">
+          <gml:lowerCorner>21375 152250</gml:lowerCorner><gml:upperCorner>259500 244875</gml:upperCorner>
+        </gml:Envelope></gml:boundedBy>
+        <wcs:CoverageId>{coverage_id}</wcs:CoverageId>
+        <gml:domainSet><gml:RectifiedGrid dimension="2">
+          <gml:offsetVector>{resolution} 0</gml:offsetVector><gml:offsetVector>0 -{resolution}</gml:offsetVector>
+        </gml:RectifiedGrid></gml:domainSet>
+        <gmlcov:rangeType><swe:DataRecord>
+          <swe:field name="Band_1"/><swe:field name="Band_2"/><swe:field name="Band_3"/>
+        </swe:DataRecord></gmlcov:rangeType>
+        <wcs:ServiceParameters><wcs:CoverageSubtype>RectifiedGridCoverage</wcs:CoverageSubtype>
+          <wcs:nativeFormat>image/tiff</wcs:nativeFormat></wcs:ServiceParameters>
+      </wcs:CoverageDescription>
+    </wcs:CoverageDescriptions>
+    """.encode()
+
+
+def catalog_item(body: bytes, *, local: str | None = "most_recent_at_2026-07-15", remote: str = "2025.04") -> dict:
+    if local is None:
+        comparison = "no_local_data"
+    elif PREFLIGHT.EDITION_PATTERN.fullmatch(local):
+        comparison = "same" if local == remote else "different"
+    else:
+        comparison = "not_comparable"
+    return {
+        "source_name": PREFLIGHT.SOURCE_NAME,
+        "status": "available",
+        "reachable": True,
+        "matched_layers": ["Ortho", "Vliegdagcontour"],
+        "missing_layers": [],
+        "metadata_identifier": PREFLIGHT.METADATA_IDENTIFIER,
+        "metadata_url": f"https://metadata.vlaanderen.be/{PREFLIGHT.METADATA_IDENTIFIER}",
+        "remote_title": f"Orthofoto meest recent, {remote}",
+        "remote_version": remote,
+        "remote_modified_at": "2026-04-27T00:00:00Z",
+        "remote_published_at": "2025-12-11T00:00:00Z",
+        "local_source_version": local,
+        "comparison_status": comparison,
+        "checked_at": "2026-07-17T00:00:00Z",
+        "endpoint_url": (
+            "https://geo.api.vlaanderen.be/OMWRGBMRVL/wms?"
+            "SERVICE=WMS&VERSION=1.3.0&REQUEST=GetCapabilities"
+        ),
+        "capabilities_sha256": sha256(body).hexdigest(),
+    }
+
+
+def product() -> dict:
+    return {
+        "key": "most_recent",
+        "display_name": "Meest recente winterluchtbeeld",
+        "observation_label": "Meest recent beschikbaar",
+        "temporal_granularity": "snapshot",
+        "native_resolution_m": 0.15,
+        "supports_detection": True,
+        "color_mode": "rgb",
+        "catalog_url": PREFLIGHT.CATALOG_URL,
+        "limitation_message": "rolling source",
+    }
+
+
+def loader(body: bytes, item: dict):
+    def load(_api_url: str, path: str, _timeout: int) -> dict:
+        if path == f"projects/{PROJECT_ID}":
+            return {"id": PROJECT_ID, "name": "Kempen Regional Workbench"}
+        if path.endswith("/datasets/orthophoto/products"):
+            return {"items": [product()], "total": 1}
+        if "/datasets/source-catalog-probes?" in path:
+            return {"items": [item]}
+        raise AssertionError(path)
+
+    return load
+
+
+class Response:
+    def __init__(self, body: bytes, *, url: str, content_type: str) -> None:
+        self.body = body
+        self.url = url
+        self.headers = Message()
+        self.headers["Content-Type"] = content_type
+        self.headers["Content-Length"] = str(len(body))
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args) -> None:
+        return None
+
+    def geturl(self) -> str:
+        return self.url
+
+    def read(self, size: int = -1) -> bytes:
+        return self.body if size < 0 else self.body[:size]
+
+
+def opener(
+    body: bytes,
+    *,
+    flight_year: int = 2025,
+    empty: bool = False,
+    requests: list[str] | None = None,
+    coverage_body: bytes | None = None,
+):
+    def open_request(request, timeout: int):
+        assert timeout == 10
+        url = request.full_url
+        query = parse_qs(urlparse(url).query)
+        request_name = (query.get("REQUEST") or [""])[0]
+        if requests is not None:
+            requests.append(request_name)
+        if request_name == "GetCapabilities":
+            return Response(body, url=url, content_type="text/xml")
+        if request_name == "DescribeCoverage":
+            return Response(coverage_body or coverage_description(), url=url, content_type="text/xml")
+        assert request_name == "GetFeatureInfo"
+        assert query["LAYERS"] == ["Vliegdagcontour"]
+        features = [] if empty else [
+            {
+                "type": "Feature",
+                "geometry": None,
+                "properties": {"OpnDatum": f"5/4/{flight_year}", "FID": "4"},
+                "layerName": "Vliegdagcontour",
+            }
+        ]
+        payload = json.dumps({"type": "FeatureCollection", "features": features}).encode()
+        return Response(payload, url=url, content_type="application/geo+json")
+
+    return open_request
+
+
+def run(*, local: str | None = "most_recent_at_2026-07-15", remote: str = "2025.04", flight_year: int = 2025):
+    body = capabilities()
+    item = catalog_item(body, local=local, remote=remote)
+    calls: list[str] = []
+    report = PREFLIGHT.run_preflight(
+        arguments(),
+        loader=loader(body, item),
+        opener=opener(body, flight_year=flight_year, requests=calls),
+    )
+    return report, calls
+
+
+def test_live_contract_shape_blocks_non_comparable_legacy_local_version() -> None:
+    report, calls = run()
+
+    assert report["status"] == "passed"
+    assert report["release"]["status"] == "blocked_local_version"
+    assert report["release"]["remote_edition"] == "2025.04"
+    assert report["flight_day_coverage"]["sample_count"] == 20
+    assert report["flight_day_coverage"]["sample_coverage_ratio"] == 1.0
+    assert report["flight_day_coverage"]["flight_years"] == [2025]
+    assert report["staging_permitted"] is False
+    assert report["next_action"] == "establish_official_local_edition_before_staging"
+    assert report["pixel_requests_performed"] == 0
+    assert set(calls) == {"GetCapabilities", "DescribeCoverage", "GetFeatureInfo"}
+    assert report["coverage_domain"]["selected_area_fully_inside_domain"] is True
+    assert report["coverage_domain"]["pixel_data_requested"] is False
+
+
+@pytest.mark.parametrize(
+    ("local", "remote", "expected_status", "stageable"),
+    [
+        (None, "2025.04", "not_loaded", True),
+        ("2025.04", "2025.04", "current", False),
+        ("2024.01", "2025.04", "update_available", True),
+        ("2026.01", "2025.04", "blocked_remote_older", False),
+    ],
+)
+def test_release_ordering_requires_comparable_official_editions(
+    local: str | None,
+    remote: str,
+    expected_status: str,
+    stageable: bool,
+) -> None:
+    report, _ = run(local=local, remote=remote)
+
+    assert report["release"]["status"] == expected_status
+    assert report["staging_permitted"] is stageable
+
+
+def test_flight_year_must_match_remote_release_year() -> None:
+    report, _ = run(local="2024.01", flight_year=2024)
+
+    assert report["release"]["status"] == "update_available"
+    assert report["flight_year_matches_release"] is False
+    assert report["staging_permitted"] is False
+    assert report["next_action"] == "split_or_review_selection_flight_years"
+
+
+def test_capabilities_hash_drift_fails_closed() -> None:
+    body = capabilities()
+    item = catalog_item(body, local=None)
+    changed = body.replace(b"queryable=\"1\"", b"queryable=\"0\"")
+
+    with pytest.raises(RuntimeError, match="changed after"):
+        PREFLIGHT.run_preflight(
+            arguments(),
+            loader=loader(body, item),
+            opener=opener(changed),
+        )
+
+
+@pytest.mark.parametrize(
+    ("body", "message"),
+    [
+        (capabilities(queryable=False), "not queryable"),
+        (capabilities(feature_info=False), "does not advertise GeoJSON"),
+    ],
+)
+def test_flight_day_capability_requirements_fail_closed(body: bytes, message: str) -> None:
+    item = catalog_item(body, local=None)
+    with pytest.raises(RuntimeError, match=message):
+        PREFLIGHT.run_preflight(arguments(), loader=loader(body, item), opener=opener(body))
+
+
+def test_missing_flight_day_coverage_fails_closed() -> None:
+    body = capabilities()
+    item = catalog_item(body, local=None)
+    with pytest.raises(RuntimeError, match="has no coverage"):
+        PREFLIGHT.run_preflight(
+            arguments(),
+            loader=loader(body, item),
+            opener=opener(body, empty=True),
+        )
+
+
+@pytest.mark.parametrize(
+    "coverage_body",
+    [coverage_description(coverage_id="Other"), coverage_description(resolution=0.25)],
+)
+def test_coverage_domain_identity_and_resolution_fail_closed(coverage_body: bytes) -> None:
+    body = capabilities()
+    item = catalog_item(body, local=None)
+    with pytest.raises(RuntimeError, match="WCS"):
+        PREFLIGHT.run_preflight(
+            arguments(),
+            loader=loader(body, item),
+            opener=opener(body, coverage_body=coverage_body),
+        )
+
+
+def test_product_variant_and_catalog_comparison_are_bound() -> None:
+    wrong = product()
+    wrong["color_mode"] = "panchromatic"
+    with pytest.raises(RuntimeError, match="product variant"):
+        PREFLIGHT.validate_product([wrong])
+
+    body = capabilities()
+    item = catalog_item(body, local="2025.04")
+    item["comparison_status"] = "different"
+    with pytest.raises(RuntimeError, match="internally inconsistent"):
+        PREFLIGHT.catalog_decision(item)
+
+
+def test_only_exact_official_wms_url_is_allowed() -> None:
+    valid = "https://geo.api.vlaanderen.be/OMWRGBMRVL/wms?SERVICE=WMS&REQUEST=GetCapabilities"
+    assert PREFLIGHT._validate_wms_url(valid, request_name="GetCapabilities").endswith("/OMWRGBMRVL/wms")
+    with pytest.raises(RuntimeError, match="outside the official allowlist"):
+        PREFLIGHT._validate_wms_url(valid.replace("geo.api.vlaanderen.be", "example.com"))
+    with pytest.raises(RuntimeError, match="not an exact"):
+        PREFLIGHT._validate_wms_url(valid.replace("GetCapabilities", "GetMap"), request_name="GetCapabilities")
+
+
+def test_bbox_matches_acquisition_bounds_and_grid_is_bounded() -> None:
+    extent = [22000.0, 150000.0, 259000.0, 245000.0]
+    selection = PREFLIGHT.validate_bbox([5.110, 51.180, 5.117, 51.185], extent)
+    samples = PREFLIGHT._sample_grid(selection)
+    assert len(samples) == 20
+    assert len(samples) <= PREFLIGHT.MAX_SAMPLE_COUNT
+    with pytest.raises(RuntimeError, match="at least"):
+        PREFLIGHT.validate_bbox([5.110, 51.180, 5.1105, 51.1805], extent)
+    with pytest.raises(RuntimeError, match="may not exceed"):
+        PREFLIGHT.validate_bbox([5.05, 51.15, 5.25, 51.33], extent)
+
+
+def test_operator_preflight_is_read_only_packaged_and_readiness_checked() -> None:
+    script = (SCRIPTS / "orthophoto_release_preflight.py").read_text(encoding="utf-8")
+    dockerfile = (ROOT / "deploy" / "unraid" / "Dockerfile.all-in-one").read_text(encoding="utf-8")
+    readiness = (SCRIPTS / "run_readiness_check.sh").read_text(encoding="utf-8")
+
+    assert "GetMap" not in script
+    assert "GetCoverage" not in script
+    assert "/datasets/upload" not in script
+    assert "INSERT INTO" not in script
+    assert "COPY scripts/orthophoto_release_preflight.py" in dockerfile
+    assert "py_compile scripts/orthophoto_release_preflight.py" in readiness