366 lines
14 KiB
Python
366 lines
14 KiB
Python
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&service=CSW&"
|
|
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
|