fc3h: Celery task integration tests (backup, restore, prune, nightly)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,278 @@
|
||||
"""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 _fake_run(cmd, **kwargs):
|
||||
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")
|
||||
return _FakeProc()
|
||||
|
||||
monkeypatch.setattr("subprocess.run", _fake_run)
|
||||
|
||||
|
||||
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(".sql")
|
||||
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")
|
||||
monkeypatch.setattr("subprocess.run", _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"}
|
||||
|
||||
|
||||
# --- 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
|
||||
Reference in New Issue
Block a user