feat(provenance): post_id/artist_id gallery filters via EXISTS (mutually exclusive)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/test_gallery_provenance_filter.py

@@ -0,0 +1,130 @@
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from backend.app import create_app
+from backend.app.models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    Post,
+    Source,
+)
+from backend.app.services.gallery_service import GalleryService
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def app():
+    return create_app()
+
+
+@pytest.fixture
+async def client(app):
+    async with app.test_client() as c:
+        yield c
+
+
+async def _img(db, n):
+    base = datetime.now(UTC)
+    rec = ImageRecord(
+        path=f"/images/f/{n}.jpg", sha256=f"f{n:063d}",
+        size_bytes=1, mime="image/jpeg", width=1, height=1,
+        origin="imported_filesystem", integrity_status="unknown",
+    )
+    rec.created_at = base - timedelta(minutes=n)
+    db.add(rec)
+    await db.flush()
+    return rec
+
+
+async def _post(db, artist_name, slug, ext):
+    a = Artist(name=artist_name, slug=slug)
+    db.add(a)
+    await db.flush()
+    s = Source(artist_id=a.id, platform="patreon",
+               url=f"https://patreon.test/{slug}")
+    db.add(s)
+    await db.flush()
+    p = Post(source_id=s.id, external_post_id=ext)
+    db.add(p)
+    await db.flush()
+    return a, s, p
+
+
+@pytest.mark.asyncio
+async def test_scroll_post_id_filter(db):
+    i1, i2, i3 = await _img(db, 1), await _img(db, 2), await _img(db, 3)
+    _, s, p = await _post(db, "A", "a", "10")
+    db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id,
+                           source_id=s.id))
+    db.add(ImageProvenance(image_record_id=i2.id, post_id=p.id,
+                           source_id=s.id))
+    await db.flush()
+    svc = GalleryService(db)
+    page = await svc.scroll(cursor=None, limit=10, post_id=p.id)
+    assert {x.id for x in page.images} == {i1.id, i2.id}
+
+
+@pytest.mark.asyncio
+async def test_scroll_post_id_dedups_multi_rows(db):
+    i1 = await _img(db, 1)
+    _, s, p = await _post(db, "A", "a", "10")
+    # two provenance rows, same image+post (enrich-on-duplicate shape)
+    db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id,
+                           source_id=s.id))
+    await db.flush()
+    db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id,
+                           source_id=s.id))
+    await db.flush()
+    svc = GalleryService(db)
+    page = await svc.scroll(cursor=None, limit=10, post_id=p.id)
+    assert [x.id for x in page.images] == [i1.id]  # appears once
+
+
+@pytest.mark.asyncio
+async def test_scroll_artist_id_filter(db):
+    i1, i2 = await _img(db, 1), await _img(db, 2)
+    a, s, p = await _post(db, "A", "a", "10")
+    db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id,
+                           source_id=s.id))
+    await db.flush()
+    svc = GalleryService(db)
+    page = await svc.scroll(cursor=None, limit=10, artist_id=a.id)
+    assert [x.id for x in page.images] == [i1.id]
+
+
+@pytest.mark.asyncio
+async def test_mutually_exclusive_filters_raise(db):
+    svc = GalleryService(db)
+    with pytest.raises(ValueError):
+        await svc.scroll(cursor=None, limit=10, tag_id=1, post_id=2)
+
+
+@pytest.mark.asyncio
+async def test_timeline_and_jump_accept_post_id(db):
+    i1 = await _img(db, 1)
+    _, s, p = await _post(db, "A", "a", "10")
+    db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id,
+                           source_id=s.id))
+    await db.flush()
+    svc = GalleryService(db)
+    buckets = await svc.timeline(post_id=p.id)
+    assert sum(b.count for b in buckets) == 1
+    cur = await svc.jump_cursor(
+        year=i1.created_at.year, month=i1.created_at.month, post_id=p.id
+    )
+    assert cur is not None
+
+
+@pytest.mark.asyncio
+async def test_api_scroll_rejects_combined_filters(client):
+    resp = await client.get("/api/gallery/scroll?tag_id=1&post_id=2")
+    assert resp.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_api_timeline_rejects_combined_filters(client):
+    resp = await client.get("/api/gallery/timeline?post_id=1&artist_id=2")
+    assert resp.status_code == 400