bvandeusen
ed92548c0f
AllowlistService: accept (apply ml_accepted + add to allowlist + clear rejection; returns whether newly-added so API can kick retro-apply), add_alias_and_accept, dismiss, reject_applied_tag (remove + record rejection so the allowlist won't re-apply), threshold update, remove, list_all. TagService.rename: refuses on (name, kind, fandom_id) collision with a message pointing at FC-2c merge. Tests marked integration. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import TagAllowlist, TagKind, TagSuggestionRejection
|
|
from backend.app.models.tag import image_tag
|
|
from backend.app.services.ml.allowlist import AllowlistService
|
|
from backend.app.services.tag_service import TagService
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
async def _make_image(db):
|
|
from backend.app.models import ImageRecord
|
|
img = ImageRecord(
|
|
path="/images/x.jpg", sha256="x" * 64, size_bytes=1,
|
|
mime="image/jpeg", width=1, height=1,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
)
|
|
db.add(img)
|
|
await db.flush()
|
|
return img
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accept_applies_and_allowlists(db):
|
|
img = await _make_image(db)
|
|
tag = await TagService(db).find_or_create("Hero", TagKind.character)
|
|
svc = AllowlistService(db)
|
|
newly_added = await svc.accept(img.id, tag.id)
|
|
assert newly_added is True
|
|
|
|
applied = (
|
|
await db.execute(
|
|
select(image_tag.c.source)
|
|
.where(image_tag.c.image_record_id == img.id)
|
|
.where(image_tag.c.tag_id == tag.id)
|
|
)
|
|
).scalar_one()
|
|
assert applied == "ml_accepted"
|
|
assert await db.get(TagAllowlist, tag.id) is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accept_idempotent_allowlist(db):
|
|
img = await _make_image(db)
|
|
tag = await TagService(db).find_or_create("Hero2", TagKind.character)
|
|
svc = AllowlistService(db)
|
|
assert await svc.accept(img.id, tag.id) is True
|
|
assert await svc.accept(img.id, tag.id) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_applied_tag_records_rejection(db):
|
|
img = await _make_image(db)
|
|
tag = await TagService(db).find_or_create("Removeme", TagKind.general)
|
|
svc = AllowlistService(db)
|
|
await svc.accept(img.id, tag.id)
|
|
await svc.reject_applied_tag(img.id, tag.id)
|
|
|
|
still_applied = (
|
|
await db.execute(
|
|
select(image_tag.c.tag_id)
|
|
.where(image_tag.c.image_record_id == img.id)
|
|
.where(image_tag.c.tag_id == tag.id)
|
|
)
|
|
).scalar_one_or_none()
|
|
assert still_applied is None
|
|
rej = await db.get(TagSuggestionRejection, (img.id, tag.id))
|
|
assert rej is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dismiss_records_rejection(db):
|
|
img = await _make_image(db)
|
|
tag = await TagService(db).find_or_create("Dismissme", TagKind.general)
|
|
await AllowlistService(db).dismiss(img.id, tag.id)
|
|
assert await db.get(TagSuggestionRejection, (img.id, tag.id)) is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_alias_and_accept(db):
|
|
img = await _make_image(db)
|
|
canonical = await TagService(db).find_or_create(
|
|
"Canonical Char", TagKind.character
|
|
)
|
|
svc = AllowlistService(db)
|
|
await svc.add_alias_and_accept(
|
|
img.id, "model_char_name", "character", canonical.id
|
|
)
|
|
from backend.app.services.ml.aliases import AliasService
|
|
resolved = await AliasService(db).resolve("model_char_name", "character")
|
|
assert resolved.id == canonical.id
|
|
assert await db.get(TagAllowlist, canonical.id) is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_threshold_and_remove(db):
|
|
tag = await TagService(db).find_or_create("Thr", TagKind.general)
|
|
svc = AllowlistService(db)
|
|
img = await _make_image(db)
|
|
await svc.accept(img.id, tag.id)
|
|
await svc.update_threshold(tag.id, 0.80)
|
|
row = await db.get(TagAllowlist, tag.id)
|
|
assert abs(row.min_confidence - 0.80) < 1e-6
|
|
await svc.remove(tag.id)
|
|
assert await db.get(TagAllowlist, tag.id) is None
|