bvandeusen
5269cd0709
Images pulled out of a .zip/.rar previously kept no record of WHICH archive
they came from — the member->archive link was computed during extraction and
discarded, leaving only image->post. So the provenance modal could only scope
attachments to the whole post, showing every archive a 'High Resolution Files'
bundle carried instead of the one a given file lives in.
- ImageProvenance.from_attachment_id: nullable FK -> post_attachment.id
(SET NULL), migration 0055.
- importer: _import_archive stamps from_attachment_id on every member's
provenance row for the post (new + superseded + deduped members), resolving
the archive's own PostAttachment by (post, sha). Post-pass UPDATE, NULL-only
and idempotent, so it doesn't touch the dedup/supersede branches and the
backfill is safe to re-run. Nested members link to the outer stored archive.
- provenance_service.for_image: when the originating post's provenance row
records from_attachment_id, return ONLY that archive; else fall back to the
primary-post scoping from 068def2.
- ProvenancePanel: heading pluralizes ('Attachment' for a single file).
- Backfill: re-running reextract_archive_attachments (ArchiveReextractCard)
routes through _import_archive and stamps existing rows — no new code.
Tests: capture stamps on fresh import, nested-archive attribution, per-post
archive on dedup; for_image filters to the containing archive; reextract
backfill stamps the link.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
273 lines
10 KiB
Python
273 lines
10 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_filtered_to_containing_archive(db):
|
|
# Milestone #87: when the originating post's provenance row records which
|
|
# archive the file came from, show ONLY that archive — not the post's other
|
|
# attachments.
|
|
rec = await _seed_image(db)
|
|
a1, s1, post = await _seed_post(db, artist_name="Arc", slug="arc",
|
|
platform="patreon", ext_id="500")
|
|
rec.primary_post_id = post.id
|
|
att_in = PostAttachment(
|
|
post_id=post.id, artist_id=a1.id, sha256="e" + "0" * 63,
|
|
path="/images/attachments/e00/in.cbz", original_filename="in.cbz",
|
|
ext=".cbz", mime="application/zip", size_bytes=9,
|
|
)
|
|
att_other = PostAttachment(
|
|
post_id=post.id, artist_id=a1.id, sha256="f" + "0" * 63,
|
|
path="/images/attachments/f00/other.rar", original_filename="other.rar",
|
|
ext=".rar", mime="application/x-rar", size_bytes=9,
|
|
)
|
|
db.add(att_in)
|
|
db.add(att_other)
|
|
await db.flush()
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
|
|
source_id=s1.id, from_attachment_id=att_in.id))
|
|
await db.flush()
|
|
|
|
payload = await ProvenanceService(db).for_image(rec.id)
|
|
names = [a["original_filename"] for a in payload["attachments"]]
|
|
assert names == ["in.cbz"]
|
|
|
|
|
|
@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"
|
|
)
|