feat(tags): TagMergeConflict + provenance helper, enrich rename collision
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
import pytest
|
||||
|
||||
from backend.app.models import Tag, TagKind, image_tag
|
||||
from backend.app.models.tag_allowlist import TagAllowlist
|
||||
from backend.app.services.tag_service import (
|
||||
TagMergeConflict,
|
||||
TagService,
|
||||
TagValidationError,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
_IMG_SEQ = 0
|
||||
|
||||
|
||||
async def _img(db):
|
||||
"""Insert a minimal valid ImageRecord (all NOT NULL cols satisfied:
|
||||
path, sha256, size_bytes, mime, origin) and return its id."""
|
||||
global _IMG_SEQ
|
||||
_IMG_SEQ += 1
|
||||
from backend.app.models import ImageRecord
|
||||
|
||||
rec = ImageRecord(
|
||||
path=f"/tmp/fc_merge_{_IMG_SEQ}.png",
|
||||
sha256=f"{_IMG_SEQ:064d}",
|
||||
size_bytes=1,
|
||||
mime="image/png",
|
||||
origin="uploaded",
|
||||
)
|
||||
db.add(rec)
|
||||
await db.flush()
|
||||
return rec.id
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rename_collision_raises_merge_conflict_with_payload(db):
|
||||
svc = TagService(db)
|
||||
target = await svc.find_or_create("Existing", TagKind.character)
|
||||
source = await svc.find_or_create("Dupe", TagKind.character)
|
||||
img = await _img(db)
|
||||
await svc.add_to_image(img, source.id, source="manual")
|
||||
|
||||
with pytest.raises(TagMergeConflict) as ei:
|
||||
await svc.rename(source.id, "Existing")
|
||||
exc = ei.value
|
||||
assert exc.target_id == target.id
|
||||
assert exc.target_name == "Existing"
|
||||
assert exc.source_image_count == 1
|
||||
assert exc.will_alias is False # purely manual, ML-unknown
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_merge_conflict_is_a_tag_validation_error(db):
|
||||
assert issubclass(TagMergeConflict, TagValidationError)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_will_alias_true_when_machine_sourced(db):
|
||||
svc = TagService(db)
|
||||
await svc.find_or_create("Canon", TagKind.general)
|
||||
source = await svc.find_or_create("Auto", TagKind.general)
|
||||
img = await _img(db)
|
||||
await svc.add_to_image(img, source.id, source="ml_auto")
|
||||
with pytest.raises(TagMergeConflict) as ei:
|
||||
await svc.rename(source.id, "Canon")
|
||||
assert ei.value.will_alias is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_will_alias_true_when_allowlisted(db):
|
||||
svc = TagService(db)
|
||||
await svc.find_or_create("Canon2", TagKind.general)
|
||||
source = await svc.find_or_create("Allowed", TagKind.general)
|
||||
db.add(TagAllowlist(tag_id=source.id))
|
||||
await db.flush()
|
||||
with pytest.raises(TagMergeConflict) as ei:
|
||||
await svc.rename(source.id, "Canon2")
|
||||
assert ei.value.will_alias is True
|
||||
Reference in New Issue
Block a user