FabledCurator/tests/test_pixiv_downloader.py

"""Unit tests for PixivDownloader — no network, no DB.

The HTTP layer is stubbed via the `session` seam (a fake session whose `.get`
returns canned streaming bytes). Media items come from the real
PixivClient.extract_media over the shared fixture page, so the downloader is
exercised against the exact shapes the client produces.
"""

from __future__ import annotations

import json
from pathlib import Path

import requests

from backend.app.services.pixiv_client import MediaItem, PixivClient
from backend.app.services.pixiv_downloader import PixivDownloader

_FIXTURE = Path(__file__).parent / "fixtures" / "pixiv_user_illusts_page1.json"

# Minimal valid PNG so the file_validator passes on the happy path.
_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
_PNG_TAIL = b"\x00\x00\x00\x00IEND\xaeB`\x82"
_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + _PNG_TAIL


class _FakeResponse:
    def __init__(self, payload: bytes, status_code: int = 200, headers=None):
        self._payload = payload
        self.status_code = status_code
        self.headers = headers or {}

    def raise_for_status(self):
        if self.status_code >= 400:
            raise requests.HTTPError(f"HTTP {self.status_code}")
        return None

    def iter_content(self, chunk_size=65536):
        for i in range(0, len(self._payload), chunk_size):
            yield self._payload[i : i + chunk_size]


class _FakeSession:
    """Records GETs and returns canned bytes (per-URL, default _PNG_BYTES)."""

    def __init__(self, payloads: dict[str, bytes] | None = None):
        self.payloads = payloads or {}
        self.calls: list[str] = []

    def get(self, url, stream=False, timeout=None, headers=None):
        self.calls.append(url)
        return _FakeResponse(self.payloads.get(url, _PNG_BYTES))


def _client():
    # Parsing only — extract_media for non-ugoira works never issues a request.
    return PixivClient("tok", session=_FakeSession())


def _post_and_media(work_id):
    page = json.loads(_FIXTURE.read_text())
    client = _client()
    for work in page["illusts"]:
        if work["id"] == work_id:
            post = client._normalize(work)
            return post, client.extract_media(post, {})
    raise AssertionError(f"no work {work_id} in fixture")


def _post_only(work_id):
    """Normalized post WITHOUT media extraction — the write_post_record tests
    don't need media, and extracting an ugoira would hit the API seam."""
    page = json.loads(_FIXTURE.read_text())
    for work in page["illusts"]:
        if work["id"] == work_id:
            return _client()._normalize(work)
    raise AssertionError(f"no work {work_id} in fixture")


def _downloader(tmp_path, session=None):
    # validate=False: the stub payload is PNG bytes regardless of the target
    # extension, so real validation would quarantine the .jpg cases. The
    # validate/quarantine plumbing is BaseNativeDownloader's, covered by the
    # Patreon downloader tests.
    return PixivDownloader(
        tmp_path, validate=False,
        session=session if session is not None else _FakeSession(),
    )


def test_download_post_writes_gallery_dl_layout(tmp_path):
    post, media = _post_and_media(111)
    dl = _downloader(tmp_path)
    outcomes = dl.download_post(post, media, "artist-a")

    assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
    p0 = flat / "111_Multi Page Adventure_00.png"
    p1 = flat / "111_Multi Page Adventure_01.png"
    assert p0.is_file()
    assert p1.is_file()
    # No per-post directory — pixiv's layout is flat (gallery-dl parity).
    assert {p.parent for p in (p0, p1)} == {flat}


def test_download_post_writes_minimal_sidecar(tmp_path):
    post, media = _post_and_media(222)
    dl = _downloader(tmp_path)
    outcomes = dl.download_post(post, media, "artist-a")

    assert outcomes[0].status == "downloaded"
    sidecar = outcomes[0].path.with_suffix(".json")
    data = json.loads(sidecar.read_text())
    # Post-first: image identity ONLY — the body lives in the post record.
    assert data == {
        "category": "pixiv",
        "id": "222",
        "source_url": media[0].url,
    }


