169 lines
6.2 KiB
Python
169 lines
6.2 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
|
|
|
|
|
|
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_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 = (root / "frontend" / "src" / "App.tsx").read_text(encoding="utf-8")
|
|
api = (root / "frontend" / "src" / "services" / "api" / "datasets.ts").read_text(encoding="utf-8")
|
|
|
|
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
|