feat(fc2b): add AllowlistService + TagService.rename

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>
This commit is contained in:
2026-05-15 07:39:33 -04:00
parent 7860b86a13
commit ed92548c0f
4 changed files with 309 additions and 0 deletions
+106
View File
@@ -0,0 +1,106 @@
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
+41
View File
@@ -0,0 +1,41 @@
import pytest
from backend.app.models import TagKind
from backend.app.services.tag_service import TagService, TagValidationError
pytestmark = pytest.mark.integration
@pytest.mark.asyncio
async def test_rename_success(db):
svc = TagService(db)
t = await svc.find_or_create("uchiha_sasuke", TagKind.character)
renamed = await svc.rename(t.id, "Sasuke Uchiha")
assert renamed.id == t.id
assert renamed.name == "Sasuke Uchiha"
@pytest.mark.asyncio
async def test_rename_collision_raises(db):
svc = TagService(db)
await svc.find_or_create("Existing", TagKind.character)
other = await svc.find_or_create("Other", TagKind.character)
with pytest.raises(TagValidationError, match="already exists"):
await svc.rename(other.id, "Existing")
@pytest.mark.asyncio
async def test_rename_same_name_different_kind_ok(db):
svc = TagService(db)
await svc.find_or_create("Shared", TagKind.character)
artist = await svc.find_or_create("ArtistTag", TagKind.artist)
renamed = await svc.rename(artist.id, "Shared")
assert renamed.name == "Shared"
@pytest.mark.asyncio
async def test_rename_empty_raises(db):
svc = TagService(db)
t = await svc.find_or_create("Whatever", TagKind.general)
with pytest.raises(TagValidationError):
await svc.rename(t.id, " ")