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
130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
|
|
|
|
def load_cleanup_module() -> ModuleType:
|
|
script = Path(__file__).resolve().parents[2] / "scripts" / "cleanup_demo_artifacts.py"
|
|
spec = importlib.util.spec_from_file_location("cleanup_demo_artifacts", script)
|
|
assert spec is not None
|
|
assert spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@dataclass
|
|
class ExportRow:
|
|
id: str
|
|
created_at: datetime | None
|
|
storage_path: str
|
|
export_type: str = "project_metadata_json"
|
|
|
|
|
|
def test_cleanup_candidate_selection_keeps_newest_exports() -> None:
|
|
cleanup = load_cleanup_module()
|
|
base = datetime(2026, 1, 1, 12, 0, 0)
|
|
exports = [
|
|
ExportRow("old", base, "/tmp/old.json"),
|
|
ExportRow("new", base + timedelta(days=2), "/tmp/new.json"),
|
|
ExportRow("middle", base + timedelta(days=1), "/tmp/middle.json"),
|
|
ExportRow("unknown", None, "/tmp/unknown.json"),
|
|
]
|
|
|
|
kept, candidates = cleanup.select_cleanup_candidates(exports, keep_latest=2)
|
|
|
|
assert [export.id for export in kept] == ["new", "middle"]
|
|
assert [export.id for export in candidates] == ["old", "unknown"]
|
|
|
|
|
|
def test_cleanup_candidate_selection_rejects_negative_keep_latest() -> None:
|
|
cleanup = load_cleanup_module()
|
|
|
|
try:
|
|
cleanup.select_cleanup_candidates([], keep_latest=-1)
|
|
except ValueError as exc:
|
|
assert "keep_latest" in str(exc)
|
|
else:
|
|
raise AssertionError("negative keep_latest should fail")
|
|
|
|
|
|
def test_cleanup_can_filter_candidates_by_export_type() -> None:
|
|
cleanup = load_cleanup_module()
|
|
base = datetime(2026, 1, 1, 12, 0, 0)
|
|
exports = [
|
|
ExportRow("metadata", base + timedelta(days=2), "/tmp/metadata.json", "project_metadata_json"),
|
|
ExportRow("report", base + timedelta(days=1), "/tmp/report.html", "project_report_html"),
|
|
ExportRow("dataset", base, "/tmp/dataset.geojson", "dataset_geojson"),
|
|
]
|
|
|
|
filtered = cleanup.filter_exports_by_type(exports, ["project_report_html"])
|
|
|
|
assert [export.id for export in filtered] == ["report"]
|
|
|
|
|
|
def test_cleanup_path_safety_requires_storage_root_containment(tmp_path: Path) -> None:
|
|
cleanup = load_cleanup_module()
|
|
storage_root = tmp_path / "storage"
|
|
safe_export = storage_root / "exports" / "project" / "report.html"
|
|
unsafe_export = tmp_path / "outside" / "report.html"
|
|
|
|
safe_export.parent.mkdir(parents=True)
|
|
unsafe_export.parent.mkdir(parents=True)
|
|
|
|
assert cleanup.is_within_storage_root(safe_export, storage_root) is True
|
|
assert cleanup.is_within_storage_root(unsafe_export, storage_root) is False
|
|
|
|
|
|
def test_cleanup_script_defaults_to_explicit_demo_project() -> None:
|
|
cleanup = load_cleanup_module()
|
|
parser = cleanup.build_parser()
|
|
|
|
args = parser.parse_args([])
|
|
|
|
assert args.project_name == cleanup.DEMO_PROJECT_NAME
|
|
assert args.keep_latest == 3
|
|
assert args.max_delete == 25
|
|
assert args.export_type is None
|
|
assert args.apply is False
|
|
|
|
|
|
def test_cleanup_script_accepts_max_delete_and_repeated_export_type() -> None:
|
|
cleanup = load_cleanup_module()
|
|
parser = cleanup.build_parser()
|
|
|
|
args = parser.parse_args(
|
|
[
|
|
"--keep-latest",
|
|
"10",
|
|
"--max-delete",
|
|
"100",
|
|
"--export-type",
|
|
"project_report_html",
|
|
"--export-type",
|
|
"project_metadata_json",
|
|
"--apply",
|
|
]
|
|
)
|
|
|
|
assert args.keep_latest == 10
|
|
assert args.max_delete == 100
|
|
assert args.export_type == ["project_report_html", "project_metadata_json"]
|
|
assert args.apply is True
|
|
|
|
|
|
def test_cleanup_script_reports_dry_run_candidates_separately() -> None:
|
|
script = Path(__file__).resolve().parents[2] / "backend" / "scripts" / "cleanup_demo_artifacts.py"
|
|
content = script.read_text(encoding="utf-8")
|
|
|
|
assert '"candidate_files": []' in content
|
|
assert '"candidate_exports": []' in content
|
|
assert '"export_id": str(export.id)' in content
|
|
assert '"export_type": str(getattr(export, "export_type", ""))' in content
|
|
assert 'summary["candidate_files"].append(str(path))' in content
|
|
assert '"max_delete": max_delete' in content
|
|
assert "blocked_reason" in content
|