Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
477 lines
20 KiB
Python
477 lines
20 KiB
Python
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"""
|
||
<wfs:FeatureType>
|
||
<wfs:Name>GRB:{name}</wfs:Name>
|
||
<wfs:MetadataURL xlink:href="{metadata_url.replace('&', '&')}" />
|
||
</wfs:FeatureType>
|
||
"""
|
||
for name in ("GBG", "WBN", "WGO", "ADP", "WTZ")
|
||
)
|
||
return f"""
|
||
<wfs:WFS_Capabilities xmlns:wfs="http://www.opengis.net/wfs/2.0"
|
||
xmlns:xlink="http://www.w3.org/1999/xlink">
|
||
<wfs:FeatureTypeList>{layers}</wfs:FeatureTypeList>
|
||
</wfs:WFS_Capabilities>
|
||
""".encode()
|
||
|
||
|
||
def _wms_capabilities(metadata_url: str = ORTHO_METADATA_URL) -> bytes:
|
||
return f"""
|
||
<WMS_Capabilities xmlns="http://www.opengis.net/wms"
|
||
xmlns:xlink="http://www.w3.org/1999/xlink">
|
||
<Capability><Layer><Title>Orthofoto</Title>
|
||
<Layer><Name>Ortho</Name><MetadataURL><Format>text/xml</Format>
|
||
<OnlineResource xlink:href="{metadata_url.replace('&', '&')}" />
|
||
</MetadataURL></Layer>
|
||
<Layer><Name>Vliegdagcontour</Name><MetadataURL><Format>text/xml</Format>
|
||
<OnlineResource xlink:href="{metadata_url.replace('&', '&')}" />
|
||
</MetadataURL></Layer>
|
||
</Layer></Capability>
|
||
</WMS_Capabilities>
|
||
""".encode()
|
||
|
||
|
||
def _metadata(identifier: str, title: str, edition: str, modified: str, published: str) -> bytes:
|
||
return f"""
|
||
<csw:GetRecordByIdResponse xmlns:csw="http://www.opengis.net/cat/csw/2.0.2"
|
||
xmlns:gmd="http://www.isotc211.org/2005/gmd"
|
||
xmlns:gco="http://www.isotc211.org/2005/gco">
|
||
<gmd:MD_Metadata>
|
||
<gmd:fileIdentifier><gco:CharacterString>{identifier}</gco:CharacterString></gmd:fileIdentifier>
|
||
<gmd:dateStamp><gco:Date>{modified}</gco:Date></gmd:dateStamp>
|
||
<gmd:identificationInfo><gmd:MD_DataIdentification><gmd:citation><gmd:CI_Citation>
|
||
<gmd:title><gco:CharacterString>{title}</gco:CharacterString></gmd:title>
|
||
<gmd:edition><gco:CharacterString>{edition}</gco:CharacterString></gmd:edition>
|
||
<gmd:date><gmd:CI_Date><gmd:date><gco:Date>{published}</gco:Date></gmd:date>
|
||
<gmd:dateType><gmd:CI_DateTypeCode codeListValue="publication">publication</gmd:CI_DateTypeCode></gmd:dateType>
|
||
</gmd:CI_Date></gmd:date>
|
||
</gmd:CI_Citation></gmd:citation></gmd:MD_DataIdentification></gmd:identificationInfo>
|
||
</gmd:MD_Metadata>
|
||
</csw:GetRecordByIdResponse>
|
||
""".encode()
|
||
|
||
|
||
def _alz_release_page(*, include_snapshot: bool = True, download_host: str = "www.landbouwvlaanderen.be") -> bytes:
|
||
snapshot = (
|
||
f'<a href="https://{download_host}/bestanden/gis/agpa_2026_2026-06-02_public.zip">'
|
||
"Landbouwgebruikspercelen 2026 – 1e snapshot (extractie 02-06-2026) - GPKG</a>"
|
||
if include_snapshot
|
||
else ""
|
||
)
|
||
return f"""
|
||
<html><body>
|
||
{snapshot}
|
||
<p>Definitieve datasets</p>
|
||
<a href="https://{download_host}/bestanden/gis/agpa_2025_2026-05-13_public.zip">Downloaden</a>
|
||
<a href="https://{download_host}/bestanden/gis/agpa_2024_2025-03-27_public.zip">Downloaden</a>
|
||
</body></html>
|
||
""".encode()
|
||
|
||
|
||
def _statbel_dcat() -> bytes:
|
||
return b"""
|
||
@prefix dcat: <http://www.w3.org/ns/dcat#> .
|
||
@prefix dct: <http://purl.org/dc/terms/> .
|
||
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
|
||
|
||
<http://data.gov.be/catalog/statbelopen> a dcat:Catalog ;
|
||
dct:modified "2026-07-07"^^xsd:date .
|
||
|
||
<https://statbel.fgov.be/node/6475#id> a dcat:Dataset ;
|
||
dct:title "Bevolking per statistische sector"@nl ;
|
||
dct:alternative "Bevolking per statistische sector [Periode: 2025]"@nl ;
|
||
dct:identifier "NodeID6475" ;
|
||
dct:license <https://creativecommons.org/licenses/by/4.0/> ;
|
||
dct:temporal [ dcat:startDate "2025-01-01"^^xsd:date ] ;
|
||
dcat:landingPage <https://statbel.fgov.be/nl/open-data/bevolking-statistische-sector-12> ;
|
||
dcat:distribution
|
||
<https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.zip#distribution6475>,
|
||
<https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_NEW.xlsx#distribution6475>,
|
||
<https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_OLD.zip#distribution6475>,
|
||
<https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/OPENDATA_SECTOREN_2025_OLD.xlsx#distribution6475> .
|
||
"""
|
||
|
||
|
||
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"<xml />", 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"
|