FabledCurator/tests/test_presentation_auto_apply.py

"""Presentation-chrome auto-hide sweep (#141). numpy-only (no sklearn), so the
apply + guard logic is tested directly via the sync session."""
import pytest
from sqlalchemy import select

from backend.app.models import (
    HeadAutoApplyRun,
    ImageRecord,
    MLSettings,
    PresentationReview,
    Tag,
    TagHead,
    TagKind,
)
from backend.app.models.tag import image_tag
from backend.app.services.ml.heads import (
    auto_apply_sweep,
    system_tag_auto_apply_sweep,
)

pytestmark = pytest.mark.integration


def _emb(slot: int) -> list[float]:
    v = [0.0] * 1152
    v[slot] = 3.0
    return v


def _img(db, sha: str, emb) -> ImageRecord:
    img = ImageRecord(
        path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1, mime="image/jpeg",
        width=1, height=1, origin="imported_filesystem",
        integrity_status="unknown", siglip_embedding=emb,
    )
    db.add(img)
    db.flush()
    return img


def _head(db, tag_id: int, slot: int, *, weight=1.0):
    # weight 3.0 → score sigmoid(3)=0.95 clears the 0.90 presentation floor;
    # weight 1.0 → sigmoid(1)=0.73 clears the 0.50 conflict floor.
    s = db.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
    w = [0.0] * 1152
    w[slot] = weight
    db.add(TagHead(
        tag_id=tag_id, embedding_version=s.embedder_model_version,
        weights=w, bias=0.0, suggest_threshold=0.5, auto_apply_threshold=0.5,
        n_pos=60, n_neg=90, ap=0.9, precision_cv=0.98, recall=0.7,
    ))


def _system_tag(db, name):
    return db.execute(
        select(Tag).where(Tag.is_system.is_(True), Tag.name == name)
    ).scalar_one()


def _source(db, image_id, tag_id):
    return db.execute(
        select(image_tag.c.source)
        .where(image_tag.c.image_record_id == image_id)
        .where(image_tag.c.tag_id == tag_id)
    ).scalar_one_or_none()


def test_presentation_sweep_hides_chrome(db_sync):
    banner = _system_tag(db_sync, "banner")
    _head(db_sync, banner.id, 0, weight=3.0)
    img = _img(db_sync, "a" * 64, _emb(0))
    db_sync.commit()
    res = system_tag_auto_apply_sweep(db_sync, mode="chrome")
    assert res["n_applied"] == 1
    assert _source(db_sync, img.id, banner.id) == "presentation_auto"


def test_presentation_sweep_hard_skips_valued_image(db_sync):
    # An image the operator already tagged with a content tag is never auto-hidden.
    banner = _system_tag(db_sync, "banner")
    _head(db_sync, banner.id, 0, weight=3.0)
    content = Tag(name="mychar", kind=TagKind.character)
    db_sync.add(content)
    db_sync.flush()
    img = _img(db_sync, "b" * 64, _emb(0))
    db_sync.execute(image_tag.insert().values(
        image_record_id=img.id, tag_id=content.id, source="manual"))
    db_sync.commit()
    res = system_tag_auto_apply_sweep(db_sync, mode="chrome")
    assert res["n_applied"] == 0
    assert _source(db_sync, img.id, banner.id) is None


def test_presentation_sweep_flags_conflict(db_sync):
    # Matches banner AND scores high on a content head → hidden but flagged with
    # that content tag as the conflict.
    banner = _system_tag(db_sync, "banner")
    _head(db_sync, banner.id, 0, weight=3.0)
    content = Tag(name="looksreal", kind=TagKind.general)
    db_sync.add(content)
    db_sync.flush()
    _head(db_sync, content.id, 0, weight=1.0)     # content head also fires
    img = _img(db_sync, "c" * 64, _emb(0))
    db_sync.commit()
    res = system_tag_auto_apply_sweep(db_sync, mode="chrome")
    assert res["n_applied"] == 1
    assert res["n_flagged"] == 1
    assert _source(db_sync, img.id, banner.id) == "presentation_auto"
    flag = db_sync.execute(
        select(PresentationReview).where(
            PresentationReview.image_record_id == img.id,
            PresentationReview.tag_id == banner.id,
        )
    ).scalar_one()
    assert flag.conflict_tag_id == content.id
    assert flag.conflict_score >= 0.5


def test_presentation_sweep_disabled_is_noop(db_sync):
    s = db_sync.execute(select(MLSettings).where(MLSettings.id == 1)).scalar_one()
    s.presentation_auto_apply_enabled = False
    banner = _system_tag(db_sync, "banner")
    _head(db_sync, banner.id, 0, weight=3.0)
    img = _img(db_sync, "d" * 64, _emb(0))
    db_sync.commit()
    res = system_tag_auto_apply_sweep(db_sync, mode="chrome")
    assert res["n_applied"] == 0
    assert _source(db_sync, img.id, banner.id) is None


def test_presentation_sweep_ignores_wip(db_sync):
    # wip is a system tag but NOT presentation chrome → never auto-applied.
    wip = _system_tag(db_sync, "wip")
    _head(db_sync, wip.id, 0, weight=3.0)
    img = _img(db_sync, "e" * 64, _emb(0))
    db_sync.commit()
    res = system_tag_auto_apply_sweep(db_sync, mode="chrome")
    assert res["n_applied"] == 0
    assert _source(db_sync, img.id, wip.id) is None


def test_content_sweep_never_fires_system_tags(db_sync):
    # A graduated banner (system) head must NOT auto-apply via the content path.
    banner = _system_tag(db_sync, "banner")
    _head(db_sync, banner.id, 0, weight=3.0)
    img = _img(db_sync, "f" * 64, _emb(0))
    run = HeadAutoApplyRun(dry_run=False, params={}, status="running")
    db_sync.add(run)
    db_sync.flush()
    db_sync.commit()
    auto_apply_sweep(db_sync, run, dry_run=False)
    assert _source(db_sync, img.id, banner.id) is None