FabledCurator/tests/test_tasks_admin.py

"""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 pytest
from sqlalchemy import func, select

import backend.app.tasks.admin  # noqa: F401 — register tasks
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, mime="image/jpeg",
            origin="imported_filesystem",
        ))
    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, mime="image/jpeg",
            origin="imported_filesystem",
        )
        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]