diff --git a/tests/test_tasks_admin.py b/tests/test_tasks_admin.py new file mode 100644 index 0000000..da38e2b --- /dev/null +++ b/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]