FabledCurator/tests/test_pixiv_ingester.py

"""PixivIngester tests — the adapter's wiring over the shared core.

The core's walk/skip/cursor/budget behavior is exercised exhaustively by
test_patreon_ingester / test_subscribestar_native; these cover what is
pixiv-SPECIFIC: the synthesized ledger key, the real Postgres pixiv ledgers
(the upserts' constraint names must match migration 0076), the failure
mapping including the rate-limit retry_after carry, and the #862 body-canary
opt-out (caption-less pixiv feeds must not fail API_DRIFT).
"""

import pytest
from sqlalchemy import func, select
from sqlalchemy.orm import sessionmaker

from backend.app.models import Artist, PixivFailedMedia, PixivSeenMedia, Source
from backend.app.services.gallery_dl import ErrorType
from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
from backend.app.services.native_ingest_common import MediaOutcome, PostRecordOutcome
from backend.app.services.pixiv_client import (
    MediaItem,
    PixivAPIError,
    PixivAuthError,
    PixivDriftError,
)
from backend.app.services.pixiv_ingester import PixivIngester, _ledger_key

pytestmark = pytest.mark.integration


# --- fakes ----------------------------------------------------------------


def _media(post_id, num, *, filehash=None, media_id=None):
    mid = media_id if media_id is not None else f"p{num}"
    return MediaItem(
        url=f"https://i.pximg.net/img-original/img/2026/07/01/00/00/00/{post_id}_{mid}.png",
        filename=f"{post_id}_Work_{num:02d}.png",
        kind="image",
        filehash=filehash,
        post_id=str(post_id),
        media_id=mid,
    )


class _FakeClient:
    """Stub PixivClient. `pages` is a list of (page_cursor, [posts]); each post
    is (post_id, [MediaItem]). `raise_exc` trips a client-level failure;
    `captions` toggles per-post body text (the canary input)."""

    def __init__(self, pages, raise_exc=None, captions=True):
        self._pages = pages
        self._raise_exc = raise_exc
        self._captions = captions

    def iter_posts(self, campaign_id, cursor=None):
        if self._raise_exc is not None:
            raise self._raise_exc
        for page_cursor, posts in self._pages:
            for post_id, media in posts:
                caption = f"<p>caption {post_id}</p>" if self._captions else ""
                yield (
                    {
                        "id": post_id,
                        "_media": media,
                        "attributes": {
                            "title": f"Work {post_id}",
                            "post_type": "illust",
                            "content": caption,
                            "published_at": "2026-07-01T00:00:00+09:00",
                        },
                        "_work": {},
                    },
                    {},
                    page_cursor,
                )

    def extract_media(self, post, included_index):
        return post["_media"]

    def post_meta(self, post):
        return {"title": post.get("id"), "date": None}

    @staticmethod
    def post_is_gated(post):
        return False

    @staticmethod
    def post_record_key(post):
        pid = str(post.get("id") or "")
        return (f"post:{pid}", pid) if pid else None


class _FakeDownloader:
    """Stub PixivDownloader. Honors the injected is_seen; media whose ledger
    key is in `error` report a download failure."""

    def __init__(self, tmp_path, error=None):
        self.tmp_path = tmp_path
        self.error = set(error or ())
        self.download_calls = 0

    def download_post(self, post, media_items, artist_slug, *, is_seen,
                      should_stop=lambda: False, recapture=False):
        outcomes = []
        for m in media_items:
            if should_stop():
                break
            if is_seen(m) and not recapture:
                outcomes.append(
                    MediaOutcome(media=m, status="skipped_seen", path=None, error=None)
                )
            elif _ledger_key(m) in self.error:
                self.download_calls += 1
                outcomes.append(
                    MediaOutcome(media=m, status="error", path=None, error="403 pximg")
                )
            else:
                self.download_calls += 1
                p = self.tmp_path / m.filename
                p.write_bytes(b"x")
                outcomes.append(
                    MediaOutcome(media=m, status="downloaded", path=p, error=None)
                )
        return outcomes

    def write_post_record(self, post, artist_slug):
        p = self.tmp_path / f"_post_{post.get('id')}.json"
        p.write_text("{}")
        attrs = post.get("attributes") or {}
        body = attrs.get("content")
        return PostRecordOutcome(
            path=p,
            post_type=attrs.get("post_type"),
            title=attrs.get("title"),
            body_chars=len(body) if isinstance(body, str) else 0,
        )


@pytest.fixture
async def source_id(db):
    artist = Artist(name="Pixland", slug="pixland")
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform="pixiv",
        url="https://www.pixiv.net/users/99", enabled=True, config_overrides={},
    )
    db.add(source)
    await db.commit()
    return source.id


def _ingester(sync_engine, tmp_path, client, downloader):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    return PixivIngester(
        images_root=tmp_path, cookies_path=None, session_factory=factory,
        client=client, downloader=downloader,
    )


