fc3k: admin Celery task tests — registration, success, failure, missing-id idempotency

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/test_tasks_admin.py

@@ -0,0 +1,156 @@
+"""FC-3k: admin Celery task integration tests.
+
+task_always_eager + signal handlers from FC-3i populate task_run.
+We assert the wrapper passes args through correctly and that
+task_run lifecycle status flips as expected.
+"""
+import backend.app.tasks.admin  # noqa: F401 — register tasks
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.celery_app import celery
+from backend.app.models import Artist, ImageRecord, TaskRun
+
+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_images_root(monkeypatch, tmp_path):
+    monkeypatch.setattr("backend.app.tasks.admin.IMAGES_ROOT", tmp_path)
+
+
+# --- registration ----------------------------------------------------
+
+
+def test_delete_artist_cascade_task_registered():
+    assert (
+        "backend.app.tasks.admin.delete_artist_cascade_task"
+        in celery.tasks
+    )
+
+
+def test_bulk_delete_images_task_registered():
+    assert (
+        "backend.app.tasks.admin.bulk_delete_images_task"
+        in celery.tasks
+    )
+
+
+# --- delete_artist_cascade_task -------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_delete_artist_cascade_task_removes_artist_and_records_ok(
+    db_sync, tmp_path,
+):
+    from backend.app.tasks.admin import delete_artist_cascade_task
+
+    a = Artist(name="Doomed", slug="doomed")
+    db_sync.add(a)
+    db_sync.flush()
+    for i in range(2):
+        f = tmp_path / f"d{i}.jpg"
+        f.write_bytes(b"x")
+        db_sync.add(ImageRecord(
+            artist_id=a.id, path=str(f),
+            sha256=f"{i:064x}", size_bytes=10,
+        ))
+    db_sync.commit()
+    artist_id = a.id
+
+    result = delete_artist_cascade_task.delay(artist_id=artist_id).get()
+    assert result["summary"]["images_deleted"] == 2
+
+    surviving = db_sync.execute(
+        select(func.count(Artist.id)).where(Artist.id == artist_id)
+    ).scalar_one()
+    assert surviving == 0
+
+    # FC-3i task_run lifecycle: task should have an 'ok' row.
+    status = db_sync.execute(
+        select(TaskRun.status)
+        .where(TaskRun.task_name.endswith(".delete_artist_cascade_task"))
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).scalar_one()
+    assert status == "ok"
+
+
+@pytest.mark.asyncio
+async def test_delete_artist_cascade_task_records_failure(
+    db_sync, monkeypatch,
+):
+    from backend.app.tasks.admin import delete_artist_cascade_task
+
+    def _boom(*a, **kw):
+        raise RuntimeError("synthetic cascade fail")
+    monkeypatch.setattr(
+        "backend.app.services.cleanup_service.delete_artist_cascade", _boom,
+    )
+    with pytest.raises(RuntimeError):
+        delete_artist_cascade_task.delay(artist_id=1).get()
+
+    row = db_sync.execute(
+        select(TaskRun.status, TaskRun.error_message)
+        .where(TaskRun.task_name.endswith(".delete_artist_cascade_task"))
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).one()
+    assert row.status == "error"
+    assert "synthetic" in (row.error_message or "")
+
+
+# --- bulk_delete_images_task ----------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_images_task_removes_listed_images(
+    db_sync, tmp_path,
+):
+    from backend.app.tasks.admin import bulk_delete_images_task
+
+    a = Artist(name="B", slug="b")
+    db_sync.add(a)
+    db_sync.flush()
+    ids = []
+    for i in range(3):
+        f = tmp_path / f"b{i}.jpg"
+        f.write_bytes(b"x")
+        img = ImageRecord(
+            artist_id=a.id, path=str(f),
+            sha256=f"a{i:063x}", size_bytes=10,
+        )
+        db_sync.add(img)
+        db_sync.flush()
+        ids.append(img.id)
+    db_sync.commit()
+
+    result = bulk_delete_images_task.delay(image_ids=ids).get()
+    assert result["images_deleted"] == 3
+
+    surviving = db_sync.execute(
+        select(func.count(ImageRecord.id))
+        .where(ImageRecord.id.in_(ids))
+    ).scalar_one()
+    assert surviving == 0
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_images_task_idempotent_on_empty_list(db_sync):
+    from backend.app.tasks.admin import bulk_delete_images_task
+    result = bulk_delete_images_task.delay(image_ids=[]).get()
+    assert result["images_deleted"] == 0
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_images_task_reports_missing_ids(db_sync):
+    from backend.app.tasks.admin import bulk_delete_images_task
+    result = bulk_delete_images_task.delay(
+        image_ids=[9_999_998, 9_999_999],
+    ).get()
+    assert result["images_deleted"] == 0
+    assert sorted(result["missing_ids"]) == [9_999_998, 9_999_999]