JensandClaude Opus 5
6572e4ad5f
93 test files read a single frontend source and asserted identifiers in it. The
MapWorkspace split showed what that costs: 24 tests went red for a move that
changed no behaviour at all. A contract belongs to the feature — a container,
its hooks, its domain layer — not to whichever file currently holds it.
232 read sites now resolve through read_feature(). The distinction that makes
this safe is direction: a *positive* contract ("this is wired") may widen,
because the identifier must still exist somewhere in the feature; a *negative*
one ("this component performs no transport") is a statement about one file, and
widening it would quietly weaken the check. The 73 single-file reads that
remain are exactly those, and a guard now enforces the rule for new tests.
Verified rather than assumed: of the 732 migrated positive assertions, 644 still
match exactly one module — as specific as before — and the other 86 already
spanned a container and its hook by nature. Two apparent misses are an artefact
of the checking regex reading an escaped newline literally.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
245 lines
8.5 KiB
Python
245 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from app.models import Dataset, DatasetVersion
|
|
from app.services.source_freshness_service import SourceFreshnessService
|
|
from tests.frontend_contract import read_feature
|
|
|
|
|
|
NOW = datetime(2026, 7, 16, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _dataset(
|
|
source_name: str,
|
|
*,
|
|
imported_at: datetime | None = NOW,
|
|
observed_at: datetime | None = None,
|
|
source_version: str | None = "edition-1",
|
|
storage_path: str | None = None,
|
|
checksum: str | None = "abc",
|
|
size_bytes: int | None = None,
|
|
temporal_series_key: str | None = None,
|
|
) -> Dataset:
|
|
return Dataset(
|
|
id=uuid.uuid4(),
|
|
project_id=uuid.uuid4(),
|
|
name=f"{source_name} dataset",
|
|
dataset_type="vector",
|
|
source=source_name,
|
|
source_name=source_name,
|
|
imported_at=imported_at,
|
|
observed_at=observed_at,
|
|
source_version=source_version,
|
|
storage_path=storage_path,
|
|
checksum_sha256=checksum,
|
|
size_bytes=size_bytes,
|
|
temporal_series_key=temporal_series_key,
|
|
status="ready",
|
|
)
|
|
|
|
|
|
def _version(dataset: Dataset, *, checksum: str | None = "abc") -> DatasetVersion:
|
|
return DatasetVersion(
|
|
id=uuid.uuid4(),
|
|
dataset_id=dataset.id,
|
|
version=1,
|
|
source_version=dataset.source_version,
|
|
observed_at=dataset.observed_at,
|
|
checksum_sha256=checksum,
|
|
)
|
|
|
|
|
|
def test_source_freshness_distinguishes_snapshot_annual_edition_and_local_sources(tmp_path: Path) -> None:
|
|
existing_file = tmp_path / "snapshot.geojson"
|
|
existing_file.write_text("{}", encoding="utf-8")
|
|
fresh_grb = _dataset(
|
|
"grb",
|
|
imported_at=NOW - timedelta(days=20),
|
|
storage_path=str(existing_file),
|
|
size_bytes=2,
|
|
)
|
|
old_vrbg = _dataset("vrbg", imported_at=NOW - timedelta(days=120))
|
|
current_annual = _dataset(
|
|
"statbel",
|
|
observed_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="population",
|
|
source_version="2025",
|
|
)
|
|
old_annual = _dataset(
|
|
"waterinfo",
|
|
observed_at=datetime(2023, 1, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="water-level",
|
|
source_version="2023",
|
|
)
|
|
fixed_scenario = _dataset("vmm_flood_hazard", observed_at=None, source_version="VMM OGRK")
|
|
manual = _dataset("manual", source_version=None, checksum=None)
|
|
datasets = [fresh_grb, old_vrbg, current_annual, old_annual, fixed_scenario, manual]
|
|
versions = [_version(dataset, checksum=dataset.checksum_sha256) for dataset in datasets]
|
|
|
|
report = SourceFreshnessService.build_report(fresh_grb.project_id, datasets, versions, now=NOW)
|
|
by_source = {item.source_name: item for item in report.items}
|
|
|
|
assert by_source["grb"].status == "current"
|
|
assert by_source["vrbg"].status == "due"
|
|
assert by_source["statbel"].status == "current"
|
|
assert by_source["waterinfo"].status == "due"
|
|
assert by_source["vmm_flood_hazard"].status == "current"
|
|
assert by_source["manual"].status == "local"
|
|
assert all(item.auto_refresh_supported is False for item in report.items)
|
|
assert report.summary.dataset_count == len(datasets)
|
|
|
|
|
|
def test_source_freshness_flags_local_version_and_storage_integrity(tmp_path: Path) -> None:
|
|
missing_file = tmp_path / "missing.tif"
|
|
dataset = _dataset(
|
|
"digitaal_vlaanderen_dhmv",
|
|
storage_path=str(missing_file),
|
|
checksum="dataset-checksum",
|
|
)
|
|
version = _version(dataset, checksum="different-version-checksum")
|
|
|
|
report = SourceFreshnessService.build_report(dataset.project_id, [dataset], [version], now=NOW)
|
|
item = report.items[0]
|
|
|
|
assert item.status == "review_required"
|
|
assert item.integrity.checksum_mismatch_count == 1
|
|
assert item.integrity.missing_storage_file_count == 1
|
|
assert report.summary.sources_with_integrity_issues == 1
|
|
assert report.summary.integrity_issue_count == 2
|
|
|
|
|
|
def test_source_freshness_requires_dataset_version_and_marks_temporal_series() -> None:
|
|
first = _dataset(
|
|
"department_omgeving_land_use",
|
|
observed_at=datetime(2022, 1, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="land-use",
|
|
source_version="2022-v3",
|
|
)
|
|
second = _dataset(
|
|
"department_omgeving_land_use",
|
|
observed_at=datetime(2025, 1, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="land-use",
|
|
source_version="2025-v3",
|
|
)
|
|
|
|
report = SourceFreshnessService.build_report(first.project_id, [first, second], [_version(first)], now=NOW)
|
|
item = report.items[0]
|
|
|
|
assert item.historical_series is True
|
|
assert item.status == "review_required"
|
|
assert item.integrity.missing_version_count == 1
|
|
|
|
|
|
def test_rolling_orthophoto_prefers_explicit_current_snapshot_over_historical_observation() -> None:
|
|
current = _dataset(
|
|
"digitaal_vlaanderen_orthophoto",
|
|
imported_at=NOW - timedelta(days=2),
|
|
observed_at=None,
|
|
source_version="most_recent_at_2026-07-14",
|
|
)
|
|
historical = _dataset(
|
|
"digitaal_vlaanderen_orthophoto",
|
|
imported_at=NOW - timedelta(days=1),
|
|
observed_at=datetime(2020, 6, 1, tzinfo=timezone.utc),
|
|
source_version="2020",
|
|
)
|
|
|
|
report = SourceFreshnessService.build_report(
|
|
current.project_id,
|
|
[current, historical],
|
|
[_version(current), _version(historical)],
|
|
now=NOW,
|
|
)
|
|
|
|
assert report.items[0].latest_source_version == "most_recent_at_2026-07-14"
|
|
|
|
|
|
def test_orthophoto_freshness_prefers_governed_official_edition_over_rolling_marker() -> None:
|
|
legacy = _dataset(
|
|
"digitaal_vlaanderen_orthophoto",
|
|
imported_at=NOW - timedelta(days=2),
|
|
observed_at=datetime(2026, 7, 15, tzinfo=timezone.utc),
|
|
source_version="most_recent_at_2026-07-15",
|
|
)
|
|
official = _dataset(
|
|
"digitaal_vlaanderen_orthophoto",
|
|
imported_at=NOW - timedelta(days=1),
|
|
observed_at=datetime(2025, 4, 5, tzinfo=timezone.utc),
|
|
source_version="2025.04",
|
|
)
|
|
|
|
report = SourceFreshnessService.build_report(
|
|
legacy.project_id,
|
|
[legacy, official],
|
|
[_version(legacy), _version(official)],
|
|
now=NOW,
|
|
)
|
|
|
|
item = report.items[0]
|
|
assert item.latest_source_version == "2025.04"
|
|
assert item.refresh_policy == "rolling_snapshot"
|
|
assert item.review_interval_days == 180
|
|
|
|
|
|
def test_spatial_partitions_do_not_become_a_false_historical_series() -> None:
|
|
first = _dataset(
|
|
"dov_soil_map",
|
|
observed_at=datetime(2017, 6, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="soil:mol",
|
|
source_version="2017",
|
|
)
|
|
second = _dataset(
|
|
"dov_soil_map",
|
|
observed_at=datetime(2017, 6, 1, tzinfo=timezone.utc),
|
|
temporal_series_key="soil:kempen",
|
|
source_version="2017",
|
|
)
|
|
|
|
report = SourceFreshnessService.build_report(
|
|
first.project_id,
|
|
[first, second],
|
|
[_version(first), _version(second)],
|
|
now=NOW,
|
|
)
|
|
|
|
assert report.items[0].historical_series is False
|
|
|
|
|
|
def test_source_freshness_route_returns_canonical_envelope(monkeypatch) -> None:
|
|
from app.api.routes import datasets as dataset_routes
|
|
|
|
project_id = uuid.uuid4()
|
|
expected = SourceFreshnessService.build_report(project_id, [], [], now=NOW)
|
|
monkeypatch.setattr(
|
|
dataset_routes.SourceFreshnessService,
|
|
"audit_project",
|
|
lambda db, selected_project_id: expected,
|
|
)
|
|
|
|
response = dataset_routes.audit_dataset_source_freshness(project_id=project_id, db=SimpleNamespace())
|
|
|
|
assert list(response) == ["data"]
|
|
assert response["data"]["project_id"] == project_id
|
|
assert response["data"]["summary"]["source_count"] == 0
|
|
|
|
|
|
def test_source_freshness_operator_and_ui_contract_are_read_only() -> None:
|
|
root = Path(__file__).resolve().parents[2]
|
|
script = (root / "scripts" / "audit_source_freshness.py").read_text(encoding="utf-8")
|
|
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")
|
|
app = read_feature("shell")
|
|
api = read_feature("datasets")
|
|
|
|
assert "Request(endpoint" in script
|
|
assert "method=\"POST\"" not in script
|
|
assert "urlopen(request" in script
|
|
assert "COPY scripts/audit_source_freshness.py" in dockerfile
|
|
assert "py_compile scripts/audit_source_freshness.py" in readiness
|
|
assert "<SourceFreshnessPanel" in app
|
|
assert "/datasets/source-freshness" in api
|