def _run(ing, source_id, mode="tick"):
    return ing.run(
        source_id=source_id, campaign_id="99", artist_slug="pixland",
        url="https://www.pixiv.net/users/99", mode=mode,
    )


def _seen_keys(sync_engine, source_id):
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        return set(s.execute(
            select(PixivSeenMedia.filehash).where(
                PixivSeenMedia.source_id == source_id
            )
        ).scalars().all())


# --- ledger key -------------------------------------------------------------


def test_ledger_key_shapes():
    assert _ledger_key(_media(111, 0)) == "111:p0"
    assert _ledger_key(_media(111, 4)) == "111:p4"
    assert _ledger_key(_media(333, 0, media_id="ugoira")) == "333:ugoira"
    assert _ledger_key(_media(1, 0, filehash="f" * 32)) == "f" * 32
    long_key = _ledger_key(_media("9" * 200, 0))
    assert len(long_key) <= 128


# --- walk + ledgers ---------------------------------------------------------


@pytest.mark.asyncio
async def test_tick_downloads_and_marks_seen(source_id, sync_engine, tmp_path):
    m0, m1 = _media(111, 0), _media(111, 1)
    client = _FakeClient([(None, [(111, [m0, m1])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = _run(ing, source_id)

    assert result.success is True
    assert result.files_downloaded == 2
    assert len(result.post_record_paths) == 1
    # 2 media keys + the synthetic post key, in the pixiv-shaped key format.
    assert _seen_keys(sync_engine, source_id) == {"111:p0", "111:p1", "post:111"}


@pytest.mark.asyncio
async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
    m0 = _media(111, 0)
    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        s.add(PixivSeenMedia(
            source_id=source_id, filehash=_ledger_key(m0), post_id="111",
        ))
        s.commit()

    client = _FakeClient([(None, [(111, [m0])])])
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = _run(ing, source_id)
    assert result.files_downloaded == 0
    assert downloader.download_calls == 0


@pytest.mark.asyncio
async def test_failed_media_lands_in_dead_letter_ledger(
    source_id, sync_engine, tmp_path
):
    m0 = _media(111, 0)
    client = _FakeClient([(None, [(111, [m0])])])
    downloader = _FakeDownloader(tmp_path, error={_ledger_key(m0)})
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = _run(ing, source_id)
    assert result.run_stats["per_item_failures"] == 1

    factory = sessionmaker(sync_engine, expire_on_commit=False)
    with factory() as s:
        row = s.execute(
            select(PixivFailedMedia).where(
                PixivFailedMedia.source_id == source_id
            )
        ).scalar_one()
    assert row.filehash == "111:p0"
    assert row.attempts == 1
    assert "403" in row.last_error


# --- failure mapping ----------------------------------------------------------


@pytest.mark.asyncio
async def test_auth_error_maps_to_auth_error(source_id, sync_engine, tmp_path):
    client = _FakeClient([], raise_exc=PixivAuthError(
        "rotate the token", status_code=400,
    ))
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
    result = _run(ing, source_id)
    assert result.success is False
    assert result.error_type == ErrorType.AUTH_ERROR


@pytest.mark.asyncio
async def test_drift_error_maps_to_api_drift(source_id, sync_engine, tmp_path):
    client = _FakeClient([], raise_exc=PixivDriftError("no illusts list"))
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
    result = _run(ing, source_id)
    assert result.error_type == ErrorType.API_DRIFT
    assert "Pixiv app API changed" in result.error_message


@pytest.mark.asyncio
async def test_rate_limit_carries_retry_after(source_id, sync_engine, tmp_path):
    client = _FakeClient([], raise_exc=PixivAPIError(
        "rate limited", status_code=429, retry_after=300.0,
    ))
    ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
    result = _run(ing, source_id)
    assert result.error_type == ErrorType.RATE_LIMITED
    assert result.retry_after_seconds == 300.0


# --- body canary opt-out --------------------------------------------------------


@pytest.mark.asyncio
async def test_captionless_feed_does_not_trip_body_canary(
    source_id, sync_engine, tmp_path
):
    """#862 canary opt-out: a pixiv backfill recording ≥ the canary sample of
    caption-less works is NORMAL (many artists never write captions) — it must
    complete, not fail API_DRIFT the way a zero-body Patreon walk does."""
    n = _CANARY_MIN_SAMPLE + 5
    posts = [(1000 + i, [_media(1000 + i, 0)]) for i in range(n)]
    client = _FakeClient([(None, posts)], captions=False)
    downloader = _FakeDownloader(tmp_path)
    ing = _ingester(sync_engine, tmp_path, client, downloader)

    result = _run(ing, source_id, mode="backfill")
    assert result.success is True
    assert result.error_type is None
    assert result.files_downloaded == n