diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 1484b77..6e08a91 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -93,3 +93,174 @@ def test_cleanup_old_deletes_finished_old(db_sync): remaining = {t.source_path for t in db_sync.query(ImportTask).all()} assert remaining == {"/import/b.jpg", "/import/c.jpg"} + + +# --- FC-3i: task_run sweep + retention ----------------------------- + + +def _make_task_run(db_sync, *, status, started_at, finished_at=None, + error_type=None): + from backend.app.models import TaskRun + row = TaskRun( + celery_task_id="x", + queue="ml", + task_name="backend.app.tasks.fake.t", + target_id=1, + started_at=started_at, + finished_at=finished_at, + duration_ms=1000 if finished_at else None, + status=status, + error_type=error_type, + error_message="x" if status in ("error", "timeout") else None, + ) + db_sync.add(row) + db_sync.flush() + return row.id + + +def test_recover_stalled_task_runs_flips_old_running_to_error(db_sync): + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import recover_stalled_task_runs + + now = datetime.now(UTC) + stale_id = _make_task_run( + db_sync, status="running", started_at=now - timedelta(minutes=10), + ) + db_sync.commit() + + recovered = recover_stalled_task_runs.apply().get() + assert recovered == 1 + + db_sync.expire_all() + status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == stale_id) + ).scalar_one() + error_type = db_sync.execute( + select(TaskRun.error_type).where(TaskRun.id == stale_id) + ).scalar_one() + assert status == "error" + assert error_type == "RecoverySweep" + + +def test_recover_stalled_task_runs_skips_fresh_running(db_sync): + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import recover_stalled_task_runs + + now = datetime.now(UTC) + fresh_id = _make_task_run( + db_sync, status="running", started_at=now - timedelta(seconds=30), + ) + db_sync.commit() + + recovered = recover_stalled_task_runs.apply().get() + assert recovered == 0 + + db_sync.expire_all() + status = db_sync.execute( + select(TaskRun.status).where(TaskRun.id == fresh_id) + ).scalar_one() + assert status == "running" + + +def test_prune_task_runs_deletes_ok_older_than_24h(db_sync): + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import prune_task_runs + + now = datetime.now(UTC) + old_id = _make_task_run( + db_sync, status="ok", + started_at=now - timedelta(hours=30), + finished_at=now - timedelta(hours=29), + ) + recent_id = _make_task_run( + db_sync, status="ok", + started_at=now - timedelta(hours=2), + finished_at=now - timedelta(hours=1), + ) + db_sync.commit() + + result = prune_task_runs.apply().get() + assert result["ok_deleted"] == 1 + + db_sync.expire_all() + surviving_ids = set(db_sync.execute( + select(TaskRun.id).where(TaskRun.id.in_([old_id, recent_id])) + ).scalars().all()) + assert surviving_ids == {recent_id} + + +def test_prune_task_runs_deletes_failures_older_than_7d(db_sync): + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import prune_task_runs + + now = datetime.now(UTC) + old_id = _make_task_run( + db_sync, status="error", error_type="OldError", + started_at=now - timedelta(days=10), + finished_at=now - timedelta(days=9), + ) + db_sync.commit() + + result = prune_task_runs.apply().get() + assert result["failures_deleted"] >= 1 + + db_sync.expire_all() + surviving = db_sync.execute( + select(TaskRun.id).where(TaskRun.id == old_id) + ).scalar_one_or_none() + assert surviving is None + + +def test_prune_task_runs_keeps_recent_failures(db_sync): + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import prune_task_runs + + now = datetime.now(UTC) + recent_id = _make_task_run( + db_sync, status="error", error_type="RecentError", + started_at=now - timedelta(days=6), + finished_at=now - timedelta(days=5), + ) + db_sync.commit() + + prune_task_runs.apply().get() + + db_sync.expire_all() + surviving = db_sync.execute( + select(TaskRun.id).where(TaskRun.id == recent_id) + ).scalar_one_or_none() + assert surviving == recent_id + + +def test_prune_task_runs_never_deletes_running(db_sync): + """Even a 30-day-old running row stays — recovery sweep is the + mechanism that flips them; prune doesn't touch in-flight.""" + from sqlalchemy import select + + from backend.app.models import TaskRun + from backend.app.tasks.maintenance import prune_task_runs + + now = datetime.now(UTC) + ancient_id = _make_task_run( + db_sync, status="running", + started_at=now - timedelta(days=30), + ) + db_sync.commit() + + prune_task_runs.apply().get() + + db_sync.expire_all() + surviving = db_sync.execute( + select(TaskRun.id).where(TaskRun.id == ancient_id) + ).scalar_one_or_none() + assert surviving == ancient_id