feat(provenance): ProvenanceService.for_image (peer provenance rows)

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

tests/test_provenance_service.py

@@ -0,0 +1,131 @@
+from datetime import UTC, datetime
+
+import pytest
+
+from backend.app.models import (
+    Artist,
+    ImageProvenance,
+    ImageRecord,
+    Post,
+    Source,
+)
+from backend.app.services.provenance_service import ProvenanceService
+
+pytestmark = pytest.mark.integration
+
+
+async def _seed_image(db, sha="a" + "0" * 63) -> ImageRecord:
+    rec = ImageRecord(
+        path=f"/images/test/{sha}.jpg",
+        sha256=sha,
+        size_bytes=1, mime="image/jpeg", width=1, height=1,
+        origin="imported_filesystem", integrity_status="unknown",
+    )
+    db.add(rec)
+    await db.flush()
+    return rec
+
+
+async def _seed_post(db, *, artist_name, slug, platform, ext_id,
+                     title=None, desc=None, count=None) -> tuple:
+    artist = Artist(name=artist_name, slug=slug)
+    db.add(artist)
+    await db.flush()
+    source = Source(artist_id=artist.id, platform=platform,
+                    url=f"https://{platform}.test/{slug}")
+    db.add(source)
+    await db.flush()
+    post = Post(
+        source_id=source.id, external_post_id=ext_id,
+        post_url=f"https://{platform}.test/p/{ext_id}",
+        post_title=title, post_date=datetime(2023, 8, 1, tzinfo=UTC),
+        description=desc, attachment_count=count,
+    )
+    db.add(post)
+    await db.flush()
+    return artist, source, post
+
+
+@pytest.mark.asyncio
+async def test_for_image_missing_returns_none(db):
+    svc = ProvenanceService(db)
+    assert await svc.for_image(999999) is None
+
+
+@pytest.mark.asyncio
+async def test_for_image_no_provenance_returns_empty_list(db):
+    rec = await _seed_image(db)
+    svc = ProvenanceService(db)
+    payload = await svc.for_image(rec.id)
+    assert payload == {"image_id": rec.id, "provenance": []}
+
+
+@pytest.mark.asyncio
+async def test_for_image_single_provenance_full_shape(db):
+    rec = await _seed_image(db)
+    artist, source, post = await _seed_post(
+        db, artist_name="Alice", slug="alice", platform="patreon",
+        ext_id="555", title="Set 1", desc="<p>hi</p><script>x</script>",
+        count=2,
+    )
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
+                           source_id=source.id))
+    await db.flush()
+
+    svc = ProvenanceService(db)
+    payload = await svc.for_image(rec.id)
+    assert payload["image_id"] == rec.id
+    assert len(payload["provenance"]) == 1
+    e = payload["provenance"][0]
+    assert e["post"]["id"] == post.id
+    assert e["post"]["external_post_id"] == "555"
+    assert e["post"]["title"] == "Set 1"
+    assert e["post"]["attachment_count"] == 2
+    assert e["post"]["description_html"] == "<p>hi</p>"  # script removed
+    assert e["post"]["url"] == "https://patreon.test/p/555"
+    assert e["post"]["date"].startswith("2023-08-01")
+    assert e["source"] == {"id": source.id, "platform": "patreon",
+                           "url": source.url}
+    assert e["artist"] == {"id": artist.id, "name": "Alice",
+                           "slug": "alice"}
+    assert e["provenance_id"] is not None
+    assert e["captured_at"] is not None
+
+
+@pytest.mark.asyncio
+async def test_for_image_multiple_provenance_peer_ordering(db):
+    rec = await _seed_image(db)
+    _, s1, p1 = await _seed_post(db, artist_name="A1", slug="a1",
+                                 platform="patreon", ext_id="1")
+    _, s2, p2 = await _seed_post(db, artist_name="A2", slug="a2",
+                                 platform="fanbox", ext_id="2")
+    ip1 = ImageProvenance(image_record_id=rec.id, post_id=p1.id,
+                          source_id=s1.id)
+    ip2 = ImageProvenance(image_record_id=rec.id, post_id=p2.id,
+                          source_id=s2.id)
+    db.add(ip1)
+    await db.flush()
+    db.add(ip2)
+    await db.flush()
+
+    svc = ProvenanceService(db)
+    payload = await svc.for_image(rec.id)
+    ids = [e["provenance_id"] for e in payload["provenance"]]
+    assert ids == sorted(ids)  # captured_at, id ascending → insertion order
+    assert len(payload["provenance"]) == 2
+
+
+@pytest.mark.asyncio
+async def test_for_image_null_post_fields_serialize_null(db):
+    rec = await _seed_image(db)
+    _, source, post = await _seed_post(
+        db, artist_name="Bob", slug="bob", platform="x", ext_id="9",
+    )  # title/desc/count/post_url default-ish
+    db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
+                           source_id=source.id))
+    await db.flush()
+    svc = ProvenanceService(db)
+    e = (await svc.for_image(rec.id))["provenance"][0]
+    assert e["post"]["title"] is None
+    assert e["post"]["description_html"] is None
+    assert e["post"]["attachment_count"] is None