520 lines
20 KiB
Python
520 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
if str(BACKEND_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_DIR))
|
|
|
|
from service import app as service_app # noqa: E402
|
|
from service.cleanup_service import CleanupService # noqa: E402
|
|
from service.metadata_store import MetadataStore # noqa: E402
|
|
from service.product_archive_store import ProductArchiveStore # noqa: E402
|
|
from service.schemas import JobStatus, ProductInput # noqa: E402
|
|
from service.storage import Storage # noqa: E402
|
|
from service import storage_metrics # noqa: E402
|
|
|
|
|
|
NOW = datetime(2026, 9, 12, 12, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def make_cleanup_service(tmp_path: Path) -> CleanupService:
|
|
return CleanupService(
|
|
Storage(tmp_path / "workspace"),
|
|
MetadataStore(tmp_path / "metadata" / "app.db"),
|
|
ProductArchiveStore(tmp_path / "products" / "metadata"),
|
|
)
|
|
|
|
|
|
def create_success_job(
|
|
service: CleanupService,
|
|
*,
|
|
age_days: float,
|
|
archived: bool = False,
|
|
create_folder: bool = True,
|
|
) -> str:
|
|
job_id = f"{len(service.metadata_store.job_ids()) + 1:032x}"
|
|
created_at = NOW - timedelta(days=age_days)
|
|
db_path = service.storage.job_root(job_id) / "output" / "word_locations.sqlite"
|
|
if create_folder:
|
|
db_path.parent.mkdir(parents=True)
|
|
db_path.write_bytes(b"db-content")
|
|
service.metadata_store.upsert_job(
|
|
JobStatus(
|
|
job_id=job_id,
|
|
status="success",
|
|
stage="done",
|
|
progress_percent=100,
|
|
message="done",
|
|
artifacts={"db": str(db_path)},
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
)
|
|
)
|
|
if archived:
|
|
product = service.product_store.upsert_product(ProductInput(name="Archived", source="manual"))
|
|
service.product_store.write_archive_version(
|
|
product.product_id,
|
|
version_id=f"ver_{job_id}",
|
|
metadata={},
|
|
preview_path="",
|
|
archives=[
|
|
{
|
|
"archive_path": "snapshot.sqlite",
|
|
"source_job_id": job_id,
|
|
"source_asset_id": "asset-1",
|
|
"db_checksum": "checksum",
|
|
}
|
|
],
|
|
now=NOW,
|
|
)
|
|
return job_id
|
|
|
|
|
|
def create_pending_product(
|
|
service: CleanupService,
|
|
*,
|
|
purge_after: datetime,
|
|
create_folder: bool = True,
|
|
):
|
|
product = service.product_store.upsert_product(ProductInput(name="Pending", source="manual"))
|
|
marked_at = purge_after - timedelta(days=30)
|
|
product = service.product_store.mark_pending_cleanup(product.product_id, now=marked_at)
|
|
if create_folder:
|
|
product_dir = service.product_store.root.parent / product.product_id
|
|
product_dir.mkdir(parents=True)
|
|
(product_dir / "preview.png").write_bytes(b"product-bytes")
|
|
return product
|
|
|
|
|
|
def archive_job(service: CleanupService, job_id: str) -> None:
|
|
product = service.product_store.upsert_product(ProductInput(name="Late archive", source="manual"))
|
|
service.product_store.write_archive_version(
|
|
product.product_id,
|
|
version_id=f"ver_late_{job_id}",
|
|
metadata={},
|
|
preview_path="",
|
|
archives=[
|
|
{
|
|
"archive_path": "snapshot.sqlite",
|
|
"source_job_id": job_id,
|
|
"source_asset_id": "asset-late",
|
|
"db_checksum": "checksum",
|
|
}
|
|
],
|
|
now=NOW,
|
|
)
|
|
|
|
|
|
def auth_header() -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {service_app._orders_token()}"}
|
|
|
|
|
|
def test_cleanup_skips_archived_job_and_marks_job_at_23_day_threshold_for_reminder(tmp_path):
|
|
service = make_cleanup_service(tmp_path)
|
|
archived_job = create_success_job(service, age_days=31, archived=True)
|
|
remind_job = create_success_job(service, age_days=23)
|
|
almost_remind_job = create_success_job(service, age_days=23 - (1 / 86400))
|
|
|
|
report = service.preview(now=NOW)
|
|
|
|
assert archived_job not in {item.job_id for item in report.temporary_jobs}
|
|
assert remind_job in report.reminder_job_ids
|
|
assert almost_remind_job not in report.reminder_job_ids
|
|
|
|
|
|
def test_apply_removes_only_due_unarchived_job_and_due_product(tmp_path):
|
|
service = make_cleanup_service(tmp_path)
|
|
due_job = create_success_job(service, age_days=30)
|
|
young_job = create_success_job(service, age_days=29)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
|
|
report = service.apply(now=NOW)
|
|
|
|
assert report.deleted_job_ids == [due_job]
|
|
assert not service.storage.job_root(due_job).exists()
|
|
assert service.storage.job_root(young_job).exists()
|
|
assert due_job not in service.metadata_store.job_ids()
|
|
assert product.product_id in report.deleted_product_ids
|
|
assert not (service.product_store.root.parent / product.product_id).exists()
|
|
assert service.product_store.get_product(product.product_id).status == "purged"
|
|
|
|
|
|
def test_apply_treats_missing_job_and_product_folders_as_idempotent_success(tmp_path):
|
|
service = make_cleanup_service(tmp_path)
|
|
due_job = create_success_job(service, age_days=30, create_folder=False)
|
|
product = create_pending_product(service, purge_after=NOW, create_folder=False)
|
|
|
|
report = service.apply(now=NOW)
|
|
|
|
assert report.deleted_job_ids == [due_job]
|
|
assert report.deleted_product_ids == [product.product_id]
|
|
assert due_job not in service.metadata_store.job_ids()
|
|
assert service.product_store.get_product(product.product_id).status == "purged"
|
|
|
|
|
|
def test_restored_product_is_not_deleted_when_original_purge_time_arrives(tmp_path):
|
|
service = make_cleanup_service(tmp_path)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
product_dir = service.product_store.root.parent / product.product_id
|
|
service.product_store.restore_product(product.product_id, now=NOW - timedelta(days=1))
|
|
|
|
report = service.apply(now=NOW)
|
|
|
|
assert product.product_id not in report.deleted_product_ids
|
|
assert product_dir.exists()
|
|
assert service.product_store.get_product(product.product_id).status == "active"
|
|
|
|
|
|
def test_atomic_purge_claim_blocks_restore_before_product_files_are_deleted(
|
|
tmp_path, monkeypatch
|
|
):
|
|
service = make_cleanup_service(tmp_path)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
restore_conflicts: list[str] = []
|
|
|
|
def attempt_restore(product_id: str) -> None:
|
|
with pytest.raises(ValueError, match="purging"):
|
|
service.product_store.restore_product(product_id, now=NOW)
|
|
restore_conflicts.append(product_id)
|
|
|
|
monkeypatch.setattr(
|
|
service, "_before_product_files_delete", attempt_restore, raising=False
|
|
)
|
|
|
|
report = service.apply(now=NOW)
|
|
|
|
assert restore_conflicts == [product.product_id]
|
|
assert report.deleted_product_ids == [product.product_id]
|
|
assert service.product_store.get_product(product.product_id).status == "purged"
|
|
|
|
|
|
def test_failed_product_file_deletion_enters_failed_cleanup_and_is_not_retried(
|
|
tmp_path, monkeypatch
|
|
):
|
|
service = make_cleanup_service(tmp_path)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
|
|
monkeypatch.setattr(
|
|
"service.cleanup_service.shutil.rmtree",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")),
|
|
)
|
|
|
|
with pytest.raises(OSError, match="disk failure"):
|
|
service.apply(now=NOW)
|
|
|
|
record = service.product_store.get_product(product.product_id)
|
|
cleanup = service.product_store._fetchone(
|
|
"SELECT status FROM cleanup_records WHERE product_id = ?", (product.product_id,)
|
|
)["status"]
|
|
|
|
assert record.status == "failed_cleanup"
|
|
assert record.purge_after == NOW
|
|
assert cleanup == "failed_cleanup"
|
|
assert service.product_store.due_product_cleanups(NOW) == []
|
|
assert (service.product_store.root.parent / product.product_id).exists()
|
|
|
|
|
|
def test_failed_cleanup_product_can_be_restored_or_requeued(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
|
|
def _fail_purge(purge_after: datetime) -> str:
|
|
product = create_pending_product(service, purge_after=purge_after)
|
|
monkeypatch.setattr(
|
|
"service.cleanup_service.shutil.rmtree",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")),
|
|
)
|
|
with pytest.raises(OSError):
|
|
service.apply(now=purge_after)
|
|
assert service.product_store.get_product(product.product_id).status == "failed_cleanup"
|
|
return product.product_id
|
|
|
|
failed = _fail_purge(NOW)
|
|
restored = service.product_store.restore_product(failed, now=NOW + timedelta(days=1))
|
|
assert restored.status == "active"
|
|
assert restored.purge_after is None
|
|
|
|
failed_requeue = _fail_purge(NOW)
|
|
requeued = service.product_store.mark_pending_cleanup(
|
|
failed_requeue, now=NOW + timedelta(days=1)
|
|
)
|
|
assert requeued.status == "pending_cleanup"
|
|
assert requeued.purge_after == NOW + timedelta(days=31)
|
|
assert service.product_store.due_product_cleanups(NOW + timedelta(days=5)) == []
|
|
assert service.product_store.due_product_cleanups(NOW + timedelta(days=31)) == [
|
|
requeued
|
|
]
|
|
|
|
|
|
def test_stale_purging_claim_is_reconciled_to_failed_cleanup(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
store = service.product_store
|
|
|
|
original_fail = store.fail_product_purge
|
|
calls = {"n": 0}
|
|
|
|
def flaky_fail(product_id, now=None):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise sqlite3.OperationalError("simulated busy")
|
|
return original_fail(product_id, now=now)
|
|
|
|
monkeypatch.setattr(store, "fail_product_purge", flaky_fail)
|
|
monkeypatch.setattr(
|
|
"service.cleanup_service.shutil.rmtree",
|
|
lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")),
|
|
)
|
|
|
|
with pytest.raises(sqlite3.OperationalError, match="simulated busy"):
|
|
service.apply(now=NOW)
|
|
|
|
stranded = store._fetchone(
|
|
"SELECT status, claimed_at FROM cleanup_records WHERE product_id = ?",
|
|
(product.product_id,),
|
|
)
|
|
assert stranded["status"] == "purging"
|
|
assert datetime.fromisoformat(stranded["claimed_at"]) == NOW
|
|
|
|
report = service.preview(now=NOW + timedelta(hours=2))
|
|
|
|
assert [item.product_id for item in report.failed_products] == [product.product_id]
|
|
assert store.get_product(product.product_id).status == "failed_cleanup"
|
|
assert store.due_product_cleanups(NOW + timedelta(days=1)) == []
|
|
|
|
|
|
def test_cleanup_candidates_reports_failed_products(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
product = create_pending_product(service, purge_after=NOW)
|
|
store = service.product_store
|
|
store.claim_product_purge(product.product_id, NOW)
|
|
store.fail_product_purge(product.product_id, now=NOW)
|
|
monkeypatch.setattr(service_app, "cleanup_service", service, raising=False)
|
|
client = TestClient(service_app.app)
|
|
|
|
response = client.get("/api/maintenance/cleanup-candidates", headers=auth_header())
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert [item["product_id"] for item in data["failed_products"]] == [
|
|
product.product_id
|
|
]
|
|
assert data["pending_products"] == []
|
|
|
|
|
|
def test_apply_rechecks_archive_reference_immediately_before_job_deletion(tmp_path):
|
|
service = make_cleanup_service(tmp_path)
|
|
due_job = create_success_job(service, age_days=31)
|
|
original_preview = service.preview
|
|
|
|
def preview_then_archive(now: datetime):
|
|
report = original_preview(now)
|
|
archive_job(service, due_job)
|
|
return report
|
|
|
|
service.preview = preview_then_archive # type: ignore[method-assign]
|
|
|
|
report = service.apply(now=NOW)
|
|
|
|
assert due_job not in report.deleted_job_ids
|
|
assert service.storage.job_root(due_job).exists()
|
|
assert due_job in service.metadata_store.job_ids()
|
|
|
|
|
|
def test_cleanup_routes_require_admin_auth_and_confirmed_enabled_apply(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
due_job = create_success_job(service, age_days=3650)
|
|
monkeypatch.setattr(service_app, "cleanup_service", service)
|
|
monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "false")
|
|
client = TestClient(service_app.app)
|
|
|
|
assert client.get("/api/maintenance/cleanup-candidates").status_code == 403
|
|
assert client.post("/api/maintenance/cleanup-run", json={"confirm": True}).status_code == 403
|
|
assert client.post(
|
|
"/api/maintenance/cleanup-run", json={"confirm": False}, headers=auth_header()
|
|
).status_code == 400
|
|
disabled = client.post(
|
|
"/api/maintenance/cleanup-run", json={"confirm": True}, headers=auth_header()
|
|
)
|
|
assert disabled.status_code == 409
|
|
assert service.storage.job_root(due_job).exists()
|
|
|
|
monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true")
|
|
preview = client.get("/api/maintenance/cleanup-candidates", headers=auth_header())
|
|
applied = client.post(
|
|
"/api/maintenance/cleanup-run", json={"confirm": True}, headers=auth_header()
|
|
)
|
|
|
|
assert preview.status_code == 200
|
|
assert preview.json()["temporary_jobs"][0]["job_id"] == due_job
|
|
assert applied.status_code == 200
|
|
assert applied.json()["deleted_job_ids"] == [due_job]
|
|
assert not service.storage.job_root(due_job).exists()
|
|
|
|
|
|
def test_scheduled_cleanup_is_preview_only_until_apply_flag_is_enabled(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
first_job = create_success_job(service, age_days=31)
|
|
monkeypatch.setattr(service_app, "cleanup_service", service)
|
|
monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "false")
|
|
|
|
service_app._run_scheduled_cleanup_once(now=NOW)
|
|
|
|
assert service.storage.job_root(first_job).exists()
|
|
|
|
second_job = create_success_job(service, age_days=31)
|
|
monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true")
|
|
service_app._run_scheduled_cleanup_once(now=NOW)
|
|
|
|
assert not service.storage.job_root(first_job).exists()
|
|
assert not service.storage.job_root(second_job).exists()
|
|
|
|
|
|
def test_scheduler_stop_retains_live_worker_and_restart_does_not_duplicate_cycle(monkeypatch):
|
|
class LiveThread:
|
|
def __init__(self):
|
|
self.join_calls: list[float | None] = []
|
|
|
|
def is_alive(self) -> bool:
|
|
return True
|
|
|
|
def join(self, timeout=None) -> None:
|
|
self.join_calls.append(timeout)
|
|
|
|
worker = LiveThread()
|
|
cycles: list[str] = []
|
|
monkeypatch.setattr(service_app, "_cleanup_scheduler_thread", worker)
|
|
monkeypatch.setattr(service_app, "_run_scheduled_cleanup_once", lambda: cycles.append("run"))
|
|
|
|
service_app._stop_cleanup_scheduler()
|
|
service_app._start_cleanup_scheduler()
|
|
|
|
assert service_app._cleanup_scheduler_stop.is_set()
|
|
assert service_app._cleanup_scheduler_thread is worker
|
|
assert worker.join_calls
|
|
assert cycles == []
|
|
|
|
|
|
def test_storage_summary_separates_archive_protection_from_asset_references(tmp_path, monkeypatch):
|
|
service = make_cleanup_service(tmp_path)
|
|
archived_job = create_success_job(service, age_days=3650, archived=True)
|
|
asset_job = create_success_job(service, age_days=3650)
|
|
assets_dir = tmp_path / "assets"
|
|
asset_dir = assets_dir / "as" / "asset-old"
|
|
asset_dir.mkdir(parents=True)
|
|
(asset_dir / "meta.json").write_text(
|
|
'{"asset_id":"asset-old","job_id":"' + asset_job + '"}', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(service_app, "storage", service.storage)
|
|
monkeypatch.setattr(service_app, "metadata_store", service.metadata_store)
|
|
monkeypatch.setattr(service_app, "product_archive_store", service.product_store)
|
|
monkeypatch.setattr(service_app, "cleanup_service", service)
|
|
monkeypatch.setattr(service_app, "ASSETS_DIR", assets_dir)
|
|
|
|
summary = service_app.storage_summary()
|
|
|
|
assert summary["archive_protected_job_ids"] == [archived_job]
|
|
assert summary["asset_referenced_job_ids"] == [asset_job]
|
|
assert summary["temporary_job_ids"] == [asset_job]
|
|
|
|
|
|
def test_storage_metrics_protects_archives_but_not_old_asset_references(
|
|
tmp_path, monkeypatch, capsys
|
|
):
|
|
workspace = tmp_path / "workspace"
|
|
assets = tmp_path / "assets"
|
|
metadata = tmp_path / "metadata"
|
|
products = tmp_path / "products"
|
|
storage = Storage(workspace)
|
|
archived_job = "a" * 32
|
|
asset_job = "b" * 32
|
|
for job_id in (archived_job, asset_job):
|
|
root = storage.job_root(job_id)
|
|
root.mkdir(parents=True)
|
|
(root / "payload.bin").write_bytes(b"payload")
|
|
old_timestamp = (NOW - timedelta(days=31)).timestamp()
|
|
os.utime(root, (old_timestamp, old_timestamp))
|
|
|
|
asset_dir = assets / "as" / "asset-old"
|
|
asset_dir.mkdir(parents=True)
|
|
(asset_dir / "meta.json").write_text(
|
|
json.dumps({"asset_id": "asset-old", "job_id": asset_job}), encoding="utf-8"
|
|
)
|
|
product_store = ProductArchiveStore(products / "metadata")
|
|
product = product_store.upsert_product(ProductInput(name="Archive", source="manual"))
|
|
product_store.write_archive_version(
|
|
product.product_id,
|
|
version_id="ver_archive",
|
|
metadata={},
|
|
preview_path="",
|
|
archives=[{
|
|
"archive_path": "snapshot.sqlite",
|
|
"source_job_id": archived_job,
|
|
"source_asset_id": "asset-archive",
|
|
"db_checksum": "checksum",
|
|
}],
|
|
now=NOW,
|
|
)
|
|
monkeypatch.setattr(storage_metrics, "WORKSPACE_DIR", workspace)
|
|
monkeypatch.setattr(storage_metrics, "ASSETS_DIR", assets)
|
|
monkeypatch.setattr(storage_metrics, "METADATA_DIR", metadata)
|
|
monkeypatch.setattr(storage_metrics, "PRODUCTS_DIR", products, raising=False)
|
|
monkeypatch.setattr(sys, "argv", ["storage_metrics", "--max-age-days", "30"])
|
|
|
|
storage_metrics.main()
|
|
report = json.loads(capsys.readouterr().out)
|
|
|
|
assert report["archive_protected_job_ids"] == [archived_job]
|
|
assert report["asset_referenced_job_ids"] == [asset_job]
|
|
assert [item["job_id"] for item in report["stale_jobs"]] == [asset_job]
|
|
|
|
|
|
def test_storage_metrics_apply_never_deletes_unproven_or_failed_workspaces(
|
|
tmp_path, monkeypatch
|
|
):
|
|
workspace = tmp_path / "workspace"
|
|
metadata = tmp_path / "metadata"
|
|
storage = Storage(workspace)
|
|
metadata_store = MetadataStore(metadata / "app.db")
|
|
orphan_job = "c" * 32
|
|
failed_job = "d" * 32
|
|
for job_id in (orphan_job, failed_job):
|
|
root = storage.job_root(job_id)
|
|
root.mkdir(parents=True)
|
|
(root / "payload.bin").write_bytes(b"payload")
|
|
old_timestamp = (datetime.now(timezone.utc) - timedelta(days=31)).timestamp()
|
|
os.utime(root, (old_timestamp, old_timestamp))
|
|
metadata_store.upsert_job(
|
|
JobStatus(
|
|
job_id=failed_job,
|
|
status="failed",
|
|
stage="failed",
|
|
progress_percent=100,
|
|
message="failed",
|
|
artifacts={"db": str(storage.job_root(failed_job) / "output" / "db.sqlite")},
|
|
error="generation failed",
|
|
created_at=datetime.now(timezone.utc) - timedelta(days=31),
|
|
updated_at=datetime.now(timezone.utc) - timedelta(days=31),
|
|
)
|
|
)
|
|
monkeypatch.setattr(storage_metrics, "WORKSPACE_DIR", workspace)
|
|
monkeypatch.setattr(storage_metrics, "ASSETS_DIR", tmp_path / "assets")
|
|
monkeypatch.setattr(storage_metrics, "METADATA_DIR", metadata)
|
|
monkeypatch.setattr(storage_metrics, "PRODUCTS_DIR", tmp_path / "products")
|
|
monkeypatch.setattr(sys, "argv", ["storage_metrics", "--apply"])
|
|
monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true")
|
|
|
|
storage_metrics.main()
|
|
|
|
assert storage.job_root(orphan_job).exists()
|
|
assert storage.job_root(failed_job).exists()
|