bvandeusen
c342c73a25
Phase 1 of milestone #64. The body is captured (Phase 0) but was shown as plain text. Now: - html_sanitize.py: widen the allowlist to a faithful-but-safe set — headings, inline images, lists, blockquote, hr, code/pre, figure, links (div/span stay stripped; their text is preserved). Benefits the existing ProvenancePanel too. - post_feed_service.get_post: add sanitized `description_html` to the DETAIL response (the feed list stays lightweight plain text by design). - PostCard.vue: render description_html via v-html once expanded (fetched with detail); collapsed + no-detail fallback stay plain text. Styled close to the source (headings, images max-width, accent links, lists, quotes, code). Tests: sanitizer (headings/img/lists survive, img javascript: src dropped); get_post returns sanitized description_html. Refs FC #830. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
444 lines
16 KiB
Python
444 lines
16 KiB
Python
"""FC-3e: PostFeedService unit tests.
|
|
|
|
Covers cursor encode/decode, ordering (coalesce(post_date,
|
|
downloaded_at) DESC, id DESC), artist/platform filters, pagination
|
|
boundary, and the thumbnails/attachments composition.
|
|
"""
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import (
|
|
Artist,
|
|
ImageProvenance,
|
|
ImageRecord,
|
|
Post,
|
|
PostAttachment,
|
|
Source,
|
|
)
|
|
from backend.app.services.post_feed_service import (
|
|
PostFeedService,
|
|
decode_cursor,
|
|
encode_cursor,
|
|
)
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
# --- Cursor round-trip ----------------------------------------------------
|
|
|
|
|
|
def test_cursor_round_trip():
|
|
ts = datetime(2026, 5, 21, 12, 34, 56, tzinfo=UTC)
|
|
c = encode_cursor(ts, 42)
|
|
assert decode_cursor(c) == (ts, 42)
|
|
|
|
|
|
def test_decode_cursor_rejects_garbage():
|
|
with pytest.raises(ValueError):
|
|
decode_cursor("not-base64!!!")
|
|
|
|
|
|
# --- Fixture builders -----------------------------------------------------
|
|
|
|
|
|
async def _seed_artist(db, name: str):
|
|
a = Artist(name=name, slug=name.lower().replace(" ", "-"))
|
|
db.add(a)
|
|
await db.flush()
|
|
return a
|
|
|
|
|
|
async def _seed_source(db, artist_id: int, platform: str, url: str):
|
|
s = Source(artist_id=artist_id, platform=platform, url=url, enabled=True)
|
|
db.add(s)
|
|
await db.flush()
|
|
return s
|
|
|
|
|
|
async def _seed_post(
|
|
db, source_id: int, *, external_id: str,
|
|
post_date=None, downloaded_at=None, title=None, description=None,
|
|
):
|
|
# Post.artist_id is NOT NULL since alembic 0030; look it up from
|
|
# the source so callsites don't have to thread the artist through.
|
|
artist_id = (await db.execute(
|
|
select(Source.artist_id).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
p = Post(
|
|
source_id=source_id, artist_id=artist_id, external_post_id=external_id,
|
|
post_title=title, post_date=post_date,
|
|
description=description,
|
|
)
|
|
if downloaded_at is not None:
|
|
p.downloaded_at = downloaded_at
|
|
db.add(p)
|
|
await db.flush()
|
|
return p
|
|
|
|
|
|
# --- scroll() ordering and pagination ------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_orders_by_coalesce_post_date_then_downloaded_at(db):
|
|
artist = await _seed_artist(db, "alice-order")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-order")
|
|
|
|
now = datetime.now(UTC)
|
|
# post A: post_date older than its downloaded_at; sort key = post_date
|
|
a = await _seed_post(
|
|
db, src.id, external_id="A",
|
|
post_date=now - timedelta(days=10),
|
|
downloaded_at=now,
|
|
)
|
|
# post B: post_date NULL → sort key = downloaded_at
|
|
b = await _seed_post(
|
|
db, src.id, external_id="B",
|
|
post_date=None,
|
|
downloaded_at=now - timedelta(days=5),
|
|
)
|
|
# post C: post_date newest of all
|
|
c = await _seed_post(
|
|
db, src.id, external_id="C",
|
|
post_date=now - timedelta(days=1),
|
|
downloaded_at=now - timedelta(days=20),
|
|
)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
ids = [item["id"] for item in page["items"]]
|
|
# Expected order by coalesce(post_date, downloaded_at) DESC:
|
|
# C (1 day ago) > B (5 days ago via downloaded_at) > A (10 days ago)
|
|
assert ids == [c.id, b.id, a.id]
|
|
assert page["next_cursor"] is None # 3 items, limit 10, no more
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_pagination_across_two_pages(db):
|
|
artist = await _seed_artist(db, "alice-page")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-page")
|
|
|
|
now = datetime.now(UTC)
|
|
posts = []
|
|
for i in range(5):
|
|
p = await _seed_post(
|
|
db, src.id, external_id=f"P{i}",
|
|
post_date=now - timedelta(days=i),
|
|
)
|
|
posts.append(p)
|
|
await db.commit()
|
|
|
|
page1 = await PostFeedService(db).scroll(cursor=None, limit=2)
|
|
assert len(page1["items"]) == 2
|
|
assert page1["next_cursor"] is not None
|
|
assert [it["id"] for it in page1["items"]] == [posts[0].id, posts[1].id]
|
|
|
|
page2 = await PostFeedService(db).scroll(cursor=page1["next_cursor"], limit=2)
|
|
assert [it["id"] for it in page2["items"]] == [posts[2].id, posts[3].id]
|
|
assert page2["next_cursor"] is not None
|
|
|
|
page3 = await PostFeedService(db).scroll(cursor=page2["next_cursor"], limit=2)
|
|
assert [it["id"] for it in page3["items"]] == [posts[4].id]
|
|
assert page3["next_cursor"] is None
|
|
|
|
|
|
# --- Filters --------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_filters_by_artist(db):
|
|
alice = await _seed_artist(db, "alice-filter")
|
|
bob = await _seed_artist(db, "bob-filter")
|
|
src_a = await _seed_source(db, alice.id, "patreon", "https://p/alice-f")
|
|
src_b = await _seed_source(db, bob.id, "patreon", "https://p/bob-f")
|
|
now = datetime.now(UTC)
|
|
pa = await _seed_post(db, src_a.id, external_id="PA", post_date=now)
|
|
await _seed_post(db, src_b.id, external_id="PB", post_date=now)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(
|
|
cursor=None, limit=10, artist_id=alice.id,
|
|
)
|
|
assert [it["id"] for it in page["items"]] == [pa.id]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_filters_by_platform(db):
|
|
artist = await _seed_artist(db, "alice-platf")
|
|
src_p = await _seed_source(db, artist.id, "patreon", "https://p/alice-pp")
|
|
src_d = await _seed_source(db, artist.id, "deviantart", "https://d/alice-dd")
|
|
now = datetime.now(UTC)
|
|
pp = await _seed_post(db, src_p.id, external_id="PP", post_date=now)
|
|
await _seed_post(db, src_d.id, external_id="PD", post_date=now)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(
|
|
cursor=None, limit=10, platform="patreon",
|
|
)
|
|
assert [it["id"] for it in page["items"]] == [pp.id]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_text_search_matches_title_or_description(db):
|
|
artist = await _seed_artist(db, "alice-search")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-se")
|
|
now = datetime.now(UTC)
|
|
by_title = await _seed_post(
|
|
db, src.id, external_id="T", post_date=now,
|
|
title="Dragon Quest cover", description="<p>nothing</p>",
|
|
)
|
|
by_desc = await _seed_post(
|
|
db, src.id, external_id="D", post_date=now - timedelta(days=1),
|
|
title="Untitled", description="<p>a sleeping dragon</p>",
|
|
)
|
|
await _seed_post(
|
|
db, src.id, external_id="N", post_date=now - timedelta(days=2),
|
|
title="Cat nap", description="<p>just a cat</p>",
|
|
)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10, q="dragon")
|
|
ids = {it["id"] for it in page["items"]}
|
|
# ILIKE is case-insensitive and spans title OR description; the cat post
|
|
# matches neither.
|
|
assert ids == {by_title.id, by_desc.id}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_text_search_stays_scoped_to_artist(db):
|
|
alice = await _seed_artist(db, "alice-scope")
|
|
bob = await _seed_artist(db, "bob-scope")
|
|
src_a = await _seed_source(db, alice.id, "patreon", "https://p/alice-sc")
|
|
src_b = await _seed_source(db, bob.id, "patreon", "https://p/bob-sc")
|
|
now = datetime.now(UTC)
|
|
alice_hit = await _seed_post(
|
|
db, src_a.id, external_id="AH", post_date=now, title="comet study",
|
|
)
|
|
# Same query word, different artist — must be excluded by the scope.
|
|
await _seed_post(
|
|
db, src_b.id, external_id="BH", post_date=now, title="comet sketch",
|
|
)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(
|
|
cursor=None, limit=10, q="comet", artist_id=alice.id,
|
|
)
|
|
assert [it["id"] for it in page["items"]] == [alice_hit.id]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_combined_artist_and_platform(db):
|
|
alice = await _seed_artist(db, "alice-combo")
|
|
bob = await _seed_artist(db, "bob-combo")
|
|
src_alice_patreon = await _seed_source(db, alice.id, "patreon", "https://p/alice-c")
|
|
src_alice_da = await _seed_source(db, alice.id, "deviantart", "https://d/alice-c")
|
|
src_bob_patreon = await _seed_source(db, bob.id, "patreon", "https://p/bob-c")
|
|
now = datetime.now(UTC)
|
|
target = await _seed_post(
|
|
db, src_alice_patreon.id, external_id="TARGET", post_date=now,
|
|
)
|
|
# Distractors — wrong platform for alice, wrong artist for patreon.
|
|
await _seed_post(db, src_alice_da.id, external_id="DA", post_date=now)
|
|
await _seed_post(db, src_bob_patreon.id, external_id="BP", post_date=now)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(
|
|
cursor=None, limit=10, artist_id=alice.id, platform="patreon",
|
|
)
|
|
assert [it["id"] for it in page["items"]] == [target.id]
|
|
|
|
|
|
# --- Dict shape and content ----------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_item_shape_minimal(db):
|
|
artist = await _seed_artist(db, "alice-shape")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-sh")
|
|
now = datetime.now(UTC)
|
|
await _seed_post(
|
|
db, src.id, external_id="SHAPE", post_date=now,
|
|
title="Hello", description="<p>world</p>",
|
|
)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
item = page["items"][0]
|
|
assert set(item.keys()) >= {
|
|
"id", "external_post_id", "post_url", "post_title",
|
|
"post_date", "downloaded_at",
|
|
"description_plain", "description_truncated",
|
|
"artist", "source",
|
|
"thumbnails", "thumbnails_more", "attachments",
|
|
}
|
|
assert item["post_title"] == "Hello"
|
|
assert item["description_plain"] == "world"
|
|
assert item["description_truncated"] is False
|
|
assert item["artist"]["slug"] == "alice-shape"
|
|
assert item["source"]["platform"] == "patreon"
|
|
# List response NEVER carries the full description.
|
|
assert "description_full" not in item
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_description_truncates_long_text(db):
|
|
artist = await _seed_artist(db, "alice-long")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-lo")
|
|
long_text = "word " * 200 # ~1000 chars
|
|
await _seed_post(
|
|
db, src.id, external_id="LONG",
|
|
post_date=datetime.now(UTC),
|
|
description=long_text,
|
|
)
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
item = page["items"][0]
|
|
assert item["description_truncated"] is True
|
|
assert len(item["description_plain"]) <= 281 # 280 + ellipsis char
|
|
|
|
|
|
# --- Thumbnails composition (primary_post_id linkage) --------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_thumbnails_show_first_six_with_overflow_count(db):
|
|
artist = await _seed_artist(db, "alice-thumbs")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-th")
|
|
post = await _seed_post(
|
|
db, src.id, external_id="THUMB",
|
|
post_date=datetime.now(UTC),
|
|
)
|
|
# Seed 8 ImageRecords linked to this post (unique sha + path each).
|
|
for i in range(8):
|
|
db.add(ImageRecord(
|
|
path=f"/images/thumb-{post.id}-{i}.jpg",
|
|
sha256=f"{post.id:02d}" + f"{i:062d}",
|
|
size_bytes=10, mime="image/jpeg", width=10, height=10,
|
|
origin="downloaded", primary_post_id=post.id,
|
|
artist_id=artist.id,
|
|
))
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
item = page["items"][0]
|
|
assert len(item["thumbnails"]) == 6
|
|
assert item["thumbnails_more"] == 2
|
|
for t in item["thumbnails"]:
|
|
assert "image_id" in t
|
|
assert t["thumbnail_url"].startswith("/images/thumbs/")
|
|
assert t["mime"] == "image/jpeg"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_thumbnails_include_provenance_linked_duplicates(db):
|
|
"""A cross-posted/duplicate image is linked to a second post via
|
|
image_provenance (its primary_post_id stays on the first post). The feed
|
|
renders by provenance UNION primary, so the image shows on BOTH posts —
|
|
not just its primary. Operator requirement 2026-06-08."""
|
|
artist = await _seed_artist(db, "alice-prov")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-prov")
|
|
primary_post = await _seed_post(
|
|
db, src.id, external_id="PRIMARY", post_date=datetime.now(UTC),
|
|
)
|
|
other_post = await _seed_post(
|
|
db, src.id, external_id="OTHER",
|
|
post_date=datetime.now(UTC) + timedelta(minutes=1),
|
|
)
|
|
img = ImageRecord(
|
|
path="/images/dup.jpg", sha256="d" * 64, size_bytes=10,
|
|
mime="image/jpeg", width=10, height=10, origin="downloaded",
|
|
primary_post_id=primary_post.id, artist_id=artist.id,
|
|
)
|
|
db.add(img)
|
|
await db.flush()
|
|
# The duplicate links to the OTHER post via provenance only.
|
|
db.add(ImageProvenance(image_record_id=img.id, post_id=other_post.id))
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
by_ext = {it["external_post_id"]: it for it in page["items"]}
|
|
assert len(by_ext["PRIMARY"]["thumbnails"]) == 1
|
|
assert len(by_ext["OTHER"]["thumbnails"]) == 1 # shown via provenance
|
|
assert (
|
|
by_ext["PRIMARY"]["thumbnails"][0]["image_id"]
|
|
== by_ext["OTHER"]["thumbnails"][0]["image_id"]
|
|
)
|
|
|
|
|
|
# --- Attachments composition ---------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scroll_attachments_returns_non_media_files(db):
|
|
artist = await _seed_artist(db, "alice-att")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-at")
|
|
post = await _seed_post(
|
|
db, src.id, external_id="ATT", post_date=datetime.now(UTC),
|
|
)
|
|
db.add(PostAttachment(
|
|
post_id=post.id, artist_id=artist.id,
|
|
sha256="a" * 64, path="/images/att/x.pdf",
|
|
original_filename="notes.pdf", ext="pdf",
|
|
mime="application/pdf", size_bytes=12345,
|
|
))
|
|
await db.commit()
|
|
|
|
page = await PostFeedService(db).scroll(cursor=None, limit=10)
|
|
item = page["items"][0]
|
|
assert len(item["attachments"]) == 1
|
|
att = item["attachments"][0]
|
|
assert att["original_filename"] == "notes.pdf"
|
|
assert att["ext"] == "pdf"
|
|
assert att["mime"] == "application/pdf"
|
|
assert att["size_bytes"] == 12345
|
|
assert att["download_url"] == f"/api/attachments/{att['id']}/download"
|
|
|
|
|
|
# --- get_post() detail ---------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_post_returns_full_description(db):
|
|
artist = await _seed_artist(db, "alice-detail")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-de")
|
|
long_text = "word " * 200
|
|
p = await _seed_post(
|
|
db, src.id, external_id="DETAIL",
|
|
post_date=datetime.now(UTC),
|
|
description=long_text,
|
|
)
|
|
await db.commit()
|
|
|
|
item = await PostFeedService(db).get_post(p.id)
|
|
assert item is not None
|
|
assert "description_full" in item
|
|
# Full text is HTML-stripped but NOT truncated.
|
|
assert len(item["description_full"]) > 280
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_post_returns_sanitized_html_body(db):
|
|
artist = await _seed_artist(db, "alice-html")
|
|
src = await _seed_source(db, artist.id, "patreon", "https://p/alice-html")
|
|
p = await _seed_post(
|
|
db, src.id, external_id="HTMLBODY", post_date=datetime.now(UTC),
|
|
description='<h2>Heading</h2><p>body</p><script>alert(1)</script>',
|
|
)
|
|
await db.commit()
|
|
|
|
item = await PostFeedService(db).get_post(p.id)
|
|
# Detail carries the sanitized HTML body for faithful rendering; <script> is
|
|
# stripped, semantic tags survive.
|
|
assert item["description_html"] is not None
|
|
assert "<h2>" in item["description_html"]
|
|
assert "<script>" not in item["description_html"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_post_unknown_returns_none(db):
|
|
item = await PostFeedService(db).get_post(99999)
|
|
assert item is None
|