FabledCurator/tests/test_tasks_library_audit.py

"""Tests for scan_library_for_rule Celery task.

Eager mode is used so the task runs synchronously in-test and we can
assert state via column selects (post-DML ORM access banned per
reference-async-coredml-test-assertions)."""

import pytest
from PIL import Image
from sqlalchemy import select

import backend.app.tasks.library_audit  # noqa: F401 — celery registration
from backend.app.models import ImageRecord, LibraryAuditRun

pytestmark = pytest.mark.integration


def _mk_image(db_sync, tmp_path, *, mode, color, name):
    path = tmp_path / name
    Image.new(mode, (10, 10), color).save(path)
    # sha256 column is varchar(64) — pad/truncate a per-file pseudo hash
    # exactly to 64 chars. Each test fixture file must have a unique
    # sha256 (reference-image-record-path-unique peer constraint).
    sha = f"audit-{name}".ljust(64, "x")[:64]
    rec = ImageRecord(
        path=str(path),
        sha256=sha,
        size_bytes=path.stat().st_size,
        mime="image/png",
        width=10, height=10,
        origin="imported_filesystem",
        integrity_status="ok",
    )
    db_sync.add(rec)
    db_sync.flush()
    return rec, path


def test_scan_library_for_rule_populates_matched_ids_for_transparency(
    db_sync, tmp_path, celery_conf,
):
    transparent_rec, _ = _mk_image(
        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="trans.png",
    )
    opaque_rec, _ = _mk_image(
        db_sync, tmp_path, mode="RGBA", color=(200, 0, 0, 255), name="opaque.png",
    )
    audit = LibraryAuditRun(
        rule="transparency", params={"threshold": 0.5},
        status="running", matched_ids=[],
    )
    db_sync.add(audit)
    db_sync.commit()
    audit_id = audit.id

    celery_conf(task_always_eager=True)
    from backend.app.tasks.library_audit import scan_library_for_rule
    scan_library_for_rule.run(audit_id)
    celery_conf(task_always_eager=False)

    matched = db_sync.execute(
        select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    status = db_sync.execute(
        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    assert transparent_rec.id in matched
    assert opaque_rec.id not in matched
    assert status == "ready"


def test_scan_time_boxes_and_reenqueues(db_sync, tmp_path, monkeypatch):
    """A zero chunk budget stops before scanning and re-enqueues to continue,
    leaving the audit 'running' — so a huge library can't run the task into the
    Celery limit or hog the maintenance queue (operator-flagged 2026-06-07)."""
    _mk_image(db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="a.png")
    audit = LibraryAuditRun(
        rule="transparency", params={"threshold": 0.5},
        status="running", matched_ids=[],
    )
    db_sync.add(audit)
    db_sync.commit()
    audit_id = audit.id

    from backend.app.tasks import library_audit as la
    delays = []
    monkeypatch.setattr(la.scan_library_for_rule, "delay", lambda aid: delays.append(aid))
    monkeypatch.setattr(la, "_CHUNK_SECONDS", 0)  # time-box on the first iteration
    la.scan_library_for_rule.run(audit_id)

    status = db_sync.execute(
        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    assert status == "running"      # not finished — handed off to the next chunk
    assert delays == [audit_id]     # re-enqueued itself


def test_scan_resumes_and_accumulates_matched(db_sync, tmp_path, monkeypatch):
    """A later chunk keeps the prior chunks' matches and appends its own."""
    trans, _ = _mk_image(
        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="t.png",
    )
    audit = LibraryAuditRun(
        rule="transparency", params={"threshold": 0.5},
        status="running", matched_ids=[424242], resume_after_id=0,
    )
    db_sync.add(audit)
    db_sync.commit()
    audit_id = audit.id

    from backend.app.tasks import library_audit as la
    monkeypatch.setattr(la.scan_library_for_rule, "delay", lambda aid: None)
    la.scan_library_for_rule.run(audit_id)

    matched = db_sync.execute(
        select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    status = db_sync.execute(
        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    assert status == "ready"
    assert 424242 in matched     # carried over from a prior chunk
    assert trans.id in matched   # found this chunk


def test_scan_library_for_rule_skips_missing_files_gracefully(
    db_sync, tmp_path, celery_conf,
):
    rec, path = _mk_image(
        db_sync, tmp_path, mode="RGBA", color=(0, 0, 0, 0), name="ghost.png",
    )
    path.unlink()  # delete the file but leave the DB row
    audit = LibraryAuditRun(
        rule="transparency", params={"threshold": 0.5},
        status="running", matched_ids=[],
    )
    db_sync.add(audit)
    db_sync.commit()
    audit_id = audit.id

    celery_conf(task_always_eager=True)
    from backend.app.tasks.library_audit import scan_library_for_rule
    scan_library_for_rule.run(audit_id)
    celery_conf(task_always_eager=False)

    status = db_sync.execute(
        select(LibraryAuditRun.status).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    matched = db_sync.execute(
        select(LibraryAuditRun.matched_ids).where(LibraryAuditRun.id == audit_id)
    ).scalar_one()
    # Missing file is skipped (warning logged), audit completes successfully.
    assert status == "ready"
    assert rec.id not in matched