def test_download_post_skips_seen(tmp_path):
    post, media = _post_and_media(222)
    session = _FakeSession()
    dl = _downloader(tmp_path, session=session)
    outcomes = dl.download_post(post, media, "artist-a", is_seen=lambda m: True)
    assert [o.status for o in outcomes] == ["skipped_seen"]
    assert session.calls == []


def test_download_post_skips_on_disk(tmp_path):
    post, media = _post_and_media(222)
    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
    flat.mkdir(parents=True)
    existing = flat / "222_Single Piece_00.jpg"
    existing.write_bytes(b"already here")

    session = _FakeSession()
    dl = _downloader(tmp_path, session=session)
    outcomes = dl.download_post(post, media, "artist-a")
    assert [o.status for o in outcomes] == ["skipped_disk"]
    assert outcomes[0].path == existing
    assert session.calls == []
    assert existing.read_bytes() == b"already here"


def test_recapture_surfaces_on_disk_path_for_seen_media(tmp_path):
    post, media = _post_and_media(222)
    flat = tmp_path / "artist-a" / "pixiv" / "pixiv"
    flat.mkdir(parents=True)
    (flat / "222_Single Piece_00.jpg").write_bytes(b"kept")

    dl = _downloader(tmp_path)
    outcomes = dl.download_post(
        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
    )
    # On disk → surfaced with its path (relink channel); seen-but-missing
    # stays skipped_seen (recovery's job, not recapture's).
    assert outcomes[0].status == "skipped_disk"
    assert outcomes[0].path is not None


def test_recapture_does_not_redownload_missing_seen_media(tmp_path):
    post, media = _post_and_media(222)
    session = _FakeSession()
    dl = _downloader(tmp_path, session=session)
    outcomes = dl.download_post(
        post, media, "artist-a", is_seen=lambda m: True, recapture=True,
    )
    assert [o.status for o in outcomes] == ["skipped_seen"]
    assert session.calls == []


def test_download_post_honours_should_stop(tmp_path):
    post, media = _post_and_media(111)
    dl = _downloader(tmp_path)
    outcomes = dl.download_post(post, media, "artist-a", should_stop=lambda: True)
    assert outcomes == []


def test_filename_matches_gallery_dl_linux_restrict(tmp_path):
    # gallery-dl on Linux replaces ONLY "/" and deletes control chars — the
    # Windows-forbidden set (<>:"|?*) stays RAW in the on-disk name. Matching
    # that byte-for-byte is what lets the tier-2 disk-skip recognize
    # pre-cutover files; a stricter sanitizer would re-download them.
    post, _ = _post_and_media(222)
    weird = MediaItem(
        url="https://i.pximg.net/img-original/img/2026/06/18/12/30/00/222_p0.jpg",
        filename='222_What? A "Title"/Slash\x0900.jpg',  # ? " stay, / → _, \t dropped
        kind="image",
        filehash=None,
        post_id="222",
        media_id="p0",
    )
    dl = _downloader(tmp_path)
    outcomes = dl.download_post(post, [weird], "artist-a")
    assert outcomes[0].status == "downloaded"
    assert outcomes[0].path.name == '222_What? A "Title"_Slash00.jpg'


def test_gdl_clean_filename_parity():
    from backend.app.services.pixiv_downloader import gdl_clean_filename
    # Only "/" is replaced; the Windows-forbidden set is untouched.
    assert gdl_clean_filename('a<>:"|?*b.png') == 'a<>:"|?*b.png'
    assert gdl_clean_filename("a/b/c.png") == "a_b_c.png"
    # Control chars (newline, tab, DEL) are deleted, not replaced.
    assert gdl_clean_filename("a\nb\tc\x7fd.png") == "abcd.png"
    # Trailing dots/spaces are NOT stripped on Linux.
    assert gdl_clean_filename("title . .png") == "title . .png"


