bvandeusen
cabd73287a
Removing the create_app import left 2 blank lines before pytestmark in 9 files where ruff isort wants 1. Also stripped two pre-existing whitespace-only blank lines surfaced by the file change. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
import pytest
|
|
|
|
from backend.app import create_app
|
|
from backend.app.models import ImageRecord, TagKind
|
|
from backend.app.models.tag import image_tag
|
|
from backend.app.services.ml.suggestions import SuggestionService
|
|
from backend.app.services.tag_service import TagService
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _img(sha: str, predictions: dict) -> ImageRecord:
|
|
return ImageRecord(
|
|
path=f"/images/{sha}.jpg", sha256=sha, size_bytes=1,
|
|
mime="image/jpeg", width=1, height=1,
|
|
origin="imported_filesystem", integrity_status="unknown",
|
|
tagger_predictions=predictions,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consensus_includes_tag_over_threshold(db):
|
|
tags = TagService(db)
|
|
t = await tags.find_or_create("sword", TagKind.general)
|
|
a = _img("a" * 64, {"sword": {"category": "general", "confidence": 0.97}})
|
|
b = _img("b" * 64, {"sword": {"category": "general", "confidence": 0.95}})
|
|
db.add_all([a, b])
|
|
await db.flush()
|
|
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
|
|
gen = res["general"]
|
|
assert any(s["canonical_tag_id"] == t.id for s in gen)
|
|
s = next(s for s in gen if s["canonical_tag_id"] == t.id)
|
|
assert s["coverage"] == 1.0
|
|
assert 0.95 <= s["confidence"] <= 0.97
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consensus_counts_already_applied_for_coverage(db):
|
|
tags = TagService(db)
|
|
t = await tags.find_or_create("sky", TagKind.general)
|
|
a = _img("c" * 64, {"sky": {"category": "general", "confidence": 0.96}})
|
|
b = _img("d" * 64, {}) # no prediction
|
|
db.add_all([a, b])
|
|
await db.flush()
|
|
# b already has the tag applied -> counts toward coverage, not confidence
|
|
await db.execute(
|
|
image_tag.insert().values(
|
|
image_record_id=b.id, tag_id=t.id, source="manual"
|
|
)
|
|
)
|
|
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
|
|
s = next(s for s in res["general"] if s["canonical_tag_id"] == t.id)
|
|
assert s["coverage"] == 1.0 # 1 suggested + 1 applied / 2
|
|
assert s["confidence"] == pytest.approx(0.96, abs=1e-4)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consensus_excludes_below_threshold(db):
|
|
tags = TagService(db)
|
|
await tags.find_or_create("rare", TagKind.general)
|
|
a = _img("e" * 64, {"rare": {"category": "general", "confidence": 0.96}})
|
|
b = _img("f" * 64, {})
|
|
db.add_all([a, b])
|
|
await db.flush()
|
|
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
|
|
assert all(
|
|
s["name"] != "rare" for s in res.get("general", [])
|
|
) # coverage 0.5 < 0.8
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consensus_skips_creates_new_tag(db):
|
|
a = _img("g" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
|
|
b = _img("h" * 64, {"neverseen": {"category": "general", "confidence": 0.99}})
|
|
db.add_all([a, b])
|
|
await db.flush()
|
|
res = await SuggestionService(db).for_selection([a.id, b.id], threshold=0.8)
|
|
# 'neverseen' has no Tag row -> creates_new_tag -> excluded from consensus
|
|
assert all(s["name"] != "neverseen" for s in res.get("general", []))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_consensus_threshold_clamped_and_empty_for_no_ids(db):
|
|
res = await SuggestionService(db).for_selection([], threshold=5.0)
|
|
assert res == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_suggestions_route(db):
|
|
|
|
tags = TagService(db)
|
|
await tags.find_or_create("sword", TagKind.general)
|
|
a = _img("i" * 64, {"sword": {"category": "general", "confidence": 0.97}})
|
|
db.add(a)
|
|
await db.commit()
|
|
app = create_app()
|
|
async with app.test_client() as c:
|
|
resp = await c.post(
|
|
"/api/suggestions/bulk", json={"image_ids": [a.id]}
|
|
)
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
assert body["evaluated"] == 1
|
|
assert body["threshold"] == 0.8
|
|
assert "suggestions" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_bulk_suggestions_requires_ids(db):
|
|
|
|
app = create_app()
|
|
async with app.test_client() as c:
|
|
resp = await c.post("/api/suggestions/bulk", json={})
|
|
assert resp.status_code == 400
|