bvandeusenandClaude Opus 5.5
ff70f837d0
CI and images / lint (push) Successful in 2s
CI and images / extension-version (push) Successful in 2s
CI and images / frontend-build (push) Successful in 20s
CI and images / backend-lint-and-test (push) Successful in 30s
CI and images / integration (push) Successful in 2m18s
CI and images / sign-extension (push) Successful in 3s
CI and images / build-agent (push) Successful in 6s
CI and images / build-web (push) Successful in 1m44s
CI and images / smoke-web (push) Successful in 56s
CI and images / promote (push) Skipped
tests/factories.py holds image_row/make_image/make_image_async/make_tag. The 17 byte-identical _img/_tag helpers (15 modules) now import them under their old names, so no call site changed. frontend/test/support/stubFetch.js replaces 15 copies that differed only in formatting. Copies whose bodies differ (other defaults, other columns, a url-only stub) are left as they are; folding those needs a look at each caller. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR
99 lines
3.5 KiB
Python
99 lines
3.5 KiB
Python
"""Auto-apply observability (#114): misfire/under-fire counters captured on
|
|
operator corrections, the daily snapshot time-series, and the metrics API."""
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import HeadMetric, HeadMetricsSnapshot, ImageRecord, TagHead, TagKind
|
|
from backend.app.models.tag import image_tag
|
|
from backend.app.services.tag_service import TagService
|
|
from tests.factories import make_image_async as _img
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
def _head(tag_id):
|
|
return TagHead(
|
|
tag_id=tag_id, embedding_version="siglip-test", weights=[0.0] * 1152,
|
|
bias=0.0, suggest_threshold=0.5, auto_apply_threshold=0.6,
|
|
n_pos=30, n_neg=90, ap=0.9, precision_cv=0.95, recall=0.7,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_removing_head_auto_tag_counts_misfire(db):
|
|
img = await _img(db, "a" * 64)
|
|
tag = await TagService(db).find_or_create("misfire", TagKind.general)
|
|
await db.execute(image_tag.insert().values(
|
|
image_record_id=img.id, tag_id=tag.id, source="head_auto",
|
|
))
|
|
await db.commit()
|
|
await TagService(db).remove_from_image(img.id, tag.id)
|
|
await db.commit()
|
|
m = await db.get(HeadMetric, tag.id)
|
|
assert m is not None and m.n_misfires == 1 and m.n_underfires == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_removing_manual_tag_is_not_a_misfire(db):
|
|
img = await _img(db, "b" * 64)
|
|
tag = await TagService(db).find_or_create("manualrm", TagKind.general)
|
|
await db.execute(image_tag.insert().values(
|
|
image_record_id=img.id, tag_id=tag.id, source="manual",
|
|
))
|
|
await db.commit()
|
|
await TagService(db).remove_from_image(img.id, tag.id)
|
|
await db.commit()
|
|
assert await db.get(HeadMetric, tag.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manual_add_with_head_counts_underfire(db):
|
|
img = await _img(db, "c" * 64)
|
|
tag = await TagService(db).find_or_create("underfire", TagKind.general)
|
|
db.add(_head(tag.id))
|
|
await db.commit()
|
|
await TagService(db).add_to_image(img.id, tag.id, source="manual")
|
|
await db.commit()
|
|
m = await db.get(HeadMetric, tag.id)
|
|
assert m is not None and m.n_underfires == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manual_add_without_head_no_underfire(db):
|
|
img = await _img(db, "d" * 64)
|
|
tag = await TagService(db).find_or_create("nohead", TagKind.general)
|
|
await db.commit()
|
|
await TagService(db).add_to_image(img.id, tag.id, source="manual")
|
|
await db.commit()
|
|
assert await db.get(HeadMetric, tag.id) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_records_timeseries_point(db):
|
|
tag = await TagService(db).find_or_create("snap", TagKind.general)
|
|
db.add(_head(tag.id))
|
|
await db.commit()
|
|
from backend.app.tasks.maintenance import snapshot_head_metrics
|
|
|
|
n = snapshot_head_metrics() # sync task, own session
|
|
assert n >= 1
|
|
snaps = (await db.execute(
|
|
select(HeadMetricsSnapshot).where(HeadMetricsSnapshot.tag_id == tag.id)
|
|
)).scalars().all()
|
|
assert len(snaps) == 1
|
|
assert snaps[0].name == "snap"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metrics_api_returns_concept(client, db):
|
|
tag = await TagService(db).find_or_create("apimetric", TagKind.general)
|
|
db.add(_head(tag.id))
|
|
await db.commit()
|
|
resp = await client.get("/api/heads/metrics")
|
|
assert resp.status_code == 200
|
|
body = await resp.get_json()
|
|
c = next(x for x in body["concepts"] if x["name"] == "apimetric")
|
|
assert c["auto_apply"] is True
|
|
assert c["n_misfires"] == 0
|
|
assert "snapshots" in body
|