bvandeusenandClaude Opus 5
6f5ea5d1d3
CI / lint (push) Successful in 8s
CI / extension-version (push) Successful in 9s
Build images / sign-extension (push) Successful in 9s
Build images / build-agent (push) Successful in 17s
CI / frontend-build (push) Successful in 35s
CI / backend-lint-and-test (push) Successful in 1m18s
Build images / build-web (push) Successful in 1m22s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 2m19s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m38s
Two surfaces render a post's HTML body. get_post did _localize_inline_images(sanitize_post_html(...)); provenance_service._post_dict called sanitize_post_html alone. So opening the Provenance panel fetched images from Patreon's CDN for files FC had already downloaded - the archive reaching out to the platform to display what it had archived, which is the thing 830 Phase 2 set out to stop. The same bodies also break when a CDN URL expires or a post is removed, while the identical local copy sits unused. The fix is the shape, not the call. Those were two separately-callable halves and only the first looked mandatory, so a second caller was always going to do half of it. render_post_body in the new services/post_body.py is the whole pipeline in one call, and sanitizing-without-localizing is no longer a reachable operation. _localize_inline_images moves there verbatim; post_feed_service loses five imports that went with it. _post_dict becomes async and takes the session. Both call sites are already inside async methods, so for_image's list comprehension awaits per entry - fine, because localization issues ZERO queries for a body with no inline <img>, which is most of them. Recorded that early exit in the module so the loop isn't "optimized" into a batch without a measurement. Deliberately NOT fixed: provenance still names the same columns url/title/date where the feed says post_url/post_title/post_date, and its description_translated is full text where the feed truncates to DESCRIPTION_LIMIT. That is 3965's wider half - a breaking payload change for ProvenancePanel with no second reason to spend it today. Noted in _post_dict's docstring so the next reader knows it was seen and left. Four regression tests on the provenance path, covering both entry points plus the two refusals the feed already pins: an uncaptured image stays hotlinked (a broken local path is worse than an intact remote one), and a filehash owned by another artist never leaks in. Reverting render_post_body to a bare sanitize fails the first two. Recorded as snippet 3968. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SHQB1YukL3VyvMK8rcbmV9
367 lines
14 KiB
Python
367 lines
14 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"
|
|
)
|
|
|
|
|
|
# --- #3965: the body pipeline is shared with the post detail view ----------
|
|
#
|
|
# These are the regression guards for the defect: provenance used to call
|
|
# `sanitize_post_html` alone, so it served CDN-hotlinking bodies while
|
|
# `get_post` served local copies of the very same images. Reverting
|
|
# `render_post_body` back to a bare sanitize fails both.
|
|
|
|
|
|
async def _seed_localizable(db, *, slug, filehash):
|
|
"""A post whose body inlines a CDN image FC has already downloaded."""
|
|
cdn = f"https://cdn.test/p/{filehash}/inline.png"
|
|
artist, source, post = await _seed_post(
|
|
db, artist_name=slug.title(), slug=slug, platform="patreon",
|
|
ext_id=f"{slug}-1", title="T", desc=f'<p>x</p><img src="{cdn}">',
|
|
)
|
|
db.add(ImageRecord(
|
|
path=f"/images/{slug}/inline.png", sha256=f"{post.id:064d}",
|
|
size_bytes=10, mime="image/png", width=10, height=10,
|
|
origin="downloaded", primary_post_id=post.id, artist_id=artist.id,
|
|
source_url=cdn, source_filehash=filehash,
|
|
))
|
|
await db.flush()
|
|
return source, post, cdn
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_post_serves_our_own_copy_of_an_inline_image(db):
|
|
"""The archive must not fetch from the platform to show what it archived."""
|
|
_source, post, cdn = await _seed_localizable(
|
|
db, slug="localpost", filehash="0123456789abcdef0123456789abcdef",
|
|
)
|
|
payload = await ProvenanceService(db).for_post(post.id)
|
|
html = payload["post"]["description_html"]
|
|
assert 'src="/images/localpost/inline.png"' in html
|
|
assert cdn not in html, "provenance body still hotlinks the platform CDN"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_for_image_serves_our_own_copy_of_an_inline_image(db):
|
|
"""`for_image` shapes N posts, not one — the loop localizes every entry."""
|
|
rec = await _seed_image(db, sha="b" + "0" * 63)
|
|
_source, post, cdn = await _seed_localizable(
|
|
db, slug="localimg", filehash="abcdef0123456789abcdef0123456789",
|
|
)
|
|
db.add(ImageProvenance(image_record_id=rec.id, post_id=post.id,
|
|
source_id=post.source_id))
|
|
await db.flush()
|
|
|
|
payload = await ProvenanceService(db).for_image(rec.id)
|
|
html = payload["provenance"][0]["post"]["description_html"]
|
|
assert 'src="/images/localimg/inline.png"' in html
|
|
assert cdn not in html
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_uncaptured_inline_image_is_left_hotlinking(db):
|
|
"""Localization only redirects what we actually hold. No stored image means
|
|
the original src stands — a broken local path would be worse than a
|
|
hotlink, and this is the behaviour `get_post` already has."""
|
|
_artist, _source, post = await _seed_post(
|
|
db, artist_name="Nope", slug="nope", platform="patreon",
|
|
ext_id="nope-1", desc='<img src="https://cdn.test/p/'
|
|
'ffffffffffffffffffffffffffffffff/gone.png">',
|
|
)
|
|
payload = await ProvenanceService(db).for_post(post.id)
|
|
assert "cdn.test" in payload["post"]["description_html"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_localization_is_artist_scoped_in_provenance_too(db):
|
|
"""A filehash owned by a DIFFERENT artist must not leak in — the same
|
|
scoping `get_post` enforces, asserted on this path so the shared helper
|
|
cannot be loosened for one caller without failing for the other."""
|
|
filehash = "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd"
|
|
cdn = f"https://cdn.test/p/{filehash}/x.png"
|
|
_a1, _s1, post = await _seed_post(
|
|
db, artist_name="Owner", slug="owner-prov", platform="patreon",
|
|
ext_id="own-1", desc=f'<img src="{cdn}">',
|
|
)
|
|
other = Artist(name="Other", slug="other-prov")
|
|
db.add(other)
|
|
await db.flush()
|
|
db.add(ImageRecord(
|
|
path="/images/other-prov/x.png", sha256=f"{post.id:064d}",
|
|
size_bytes=10, mime="image/png", width=10, height=10,
|
|
origin="downloaded", artist_id=other.id,
|
|
source_url=cdn, source_filehash=filehash,
|
|
))
|
|
await db.flush()
|
|
|
|
payload = await ProvenanceService(db).for_post(post.id)
|
|
assert cdn in payload["post"]["description_html"]
|