"""FC-3h: backup/restore Celery task integration tests. Uses task_always_eager for synchronous in-test execution. Subprocess + IMAGES_ROOT are monkeypatched so tests don't touch real /images or shell out to pg_dump. """ from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from sqlalchemy import select from backend.app.celery_app import celery from backend.app.models import BackupRun, ImportSettings pytestmark = pytest.mark.integration @pytest.fixture(autouse=True) def _eager_celery(monkeypatch): monkeypatch.setattr(celery.conf, "task_always_eager", True) monkeypatch.setattr(celery.conf, "task_eager_propagates", False) @pytest.fixture(autouse=True) def fake_subprocess_and_images_root(monkeypatch, tmp_path): monkeypatch.setattr("backend.app.tasks.backup.IMAGES_ROOT", tmp_path) monkeypatch.setattr( "backend.app.services.backup_service._DB_SUBPROCESS_TIMEOUT_S", 5, ) class _FakeProc: returncode = 0 stdout = b"" stderr = b"" def _sentinel(cmd): if cmd[0] == "pg_dump": i = cmd.index("-f") Path(cmd[i + 1]).write_bytes(b"-- fake pg_dump\n") elif cmd[0] == "tar" and "-cf" in cmd: i = cmd.index("-cf") Path(cmd[i + 1]).write_bytes(b"fake tar payload") def _fake_run(cmd, **kwargs): _sentinel(cmd) return _FakeProc() class _FakePopen: def __init__(self, cmd, **kwargs): self.returncode = 0 _sentinel(cmd) def communicate(self, timeout=None): return (b"", b"") def kill(self): self.returncode = -9 # backup_db/backup_images go through _run_bounded (Popen); restore via run. monkeypatch.setattr("subprocess.run", _fake_run) monkeypatch.setattr("subprocess.Popen", _FakePopen) def _seed_backup(db_sync, *, kind, status, started_at, tag=None, finished_at=None): row = BackupRun( kind=kind, status=status, tag=tag, triggered_by="manual", started_at=started_at, finished_at=finished_at or (started_at + timedelta(seconds=10)), sql_path="/tmp/fake.sql" if kind == "db" else None, tar_path="/tmp/fake.tar.zst" if kind == "images" else None, manifest={}, ) db_sync.add(row) db_sync.flush() return row.id # --- backup_db_task -------------------------------------------------- @pytest.mark.asyncio async def test_backup_db_task_creates_backup_run_row_status_ok(db_sync): from backend.app.tasks.backup import backup_db_task result = backup_db_task.delay(tag=None, triggered_by="manual").get() run_id = result["backup_run_id"] row = db_sync.execute( select( BackupRun.kind, BackupRun.status, BackupRun.sql_path, BackupRun.size_bytes, BackupRun.finished_at, BackupRun.error, ).where(BackupRun.id == run_id) ).one() assert row.kind == "db" assert row.status == "ok" assert row.sql_path and row.sql_path.endswith(".dump") assert row.size_bytes is not None and row.size_bytes > 0 assert row.finished_at is not None assert row.error is None @pytest.mark.asyncio async def test_backup_db_task_records_failure_on_subprocess_error(db_sync, monkeypatch): from backend.app.tasks.backup import backup_db_task def _boom(*a, **kw): raise RuntimeError("synthetic pg_dump fail") # backup_db dumps via _run_bounded → subprocess.Popen. monkeypatch.setattr("subprocess.Popen", _boom) with pytest.raises(RuntimeError): backup_db_task.delay().get() row = db_sync.execute( select(BackupRun.status, BackupRun.error) .where(BackupRun.kind == "db") .order_by(BackupRun.id.desc()).limit(1) ).one() assert row.status == "error" assert "synthetic" in (row.error or "") @pytest.mark.asyncio async def test_backup_db_task_persists_tag(db_sync): from backend.app.tasks.backup import backup_db_task result = backup_db_task.delay(tag="pre-cutover").get() tag = db_sync.execute( select(BackupRun.tag).where(BackupRun.id == result["backup_run_id"]) ).scalar_one() assert tag == "pre-cutover" # --- backup_images_task --------------------------------------------- @pytest.mark.asyncio async def test_backup_images_task_creates_backup_run(db_sync): from backend.app.tasks.backup import backup_images_task result = backup_images_task.delay().get() row = db_sync.execute( select(BackupRun.kind, BackupRun.status, BackupRun.tar_path) .where(BackupRun.id == result["backup_run_id"]) ).one() assert row.kind == "images" assert row.status == "ok" assert row.tar_path and row.tar_path.endswith(".tar.zst") # --- restore_db_task ------------------------------------------------ @pytest.mark.asyncio async def test_restore_db_task_creates_restoring_marker_then_restored(db_sync): from backend.app.tasks.backup import backup_db_task, restore_db_task src = backup_db_task.delay().get() src_id = src["backup_run_id"] restore_db_task.delay(source_backup_run_id=src_id).get() rows = db_sync.execute( select(BackupRun.status, BackupRun.triggered_by) .where(BackupRun.restored_from_id == src_id) ).all() assert len(rows) == 1 assert rows[0].status == "restored" assert rows[0].triggered_by == "restore" @pytest.mark.asyncio async def test_restore_db_task_rejects_non_db_source(db_sync): from backend.app.tasks.backup import restore_db_task now = datetime.now(UTC) img_id = _seed_backup(db_sync, kind="images", status="ok", started_at=now) db_sync.commit() with pytest.raises(ValueError): restore_db_task.delay(source_backup_run_id=img_id).get() # --- prune_backups -------------------------------------------------- def test_prune_backups_keeps_last_n_untagged_per_kind(db_sync): from backend.app.tasks.backup import prune_backups s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_keep_last_n = 2 s.backup_images_keep_last_n = 1 db_sync.commit() now = datetime.now(UTC) for i in range(4): _seed_backup(db_sync, kind="db", status="ok", started_at=now - timedelta(hours=i)) for i in range(3): _seed_backup(db_sync, kind="images", status="ok", started_at=now - timedelta(hours=i)) db_sync.commit() result = prune_backups.apply().get() assert result["db_deleted"] == 2 assert result["images_deleted"] == 2 def test_prune_backups_protects_tagged_rows(db_sync): from backend.app.tasks.backup import prune_backups s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_keep_last_n = 1 db_sync.commit() now = datetime.now(UTC) _seed_backup(db_sync, kind="db", status="ok", started_at=now) tagged_id = _seed_backup( db_sync, kind="db", status="ok", started_at=now - timedelta(days=30), tag="forever", ) _seed_backup(db_sync, kind="db", status="ok", started_at=now - timedelta(days=1)) db_sync.commit() prune_backups.apply().get() surviving_tag = db_sync.execute( select(BackupRun.tag).where(BackupRun.id == tagged_id) ).scalar_one_or_none() assert surviving_tag == "forever" def test_prune_backups_never_deletes_running_or_restoring(db_sync): from backend.app.tasks.backup import prune_backups s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_keep_last_n = 0 db_sync.commit() ancient = datetime.now(UTC) - timedelta(days=30) running_id = _seed_backup(db_sync, kind="db", status="running", started_at=ancient) restoring_id = _seed_backup(db_sync, kind="db", status="restoring", started_at=ancient) db_sync.commit() prune_backups.apply().get() statuses = db_sync.execute( select(BackupRun.status).where(BackupRun.id.in_([running_id, restoring_id])) ).scalars().all() assert set(statuses) == {"running", "restoring"} # --- recover_stalled_backup_runs (per-kind threshold, FC #739) ------- def test_stall_sweep_flips_db_fast_but_spares_running_images(db_sync): from backend.app.tasks.maintenance import recover_stalled_backup_runs now = datetime.now(UTC) # A db backup stuck 50 min → past the 40-min db window → flipped to error. db_id = _seed_backup( db_sync, kind="db", status="running", started_at=now - timedelta(minutes=50), ) # An images backup running 50 min is still well under the 7h window → spared. img_id = _seed_backup( db_sync, kind="images", status="running", started_at=now - timedelta(minutes=50), ) db_sync.commit() recover_stalled_backup_runs.apply().get() db_status = db_sync.execute( select(BackupRun.status).where(BackupRun.id == db_id) ).scalar_one() img_status = db_sync.execute( select(BackupRun.status).where(BackupRun.id == img_id) ).scalar_one() assert db_status == "error" assert img_status == "running" # --- backup_db_nightly ---------------------------------------------- def test_nightly_skips_when_disabled(db_sync): from backend.app.tasks.backup import backup_db_nightly s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_nightly_enabled = False db_sync.commit() result = backup_db_nightly.apply().get() assert "skipped" in result and "disabled" in result["skipped"] def test_nightly_skips_when_hour_mismatch(db_sync): from backend.app.tasks.backup import backup_db_nightly s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_nightly_enabled = True s.backup_db_nightly_hour_utc = (datetime.now(UTC).hour + 12) % 24 db_sync.commit() result = backup_db_nightly.apply().get() assert "skipped" in result and "hour=" in result["skipped"] def test_nightly_dispatches_when_enabled_at_configured_hour(db_sync): from backend.app.tasks.backup import backup_db_nightly s = db_sync.execute( select(ImportSettings).where(ImportSettings.id == 1) ).scalar_one() s.backup_db_nightly_enabled = True s.backup_db_nightly_hour_utc = datetime.now(UTC).hour db_sync.commit() result = backup_db_nightly.apply().get() assert "dispatched" in result