bvandeusen
068def2f24
The Attachments section aggregated PostAttachment rows across EVERY post an image was pHash-linked to. When one of those was a 'High Resolution Files' mega-bundle (dozens of unrelated archives), the list ballooned past the viewport and overwhelmed the modal's right rail. - for_image() now scopes attachments to ImageRecord.primary_post_id (the post the file was actually captured from), falling back to all linked posts only when primary_post_id is unset (older rows / filesystem imports). - ProvenancePanel wraps the list in a max-height scroll container with a count in the heading, mirroring the cards' independent-scroll treatment. Note: FC stores archives as opaque blobs and never records which archive an extracted image came from, so attachments can't yet be scoped tighter than the post. Capturing image->archive containment is tracked as separate work. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
242 lines
9.0 KiB
Python
242 lines
9.0 KiB
Python
from datetime import UTC, datetime
|
|
|
|
import pytest
|
|
|
|
from backend.app.models import (
|
|
Artist,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
Post,
|
|
PostAttachment,
|
|
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, artist_id=artist.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": [], "attachments": []
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_attachments_scoped_to_primary_post(db):
|
|
# Image linked to TWO posts; only the primary (originating) post's
|
|
# attachments should surface — not the other post's mega-bundle.
|
|
rec = await _seed_image(db)
|
|
a1, s1, primary = await _seed_post(db, artist_name="P1", slug="p1",
|
|
platform="patreon", ext_id="100")
|
|
a2, s2, bundle = await _seed_post(db, artist_name="P2", slug="p2",
|
|
platform="patreon", ext_id="200")
|
|
rec.primary_post_id = primary.id
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=primary.id,
|
|
source_id=s1.id))
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=bundle.id,
|
|
source_id=s2.id))
|
|
db.add(PostAttachment(
|
|
post_id=primary.id, artist_id=a1.id, sha256="p" + "0" * 63,
|
|
path="/images/attachments/p00/keep.zip", original_filename="keep.zip",
|
|
ext=".zip", mime="application/zip", size_bytes=9,
|
|
))
|
|
db.add(PostAttachment(
|
|
post_id=bundle.id, artist_id=a2.id, sha256="b" + "0" * 63,
|
|
path="/images/attachments/b00/drop.rar", original_filename="drop.rar",
|
|
ext=".rar", mime="application/x-rar", size_bytes=9,
|
|
))
|
|
await db.flush()
|
|
|
|
payload = await ProvenanceService(db).for_image(rec.id)
|
|
names = [a["original_filename"] for a in payload["attachments"]]
|
|
assert names == ["keep.zip"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_attachments_fallback_to_all_posts_when_no_primary(db):
|
|
# No primary_post_id (older rows / filesystem imports) → preserve the
|
|
# aggregate-across-linked-posts behavior so attachments aren't lost.
|
|
rec = await _seed_image(db)
|
|
a1, s1, p1 = await _seed_post(db, artist_name="F1", slug="f1",
|
|
platform="patreon", ext_id="300")
|
|
a2, s2, p2 = await _seed_post(db, artist_name="F2", slug="f2",
|
|
platform="patreon", ext_id="400")
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=p1.id,
|
|
source_id=s1.id))
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=p2.id,
|
|
source_id=s2.id))
|
|
db.add(PostAttachment(
|
|
post_id=p1.id, artist_id=a1.id, sha256="c" + "0" * 63,
|
|
path="/images/attachments/c00/one.zip", original_filename="one.zip",
|
|
ext=".zip", mime="application/zip", size_bytes=9,
|
|
))
|
|
db.add(PostAttachment(
|
|
post_id=p2.id, artist_id=a2.id, sha256="d" + "0" * 63,
|
|
path="/images/attachments/d00/two.zip", original_filename="two.zip",
|
|
ext=".zip", mime="application/zip", size_bytes=9,
|
|
))
|
|
await db.flush()
|
|
|
|
payload = await ProvenanceService(db).for_image(rec.id)
|
|
names = sorted(a["original_filename"] for a in payload["attachments"])
|
|
assert names == ["one.zip", "two.zip"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_missing_returns_none(db):
|
|
svc = ProvenanceService(db)
|
|
assert await svc.for_post(999999) is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_returns_post_source_artist(db):
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name="Carol", slug="carol", platform="patreon",
|
|
ext_id="77", title="T", desc="<p>d</p>", count=3,
|
|
)
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_post(post.id)
|
|
assert payload["post"]["id"] == post.id
|
|
assert payload["post"]["title"] == "T"
|
|
assert payload["post"]["description_html"] == "<p>d</p>"
|
|
assert payload["post"]["attachment_count"] == 3
|
|
assert payload["source"] == {"id": source.id, "platform": "patreon",
|
|
"url": source.url}
|
|
assert payload["artist"] == {"id": artist.id, "name": "Carol",
|
|
"slug": "carol"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_includes_attachments(db):
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name="Att", slug="att", platform="patreon",
|
|
ext_id="55",
|
|
)
|
|
db.add(PostAttachment(
|
|
post_id=post.id, artist_id=artist.id, sha256="t" + "0" * 63,
|
|
path="/images/attachments/t00/t.zip", original_filename="t.zip",
|
|
ext=".zip", mime="application/zip", size_bytes=9,
|
|
))
|
|
await db.flush()
|
|
svc = ProvenanceService(db)
|
|
payload = await svc.for_post(post.id)
|
|
assert len(payload["attachments"]) == 1
|
|
att = payload["attachments"][0]
|
|
assert att["original_filename"] == "t.zip"
|
|
assert att["download_url"].endswith(
|
|
f"/api/attachments/{att['id']}/download"
|
|
)
|