def test_own_session_carries_app_referer(tmp_path):
    # Built session (no injection) must carry the app profile — i.pximg.net
    # 403s GETs without the app-api Referer.
    dl = PixivDownloader(tmp_path)
    assert dl.session.headers["Referer"] == "https://app-api.pixiv.net/"
    assert dl.session.headers["App-OS"] == "ios"


def test_write_post_record_enriched(tmp_path):
    post = _post_only(111)
    dl = _downloader(tmp_path)
    outcome = dl.write_post_record(post, "artist-a")

    assert outcome.path == (
        tmp_path / "artist-a" / "pixiv" / "pixiv" / "_post_111.json"
    )
    assert outcome.title == "Multi Page Adventure"
    assert outcome.post_type == "illust"
    assert outcome.body_chars == len(post["attributes"]["content"])

    data = json.loads(outcome.path.read_text())
    assert data["category"] == "pixiv"
    assert data["id"] == "111"
    assert data["published_at"] == "2026-06-20T18:00:00+09:00"
    assert data["rating"] == "General"
    assert data["page_count"] == 2
    assert data["total_bookmarks"] == 250
    assert data["total_view"] == 1000
    assert data["illust_ai_type"] == 1
    assert data["series"] == {"id": 4242, "title": "Adventure Series"}
    assert data["tags"] == [
        {"name": "オリジナル", "translated_name": "original"},
        {"name": "女の子", "translated_name": "girl"},
    ]
    assert data["user"] == {"id": 99, "account": "exartist", "name": "Example Artist"}
    # JP tags stay human-readable on disk (ensure_ascii=False).
    assert "オリジナル" in outcome.path.read_text()


def test_write_post_record_rating_r18(tmp_path):
    post = _post_only(222)
    dl = _downloader(tmp_path)
    outcome = dl.write_post_record(post, "artist-a")
    data = json.loads(outcome.path.read_text())
    assert data["rating"] == "R-18"
    assert data["is_bookmarked"] is True


def test_write_post_record_includes_ugoira_frames(tmp_path):
    post = _post_only(333)
    frames = [{"file": "000000.jpg", "delay": 90}]
    post["_work"]["_ugoira_frames"] = frames
    dl = _downloader(tmp_path)
    outcome = dl.write_post_record(post, "artist-a")
    data = json.loads(outcome.path.read_text())
    assert data["type"] == "ugoira"
    assert data["ugoira_frames"] == frames


def test_write_post_record_fetches_ugoira_frames_when_absent(tmp_path):
    # The core writes the post record BEFORE extract_media, so frames aren't
    # memoized yet — the injected fetcher must populate them so the record keeps
    # the timings (regression: they were silently always empty).
    post = _post_only(333)
    assert "_ugoira_frames" not in post["_work"]
    frames = [{"file": "000000.jpg", "delay": 120}]
    calls = []

    def fetcher(p):
        calls.append(p)
        p["_work"]["_ugoira_frames"] = frames

    dl = PixivDownloader(
        tmp_path, validate=False, session=_FakeSession(),
        ugoira_frames_fetcher=fetcher,
    )
    outcome = dl.write_post_record(post, "artist-a")
    data = json.loads(outcome.path.read_text())
    assert data["ugoira_frames"] == frames
    assert len(calls) == 1


def test_write_post_record_non_ugoira_does_not_call_fetcher(tmp_path):
    post = _post_only(111)                       # a plain multi-page illust
    calls = []
    dl = PixivDownloader(
        tmp_path, validate=False, session=_FakeSession(),
        ugoira_frames_fetcher=lambda p: calls.append(p),
    )
    dl.write_post_record(post, "artist-a")
    assert calls == []


def test_write_post_record_without_id(tmp_path):
    dl = _downloader(tmp_path)
    outcome = dl.write_post_record({"id": None, "attributes": {}}, "artist-a")
    assert outcome.path is None
    assert outcome.body_chars == 0