FabledCurator/tests/test_post_feed_service.py

"""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,
    ExternalLink,
    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_localizes_inline_images(db):
    """#830 Phase 2: a body <img src=CDN> whose filehash matches a stored image
    of this artist is rewritten to the local /images path; an unmatched src is
    left hotlinking."""
    artist = await _seed_artist(db, "alice-inline")
    src = await _seed_source(db, artist.id, "patreon", "https://p/alice-inline")
    fh = "0123456789abcdef0123456789abcdef"
    matched = f"https://cdn.test/p/{fh}/img.png"
    unmatched = "https://cdn.test/p/ffffffffffffffffffffffffffffffff/other.png"
    p = await _seed_post(
        db, src.id, external_id="INLINE", post_date=datetime.now(UTC),
        description=f'<p>x</p><img src="{matched}"><img src="{unmatched}">',
    )
    db.add(ImageRecord(
        path="/images/alice-inline/patreon/post/01_img.png",
        sha256=f"{p.id:064d}",
        size_bytes=10, mime="image/png", width=10, height=10,
        origin="downloaded", primary_post_id=p.id, artist_id=artist.id,
        source_url=matched, source_filehash=fh,
    ))
    await db.commit()

    item = await PostFeedService(db).get_post(p.id)
    html = item["description_html"]
    assert 'src="/images/alice-inline/patreon/post/01_img.png"' in html
    assert matched not in html  # CDN src swapped out
    assert unmatched in html    # no local copy → still hotlinked


@pytest.mark.asyncio
async def test_get_post_inline_image_scoped_to_artist(db):
    """A matching filehash owned by a DIFFERENT artist must not leak into this
    post's body — localization is artist-scoped."""
    a1 = await _seed_artist(db, "owner-art")
    a2 = await _seed_artist(db, "other-art")
    src = await _seed_source(db, a1.id, "patreon", "https://p/owner-art")
    fh = "abcabcabcabcabcabcabcabcabcabc12"
    cdn = f"https://cdn.test/p/{fh}/x.png"
    p = await _seed_post(
        db, src.id, external_id="SCOPED", post_date=datetime.now(UTC),
        description=f'<img src="{cdn}">',
    )
    # The only image with this filehash belongs to a DIFFERENT artist.
    db.add(ImageRecord(
        path="/images/other/x.png", sha256=f"{p.id:064d}",
        size_bytes=10, mime="image/png", width=10, height=10,
        origin="downloaded", artist_id=a2.id,
        source_url=cdn, source_filehash=fh,
    ))
    await db.commit()

    item = await PostFeedService(db).get_post(p.id)
    assert cdn in item["description_html"]  # not localized (wrong artist)


@pytest.mark.asyncio
async def test_get_post_returns_external_links(db):
    artist = await _seed_artist(db, "alice-extlinks")
    src = await _seed_source(db, artist.id, "patreon", "https://p/alice-extlinks")
    p = await _seed_post(
        db, src.id, external_id="EXTLINKS", post_date=datetime.now(UTC),
    )
    db.add(ExternalLink(
        post_id=p.id, artist_id=artist.id, host="mega",
        url="https://mega.nz/file/x#k", label="Mega",
    ))
    await db.commit()

    item = await PostFeedService(db).get_post(p.id)
    assert len(item["external_links"]) == 1
    link = item["external_links"][0]
    assert link["host"] == "mega"
    assert link["url"] == "https://mega.nz/file/x#k"
    assert link["status"] == "pending"


@pytest.mark.asyncio
async def test_get_post_unknown_returns_none(db):
    item = await PostFeedService(db).get_post(99999)
    assert item is None