from __future__ import annotations
from datetime import datetime, timezone
from email.message import Message
from pathlib import Path
from types import SimpleNamespace
from urllib.error import URLError
import uuid
from app.core.config import Settings
from app.models import Dataset
from app.services.source_catalog_probe_service import SourceCatalogProbeService
NOW = datetime(2026, 7, 16, 15, 0, tzinfo=timezone.utc)
GRB_METADATA_URL = (
"https://metadata.vlaanderen.be/srv/dut/csw?request=GetRecordById&service=CSW&"
"id=7C823055-7BBF-4D62-B55E-F85C30D53162&OUTPUTSCHEMA=http://www.isotc211.org/2005/gmd"
)
ORTHO_METADATA_URL = (
"https://metadata.vlaanderen.be/srv/dut/csw?request=GetRecordById&service=CSW&"
"id=f5304d6d-0dd4-43fd-a726-427af31e8d61&OUTPUTSCHEMA=http://www.isotc211.org/2005/gmd"
)
ALZ_RELEASE_URL = "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen"
STATBEL_DCAT_URL = "https://doc.statbel.be/publications/DCAT/DCAT_opendata_datasets.ttl"
def _wfs_capabilities(metadata_url: str = GRB_METADATA_URL) -> bytes:
layers = "".join(
f"""
GRB:{name}
"""
for name in ("GBG", "WBN", "WGO", "ADP", "WTZ")
)
return f"""
{layers}
""".encode()
def _wms_capabilities(metadata_url: str = ORTHO_METADATA_URL) -> bytes:
return f"""
Orthofoto
Orthotext/xml
Vliegdagcontourtext/xml
""".encode()
def _metadata(identifier: str, title: str, edition: str, modified: str, published: str) -> bytes:
return f"""
{identifier}
{modified}
{title}
{edition}
{published}
publication
""".encode()
def _alz_release_page(*, include_snapshot: bool = True, download_host: str = "www.landbouwvlaanderen.be") -> bytes:
snapshot = (
f''
"Landbouwgebruikspercelen 2026 – 1e snapshot (extractie 02-06-2026) - GPKG"
if include_snapshot
else ""
)
return f"""
{snapshot}
Definitieve datasets
Downloaden
Downloaden
""".encode()
def _statbel_dcat() -> bytes:
return b"""
@prefix dcat: .
@prefix dct: .
@prefix xsd: .
a dcat:Catalog ;
dct:modified "2026-07-07"^^xsd:date .
a dcat:Dataset ;
dct:title "Bevolking per statistische sector"@nl ;
dct:alternative "Bevolking per statistische sector [Periode: 2025]"@nl ;
dct:identifier "NodeID6475" ;
dct:license ;
dct:temporal [ dcat:startDate "2025-01-01"^^xsd:date ] ;
dcat:landingPage ;
dcat:distribution
,
,
,
.
"""
class _Response:
def __init__(self, content: bytes, content_type: str = "text/xml", *, content_length: int | None = None) -> None:
self.content = content
self.headers = Message()
self.headers["Content-Type"] = content_type
self.headers["Content-Length"] = str(content_length if content_length is not None else len(content))
self.headers["ETag"] = '"catalog-test"'
self.headers["Last-Modified"] = "Wed, 15 Jul 2026 10:00:00 GMT"
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self, size: int = -1) -> bytes:
return self.content if size < 0 else self.content[:size]
class _RedirectedResponse(_Response):
def __init__(self, content: bytes, final_url: str, content_type: str = "text/xml") -> None:
super().__init__(content, content_type)
self.final_url = final_url
def geturl(self) -> str:
return self.final_url
class _Query:
def __init__(self, datasets: list[Dataset]) -> None:
self.datasets = datasets
def filter(self, *_args):
return self
def all(self) -> list[Dataset]:
return self.datasets
class _Db:
def __init__(self, datasets: list[Dataset]) -> None:
self.datasets = datasets
def get(self, _model, _identifier):
return SimpleNamespace(id=_identifier)
def query(self, _model):
return _Query(self.datasets)
def _dataset(source_name: str, version: str) -> Dataset:
return Dataset(
id=uuid.uuid4(),
project_id=uuid.uuid4(),
name=f"{source_name} source",
dataset_type="vector" if source_name in {"grb", "statbel", "agentschap_landbouw_zeevisserij_agricultural_parcels"} else "raster",
source=source_name,
source_name=source_name,
source_version=version,
imported_at=NOW,
status="ready",
)
def _settings(**overrides) -> Settings:
values = {
"SOURCE_CATALOG_PROBE_ENABLED": True,
"SOURCE_CATALOG_GRB_WFS_URL": "https://geo.api.vlaanderen.be/GRB/wfs",
"SOURCE_CATALOG_ALZ_RELEASE_URL": ALZ_RELEASE_URL,
"SOURCE_CATALOG_STATBEL_DCAT_URL": STATBEL_DCAT_URL,
"SOURCE_CATALOG_STATBEL_MAX_RESPONSE_MB": 5,
"SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS": 3,
"SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB": 1,
"SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS": 0,
"ORTHOPHOTO_WMS_URL": "https://geo.api.vlaanderen.be/OMWRGBMRVL/wms",
"ORTHOPHOTO_WMS_LAYER": "Ortho",
}
values.update(overrides)
return Settings(**values)
def _opener(request, timeout):
assert timeout == 3
url = request.full_url
if url == STATBEL_DCAT_URL:
return _Response(_statbel_dcat(), "application/octet-stream")
if url == ALZ_RELEASE_URL:
return _Response(_alz_release_page(), "text/html; charset=utf-8")
if "metadata.vlaanderen.be" in url:
if "f5304d6d" in url:
return _Response(_metadata("f5304d6d", "Orthofoto meest recent, 2025.04", "2025.04", "2026-04-27", "2025-12-11"))
return _Response(_metadata("7C823055", "GRBgis", "Toestand 2026-07-15", "2026-07-15", "2026-07-15"))
if "/GRB/" in url:
return _Response(_wfs_capabilities())
return _Response(_wms_capabilities())
def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions() -> None:
SourceCatalogProbeService.clear_cache()
project_id = uuid.uuid4()
report = SourceCatalogProbeService.audit_project(
_Db(
[
_dataset("grb", "2026-07-14"),
_dataset("digitaal_vlaanderen_orthophoto", "most_recent_at_2026-07-14"),
_dataset("statbel", "2025"),
_dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2025-definitive"),
]
),
project_id,
settings=_settings(),
opener=_opener,
now=NOW,
)
by_source = {item.source_name: item for item in report.items}
assert report.summary.available_count == 4
assert report.summary.different_version_count == 1
assert by_source["grb"].remote_version == "Toestand 2026-07-15"
assert by_source["grb"].comparison_status == "different"
assert by_source["grb"].matched_layers == ["GBG", "WBN", "WGO", "ADP"]
assert by_source["grb"].advertised_layer_count == 5
assert by_source["digitaal_vlaanderen_orthophoto"].remote_version == "2025.04"
assert by_source["digitaal_vlaanderen_orthophoto"].comparison_status == "not_comparable"
assert by_source["digitaal_vlaanderen_orthophoto"].remote_published_at.year == 2025
statbel = by_source["statbel"]
assert statbel.service_type == "DCAT"
assert statbel.remote_version == "2025"
assert statbel.local_source_version == "2025"
assert statbel.comparison_status == "same"
assert statbel.metadata_identifier == "NodeID6475"
assert statbel.matched_layers == ["population_txt_current", "landing_page", "cc_by_4_0"]
assert "REDEGEO" in statbel.message
alz = by_source["agentschap_landbouw_zeevisserij_agricultural_parcels"]
assert alz.service_type == "HTML"
assert alz.remote_version == "2025-v3"
assert alz.comparison_status == "same"
assert alz.matched_layers == ["definitive_archive", "current_snapshot"]
assert "2026-v1" in alz.message
assert "voorlopig" in alz.message
assert all(item.capabilities_sha256 for item in report.items)
def test_catalog_probe_isolates_provider_failure() -> None:
def partial_opener(request, timeout):
if "/GRB/" in request.full_url:
raise URLError("offline")
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=partial_opener, now=NOW
)
by_source = {item.source_name: item for item in report.items}
assert by_source["grb"].status == "unavailable"
assert by_source["grb"].error_code == "CATALOG_PROVIDER_UNAVAILABLE"
assert by_source["digitaal_vlaanderen_orthophoto"].status == "available"
assert report.summary.unavailable_count == 1
def test_catalog_probe_rejects_metadata_redirect_outside_allowlist() -> None:
def malicious_opener(request, timeout):
if "/GRB/" in request.full_url:
return _Response(
_wfs_capabilities(
"https://example.com/csw?request=GetRecordById&id=evil&OUTPUTSCHEMA=http://www.isotc211.org/2005/gmd"
)
)
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=malicious_opener, now=NOW
)
grb = next(item for item in report.items if item.source_name == "grb")
assert grb.status == "unavailable"
assert grb.error_code == "CATALOG_METADATA_URL_REJECTED"
def test_catalog_probe_enforces_response_limit_without_reading_external_data() -> None:
def oversized_opener(request, timeout):
if "/GRB/" in request.full_url:
return _Response(b"", content_length=2 * 1024 * 1024)
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=oversized_opener, now=NOW
)
grb = next(item for item in report.items if item.source_name == "grb")
assert grb.status == "unavailable"
assert grb.error_code == "CATALOG_RESPONSE_TOO_LARGE"
def test_catalog_probe_revalidates_metadata_host_after_redirect() -> None:
def redirected_opener(request, timeout):
if "metadata.vlaanderen.be" in request.full_url and "7C823055" in request.full_url:
return _RedirectedResponse(
_metadata("7C823055", "GRBgis", "Toestand 2026-07-15", "2026-07-15", "2026-07-15"),
"https://example.com/redirected-metadata",
)
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=redirected_opener, now=NOW
)
grb = next(item for item in report.items if item.source_name == "grb")
assert grb.status == "unavailable"
assert grb.error_code == "CATALOG_METADATA_URL_REJECTED"
def test_catalog_probe_cache_is_explicitly_bypassable() -> None:
calls: list[str] = []
def counting_opener(request, timeout):
calls.append(request.full_url)
return _opener(request, timeout)
SourceCatalogProbeService.clear_cache()
settings = _settings(SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS=900)
db = _Db([])
project_id = uuid.uuid4()
first = SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW)
first_call_count = len(calls)
second = SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW)
SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW, force=True)
assert first_call_count == 6
assert all(item.cached is False for item in first.items)
assert all(item.cached is True for item in second.items)
assert len(calls) == 12
def test_catalog_probe_can_be_disabled_without_network_access() -> None:
def forbidden_opener(*_args, **_kwargs):
raise AssertionError("network must not be called")
report = SourceCatalogProbeService.audit_project(
_Db([]),
uuid.uuid4(),
settings=_settings(SOURCE_CATALOG_PROBE_ENABLED=False),
opener=forbidden_opener,
now=NOW,
)
assert report.summary.disabled_count == 4
assert all(item.status == "disabled" for item in report.items)
def test_catalog_probe_route_returns_canonical_envelope(monkeypatch) -> None:
from app.api.routes import datasets as dataset_routes
project_id = uuid.uuid4()
expected = SourceCatalogProbeService.audit_project(
_Db([]), project_id, settings=_settings(SOURCE_CATALOG_PROBE_ENABLED=False), now=NOW
)
monkeypatch.setattr(
dataset_routes.SourceCatalogProbeService,
"audit_project",
lambda db, selected_project_id, force=False: expected,
)
response = dataset_routes.probe_dataset_source_catalogs(
project_id=project_id, refresh=False, db=SimpleNamespace()
)
assert list(response) == ["data"]
assert response["data"]["project_id"] == project_id
assert response["data"]["summary"]["provider_count"] == 4
def test_catalog_probe_remains_explicit_and_never_imports_provider_data() -> None:
root = Path(__file__).resolve().parents[2]
hook = (root / "frontend" / "src" / "hooks" / "useSourceFreshness.ts").read_text(encoding="utf-8")
service = (root / "backend" / "app" / "services" / "source_catalog_probe_service.py").read_text(encoding="utf-8")
operator = (root / "scripts" / "audit_source_freshness.py").read_text(encoding="utf-8")
assert "void probeCatalogs(" not in hook
assert "DatasetService" not in service
assert "VectorFeatureService" not in service
assert "--probe-catalogs" in operator
assert "/datasets/source-catalog-probes" in operator
def test_alz_catalog_probe_rejects_release_page_redirect_outside_allowlist() -> None:
def redirected_opener(request, timeout):
if request.full_url == ALZ_RELEASE_URL:
return _RedirectedResponse(
_alz_release_page(),
"https://example.com/open-geodata-landbouwgebruikspercelen",
"text/html",
)
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=redirected_opener, now=NOW
)
alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
assert alz.status == "unavailable"
assert alz.error_code == "CATALOG_ALZ_RELEASE_URL_REJECTED"
def test_alz_catalog_probe_rejects_untrusted_download_host() -> None:
def malicious_opener(request, timeout):
if request.full_url == ALZ_RELEASE_URL:
return _Response(_alz_release_page(download_host="example.com"), "text/html")
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([]), uuid.uuid4(), settings=_settings(), opener=malicious_opener, now=NOW
)
alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
assert alz.status == "unavailable"
assert alz.error_code == "CATALOG_ALZ_DOWNLOAD_URL_REJECTED"
def test_alz_catalog_probe_degrades_without_current_snapshot_but_keeps_definitive_evidence() -> None:
def archive_only_opener(request, timeout):
if request.full_url == ALZ_RELEASE_URL:
return _Response(_alz_release_page(include_snapshot=False), "text/html")
return _opener(request, timeout)
report = SourceCatalogProbeService.audit_project(
_Db([_dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2024-definitive")]),
uuid.uuid4(),
settings=_settings(),
opener=archive_only_opener,
now=NOW,
)
alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
assert alz.status == "degraded"
assert alz.remote_version == "2025-v3"
assert alz.comparison_status == "different"
assert alz.matched_layers == ["definitive_archive"]
assert alz.missing_layers == ["current_snapshot"]
assert alz.error_code == "CATALOG_ALZ_CURRENT_SNAPSHOT_MISSING"
def test_alz_catalog_probe_compares_latest_definitive_year_not_latest_import_order() -> None:
older = _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2024-definitive")
newer = _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2025-definitive")
older.imported_at = NOW
newer.imported_at = NOW.replace(year=2025)
report = SourceCatalogProbeService.audit_project(
_Db([older, newer]), uuid.uuid4(), settings=_settings(), opener=_opener, now=NOW
)
alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
assert alz.local_source_version == "2025-definitive"
assert alz.comparison_status == "same"