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
570 lines
20 KiB
Python
570 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS = ROOT / "scripts"
|
|
|
|
|
|
def load_script(name: str):
|
|
path = SCRIPTS / name
|
|
if str(SCRIPTS) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
spec = importlib.util.spec_from_file_location(f"rc10_{path.stem}", path)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def write_backup(root: Path, *, created_at: datetime, inventory_mode: str = "sha256") -> None:
|
|
root.mkdir(parents=True)
|
|
manifest = {
|
|
"schema_version": 1,
|
|
"release_id": "rc10-test",
|
|
"created_at": created_at.isoformat(),
|
|
"read_only_source": True,
|
|
"database_password_secure": True,
|
|
"inventory_mode": inventory_mode,
|
|
"storage_inventory_requested": True,
|
|
"storage_snapshot_requested": True,
|
|
"models_inventory_requested": False,
|
|
"models_snapshot_requested": False,
|
|
"git_commit": "0123456789abcdef",
|
|
}
|
|
files = {
|
|
"manifest.json": json.dumps(manifest),
|
|
"database.dump": "database",
|
|
"database.list": "list",
|
|
"database-metadata.tsv": "alembic_head\t202607160001",
|
|
"table-counts.tsv": "datasets\t1",
|
|
"storage-manifest.tsv": "relative_path\tsize_bytes\tmtime_ns\tsha256",
|
|
}
|
|
for name, content in files.items():
|
|
(root / name).write_text(content, encoding="utf-8")
|
|
(root / "storage-snapshot").mkdir()
|
|
checksums = []
|
|
for name in sorted(files):
|
|
digest = hashlib.sha256((root / name).read_bytes()).hexdigest()
|
|
checksums.append(f"{digest} {name}")
|
|
(root / "CHECKSUMS.sha256").write_text("\n".join(checksums) + "\n", encoding="utf-8")
|
|
|
|
|
|
def test_backup_guard_requires_recent_complete_sha256_storage_backup(tmp_path: Path) -> None:
|
|
guard = load_script("release_backup_guard.py")
|
|
now = datetime(2026, 7, 18, 12, tzinfo=timezone.utc)
|
|
backup = tmp_path / "backup"
|
|
write_backup(backup, created_at=now - timedelta(hours=2))
|
|
|
|
verified = guard.verify_current_backup(backup, now=now)
|
|
|
|
assert verified.release_id == "rc10-test"
|
|
assert verified.age_hours == pytest.approx(2)
|
|
|
|
|
|
def test_backup_guard_rejects_stale_or_tampered_backup(tmp_path: Path) -> None:
|
|
guard = load_script("release_backup_guard.py")
|
|
now = datetime(2026, 7, 18, 12, tzinfo=timezone.utc)
|
|
stale = tmp_path / "stale"
|
|
write_backup(stale, created_at=now - timedelta(hours=30))
|
|
with pytest.raises(RuntimeError, match="maximum allowed age"):
|
|
guard.verify_current_backup(stale, now=now)
|
|
|
|
current = tmp_path / "tampered"
|
|
write_backup(current, created_at=now)
|
|
(current / "database.dump").write_text("tampered", encoding="utf-8")
|
|
with pytest.raises(RuntimeError, match="checksum mismatch"):
|
|
guard.verify_current_backup(current, now=now)
|
|
|
|
|
|
def test_storage_lifecycle_is_fail_closed_and_protects_release_evidence() -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
|
|
assert audit.classify_relative_path("release-evidence/rc11/manifest.json") == (
|
|
"release-evidence",
|
|
True,
|
|
False,
|
|
)
|
|
assert audit.classify_relative_path("operator-evidence/source/raw.json")[1:] == (True, False)
|
|
assert audit.classify_relative_path("uploads/project/data.geojson")[1:] == (True, False)
|
|
assert audit.classify_relative_path("exports/project/old.json")[1:] == (False, True)
|
|
assert audit.classify_relative_path("unknown/value.bin")[1:] == (True, False)
|
|
|
|
|
|
def test_storage_audit_only_selects_old_unreferenced_allowlisted_files(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
storage = tmp_path / "storage"
|
|
old_orphan = storage / "exports" / "project" / "old.json"
|
|
referenced = storage / "exports" / "project" / "kept.json"
|
|
protected = storage / "release-evidence" / "rc" / "manifest.json"
|
|
unknown = storage / "misc" / "unknown.bin"
|
|
for path in (old_orphan, referenced, protected, unknown):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(path.name, encoding="utf-8")
|
|
old_timestamp = (datetime.now(timezone.utc) - timedelta(days=30)).timestamp()
|
|
for path in (old_orphan, referenced, protected, unknown):
|
|
path.touch()
|
|
Path(path).chmod(0o644)
|
|
import os
|
|
|
|
os.utime(path, (old_timestamp, old_timestamp))
|
|
|
|
monkeypatch.setattr(
|
|
audit,
|
|
"collect_database_state",
|
|
lambda _db, _root: {
|
|
"references": {referenced.resolve()},
|
|
"counts": {},
|
|
"source_families": {"national": [], "regional": [], "maritime": []},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
audit,
|
|
"disk_pressure",
|
|
lambda _root: {
|
|
"status": "ok",
|
|
"total_bytes": 100,
|
|
"used_bytes": 50,
|
|
"free_bytes": 50,
|
|
"free_percent": 50.0,
|
|
"acquisition_allowed": True,
|
|
},
|
|
)
|
|
|
|
report, candidates = audit.build_report(storage, SimpleNamespace(), minimum_age_days=7)
|
|
|
|
assert [candidate.relative_path for candidate in candidates] == ["exports/project/old.json"]
|
|
assert report["cleanup"]["candidate_count"] == 1
|
|
assert "release-evidence" in report["cleanup"]["protected_prefixes"]
|
|
assert report["integrity"]["missing_referenced_path_count"] == 0
|
|
assert report["integrity"]["missing_manifest_artifact_count"] == 0
|
|
|
|
|
|
def test_referenced_tile_manifest_protects_its_tiles(tmp_path: Path) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
storage = tmp_path / "storage"
|
|
manifest = storage / "tiles" / "dataset" / "set" / "manifest.json"
|
|
tile = manifest.parent / "tile_0000.tif"
|
|
tile.parent.mkdir(parents=True)
|
|
tile.write_bytes(b"tile")
|
|
manifest.write_text(json.dumps({"tiles": [{"path": "tile_0000.tif"}]}), encoding="utf-8")
|
|
|
|
expanded = audit.expand_manifest_references({manifest.resolve()}, storage)
|
|
|
|
assert manifest.resolve() in expanded
|
|
assert tile.resolve() in expanded
|
|
|
|
|
|
def test_ordinary_json_export_is_not_treated_as_an_artifact_manifest(tmp_path: Path) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
storage = tmp_path / "storage"
|
|
export = storage / "exports" / "project" / "report.json"
|
|
export.parent.mkdir(parents=True)
|
|
export.write_text(json.dumps({"dataset_id": "not-a-file"}), encoding="utf-8")
|
|
|
|
expanded = audit.expand_manifest_references({export.resolve()}, storage)
|
|
|
|
assert expanded == {export.resolve()}
|
|
|
|
|
|
def test_manifest_ids_dates_and_labels_are_not_treated_as_paths(tmp_path: Path) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
storage = tmp_path / "storage"
|
|
manifest = storage / "operator-data" / "scope" / "scope.manifest.json"
|
|
actual = manifest.parent / "scope.geojson"
|
|
manifest.parent.mkdir(parents=True)
|
|
actual.write_text("{}", encoding="utf-8")
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"municipality_ids": ["13025", "11001"],
|
|
"generated_at": "2026-07-18T00:00:00Z",
|
|
"label": "Belgium",
|
|
"output_path": "scope.geojson",
|
|
"output_checksum_sha256": "a" * 64,
|
|
"output_crs": "EPSG:4326",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
expanded = audit.expand_manifest_references({manifest.resolve()}, storage)
|
|
|
|
assert expanded == {manifest.resolve(), actual.resolve()}
|
|
|
|
|
|
def test_disk_pressure_uses_absolute_headroom_for_large_arrays(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
monkeypatch.setattr(
|
|
audit.shutil,
|
|
"disk_usage",
|
|
lambda _path: SimpleNamespace(
|
|
total=56 * 1024**4,
|
|
used=(56 * 1024**4) - (700 * 1024**3),
|
|
free=700 * 1024**3,
|
|
),
|
|
)
|
|
|
|
pressure = audit.disk_pressure(tmp_path)
|
|
|
|
assert pressure["free_percent"] < 2
|
|
assert pressure["status"] == "ok"
|
|
assert pressure["acquisition_allowed"] is True
|
|
|
|
|
|
def test_source_family_report_covers_national_regional_and_maritime(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
audit = load_script("audit_data_operations.py")
|
|
national_id = uuid4()
|
|
regional_id = uuid4()
|
|
now = datetime.now(timezone.utc)
|
|
rows = {
|
|
audit.Project: [
|
|
SimpleNamespace(id=national_id, name=audit.NATIONAL_PROJECT_NAME, status="active"),
|
|
SimpleNamespace(id=regional_id, name="Wallonia operator", status="active"),
|
|
],
|
|
audit.Dataset: [
|
|
SimpleNamespace(
|
|
project_id=national_id,
|
|
source_name="ngi_adminvector",
|
|
source="ngi",
|
|
name="Belgium boundary",
|
|
source_metadata={"coverage_zones": ["belgium"]},
|
|
source_version="2026",
|
|
imported_at=now,
|
|
status="ready",
|
|
storage_path=None,
|
|
metadata_json=None,
|
|
provenance_metadata=None,
|
|
),
|
|
SimpleNamespace(
|
|
project_id=national_id,
|
|
source_name="rbins_marine_reporting_units",
|
|
source="rbins",
|
|
name="Belgian North Sea",
|
|
source_metadata={"coverage_zones": ["belgian_north_sea"]},
|
|
source_version="2024",
|
|
imported_at=now,
|
|
status="ready",
|
|
storage_path=None,
|
|
metadata_json=None,
|
|
provenance_metadata=None,
|
|
),
|
|
SimpleNamespace(
|
|
project_id=regional_id,
|
|
source_name="wallonia_manual",
|
|
source="manual",
|
|
name="Wallonia source",
|
|
source_metadata={},
|
|
source_version="1",
|
|
imported_at=now,
|
|
status="ready",
|
|
storage_path=None,
|
|
metadata_json=None,
|
|
provenance_metadata=None,
|
|
),
|
|
],
|
|
audit.DatasetVersion: [],
|
|
audit.Export: [],
|
|
audit.Detection: [],
|
|
audit.Segmentation: [],
|
|
audit.Job: [],
|
|
audit.AnalysisRun: [],
|
|
}
|
|
|
|
class Query:
|
|
def __init__(self, values):
|
|
self.values = values
|
|
|
|
def all(self):
|
|
return self.values
|
|
|
|
class Session:
|
|
def query(self, model, *_fields):
|
|
if model in rows:
|
|
return Query(rows[model])
|
|
owner = getattr(model, "class_", None)
|
|
if owner in rows:
|
|
return Query(rows[owner])
|
|
raise AssertionError(f"Unexpected query entity: {model!r}")
|
|
|
|
monkeypatch.setattr(audit, "query_count", lambda _db, model, *_conditions: len(rows[model]))
|
|
monkeypatch.setattr(audit, "query_distinct_nonnull", lambda _db, _column: [])
|
|
state = audit.collect_database_state(Session(), tmp_path)
|
|
|
|
assert {item["source_name"] for item in state["source_families"]["national"]} == {
|
|
"ngi_adminvector",
|
|
"rbins_marine_reporting_units",
|
|
}
|
|
assert {item["source_name"] for item in state["source_families"]["maritime"]} == {
|
|
"rbins_marine_reporting_units"
|
|
}
|
|
assert {item["source_name"] for item in state["source_families"]["regional"]} == {
|
|
"wallonia_manual"
|
|
}
|
|
|
|
|
|
def test_national_and_maritime_sources_have_explicit_freshness_policies() -> None:
|
|
from app.services.source_freshness_service import SOURCE_POLICIES
|
|
|
|
for source_name in (
|
|
"ngi_adminvector",
|
|
"rbins_marine_reporting_units",
|
|
"rbins_msp_2026",
|
|
):
|
|
assert SOURCE_POLICIES[source_name].refresh_policy == "edition"
|
|
|
|
|
|
def test_cleanup_commands_require_backup_confirmation_and_read_only_mount() -> None:
|
|
generic = (SCRIPTS / "cleanup_storage_artifacts.py").read_text(encoding="utf-8")
|
|
demo = (ROOT / "backend/scripts/cleanup_demo_artifacts.py").read_text(encoding="utf-8")
|
|
compose = (ROOT / "docker-compose.unraid.yml").read_text(encoding="utf-8")
|
|
dockerman = (ROOT / "deploy/unraid/run-dockerman-container.sh").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")
|
|
live_audit = (SCRIPTS / "run_rc10_data_operations_audit.sh").read_text(encoding="utf-8")
|
|
|
|
assert "QUARANTINE_STORAGE_ARTIFACTS" in generic
|
|
assert "verify_current_backup" in generic
|
|
assert "os.link" in generic
|
|
assert 'entry["status"] = "linked"' in generic
|
|
assert "cleanup-quarantine" in generic
|
|
assert "DELETE_DEMO_EXPORTS" in demo
|
|
assert "verify_current_backup" in demo
|
|
assert "/app/backups:ro" in compose
|
|
assert '/app/backups:ro"' in dockerman
|
|
for name in (
|
|
"release_backup_guard.py",
|
|
"audit_data_operations.py",
|
|
"cleanup_storage_artifacts.py",
|
|
"restore_storage_quarantine.py",
|
|
"release_backup_snapshot.py",
|
|
):
|
|
assert f"COPY scripts/{name}" in dockerfile
|
|
assert f"py_compile scripts/{name}" in readiness
|
|
assert "bash -n scripts/run_rc10_data_operations_audit.sh" in readiness
|
|
assert "--apply" not in live_audit
|
|
assert "table-counts-before.tsv" in live_audit
|
|
assert "table-counts-after.tsv" in live_audit
|
|
assert "deleted_count" in live_audit
|
|
assert "missing_manifest_artifact_count" in live_audit
|
|
|
|
|
|
def test_cleanup_apply_moves_bytes_to_protected_traceable_quarantine(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
capsys: pytest.CaptureFixture[str],
|
|
) -> None:
|
|
monkeypatch.syspath_prepend(str(SCRIPTS))
|
|
cleanup = load_script("cleanup_storage_artifacts.py")
|
|
storage = tmp_path / "storage"
|
|
source = storage / "derived" / "orphan.bin"
|
|
source.parent.mkdir(parents=True)
|
|
source.write_bytes(b"recoverable-derived-artifact")
|
|
candidate = SimpleNamespace(
|
|
path=source.resolve(),
|
|
relative_path="derived/orphan.bin",
|
|
size_bytes=source.stat().st_size,
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
class SessionContext:
|
|
def __enter__(self):
|
|
return SimpleNamespace()
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
monkeypatch.setattr(
|
|
cleanup,
|
|
"parse_args",
|
|
lambda: SimpleNamespace(
|
|
storage_root=storage,
|
|
minimum_age_days=7,
|
|
max_delete=1,
|
|
apply=True,
|
|
confirm="QUARANTINE_STORAGE_ARTIFACTS",
|
|
backup_dir=tmp_path / "backup",
|
|
backup_max_age_hours=24.0,
|
|
quarantine_root=None,
|
|
),
|
|
)
|
|
monkeypatch.setattr(cleanup, "SessionLocal", lambda: SessionContext())
|
|
monkeypatch.setattr(
|
|
cleanup,
|
|
"build_report",
|
|
lambda *_args, **_kwargs: ({"cleanup": {"protected_prefixes": []}}, [candidate]),
|
|
)
|
|
monkeypatch.setattr(
|
|
cleanup,
|
|
"verify_current_backup",
|
|
lambda *_args, **_kwargs: SimpleNamespace(
|
|
release_id="predeploy-test",
|
|
created_at=now,
|
|
age_hours=0.1,
|
|
backup_tool_revision="0123456789abcdef",
|
|
),
|
|
)
|
|
|
|
assert cleanup.main() == 0
|
|
payload = json.loads(capsys.readouterr().out)
|
|
manifest_path = Path(payload["quarantine_manifest"])
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
quarantined_path = storage / payload["quarantined"][0]["quarantine_relative_path"]
|
|
|
|
assert not source.exists()
|
|
assert quarantined_path.read_bytes() == b"recoverable-derived-artifact"
|
|
assert manifest["state"] == "complete"
|
|
assert manifest["backup_release_id"] == "predeploy-test"
|
|
assert manifest["entries"][0]["status"] == "quarantined"
|
|
assert payload["deleted_count"] == 0
|
|
|
|
restore = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPTS / "restore_storage_quarantine.py"),
|
|
"--storage-root",
|
|
str(storage),
|
|
"--manifest",
|
|
str(manifest_path),
|
|
"--confirm",
|
|
"RESTORE_QUARANTINED_ARTIFACTS",
|
|
],
|
|
cwd=ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
assert restore.returncode == 0, restore.stderr
|
|
assert source.read_bytes() == b"recoverable-derived-artifact"
|
|
assert not quarantined_path.exists()
|
|
restored_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert restored_manifest["state"] == "restored"
|
|
assert restored_manifest["entries"][0]["status"] == "restored"
|
|
|
|
|
|
def _write_interrupted_quarantine(
|
|
storage: Path,
|
|
*,
|
|
original_exists: bool,
|
|
quarantine_exists: bool,
|
|
hard_linked: bool = False,
|
|
) -> tuple[Path, Path, Path]:
|
|
original = storage / "derived" / "interrupted.bin"
|
|
operation = storage / "operator-evidence" / "cleanup-quarantine" / "cleanup-interrupted"
|
|
quarantined = operation / "files" / "derived" / "interrupted.bin"
|
|
original.parent.mkdir(parents=True, exist_ok=True)
|
|
quarantined.parent.mkdir(parents=True, exist_ok=True)
|
|
retained = b"interrupted-retained-bytes"
|
|
if original_exists:
|
|
original.write_bytes(retained)
|
|
if quarantine_exists:
|
|
if hard_linked:
|
|
os.link(original, quarantined)
|
|
else:
|
|
quarantined.write_bytes(retained)
|
|
manifest = operation / "manifest.json"
|
|
manifest.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"state": "in_progress",
|
|
"entries": [
|
|
{
|
|
"relative_path": "derived/interrupted.bin",
|
|
"quarantine_relative_path": quarantined.relative_to(storage).as_posix(),
|
|
"size_bytes": len(retained),
|
|
"sha256": hashlib.sha256(retained).hexdigest(),
|
|
"status": "linked" if hard_linked else "planned",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return manifest, original, quarantined
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("original_exists", "quarantine_exists", "hard_linked"),
|
|
((False, True, False), (True, True, True)),
|
|
)
|
|
def test_quarantine_restore_recovers_each_interrupted_move_window(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
original_exists: bool,
|
|
quarantine_exists: bool,
|
|
hard_linked: bool,
|
|
) -> None:
|
|
restore = load_script("restore_storage_quarantine.py")
|
|
storage = tmp_path / "storage"
|
|
manifest, original, quarantined = _write_interrupted_quarantine(
|
|
storage,
|
|
original_exists=original_exists,
|
|
quarantine_exists=quarantine_exists,
|
|
hard_linked=hard_linked,
|
|
)
|
|
monkeypatch.setattr(
|
|
restore,
|
|
"parse_args",
|
|
lambda: SimpleNamespace(
|
|
storage_root=storage,
|
|
manifest=manifest,
|
|
confirm="RESTORE_QUARANTINED_ARTIFACTS",
|
|
),
|
|
)
|
|
|
|
assert restore.main() == 0
|
|
assert original.read_bytes() == b"interrupted-retained-bytes"
|
|
assert not quarantined.exists()
|
|
assert json.loads(manifest.read_text(encoding="utf-8"))["state"] == "restored"
|
|
|
|
|
|
def test_quarantine_restore_never_clobbers_recreated_destination(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
restore = load_script("restore_storage_quarantine.py")
|
|
storage = tmp_path / "storage"
|
|
manifest, original, quarantined = _write_interrupted_quarantine(
|
|
storage,
|
|
original_exists=False,
|
|
quarantine_exists=True,
|
|
)
|
|
original.write_bytes(b"new-runtime-bytes")
|
|
monkeypatch.setattr(
|
|
restore,
|
|
"parse_args",
|
|
lambda: SimpleNamespace(
|
|
storage_root=storage,
|
|
manifest=manifest,
|
|
confirm="RESTORE_QUARANTINED_ARTIFACTS",
|
|
),
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="different bytes"):
|
|
restore.main()
|
|
assert original.read_bytes() == b"new-runtime-bytes"
|
|
assert quarantined.read_bytes() == b"interrupted-retained-bytes"
|