FabledCurator/tests/test_gallery_dl_service.py

"""GalleryDLService tests — subprocess is mocked so no real gallery-dl runs."""

from types import SimpleNamespace

import pytest

from backend.app.services.gallery_dl import (
    ErrorType,
    GalleryDLService,
    SourceConfig,
    parse_last_cursor,
)


@pytest.fixture
def gdl(tmp_path):
    return GalleryDLService(
        images_root=tmp_path / "images",
        rate_limit=3.0,
        validate_files=False,
    )


def _proc(stdout="", stderr="", returncode=0):
    return SimpleNamespace(stdout=stdout, stderr=stderr, returncode=returncode)


@pytest.mark.asyncio
async def test_download_happy_path(gdl, monkeypatch):
    out_path = gdl.images_root / "alice" / "patreon" / "post1" / "img.jpg"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 32 + b"\xff\xd9")

    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(stdout=str(out_path) + "\n", returncode=0),
    )
    result = await gdl.download(
        url="https://patreon.com/alice",
        artist_slug="alice",
        platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.success is True
    assert result.files_downloaded == 1
    assert result.error_type is None
    assert str(out_path) in result.written_paths


@pytest.mark.asyncio
async def test_download_no_new_content(gdl, monkeypatch):
    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(stdout="# /images/alice/patreon/old.jpg\n", returncode=0),
    )
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.success is True
    assert result.files_downloaded == 0


@pytest.mark.asyncio
async def test_download_auth_error_categorization(gdl, monkeypatch):
    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(
            stdout="", stderr='[patreon][error] "GET /api/foo HTTP/1.1" 401 None\n',
            returncode=1,
        ),
    )
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.success is False
    assert result.error_type == ErrorType.AUTH_ERROR


@pytest.mark.asyncio
async def test_download_timeout(gdl, monkeypatch):
    import subprocess as sp

    def _raise(*a, **k):
        raise sp.TimeoutExpired(cmd="gallery-dl", timeout=10)

    monkeypatch.setattr("backend.app.services.gallery_dl.subprocess.run", _raise)
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(timeout=10),
    )
    assert result.success is False
    assert result.error_type == ErrorType.TIMEOUT
    # Empty-partial timeout: no preserved output and zero files.
    assert result.stdout == ""
    assert result.stderr == ""
    assert result.files_downloaded == 0


@pytest.mark.asyncio
async def test_download_timeout_preserves_partial_output_and_classifies(gdl, monkeypatch):
    """When gallery-dl is killed by timeout but had emitted partial output,
    the DownloadResult preserves stdout/stderr, counts files written so
    far, and promotes to RATE_LIMITED if the partial stderr shows a rate-
    limit pattern (so the platform cooldown kicks in). Without this, the
    operator only sees 'timed out' with no clue whether it was 'lots of
    content', 'stuck retrying', or 'hung silent'."""
    import subprocess as sp

    # Simulate gallery-dl writing two files, then spinning on 429s.
    partial_stdout = (
        "/tmp/images/alice/file_001.jpg\n"
        "/tmp/images/alice/file_002.jpg\n"
    )
    partial_stderr = (
        "[urllib3] 429 Too Many Requests; sleeping 60s\n"
        "[urllib3] 429 Too Many Requests; sleeping 60s\n"
    )

    def _raise(*a, **k):
        raise sp.TimeoutExpired(
            cmd="gallery-dl", timeout=900,
            output=partial_stdout, stderr=partial_stderr,
        )

    monkeypatch.setattr("backend.app.services.gallery_dl.subprocess.run", _raise)
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(timeout=900),
    )
    assert result.success is False
    # Partial output preserved on the DownloadResult — _phase3_persist
    # writes them into download_event.metadata so the UI can render them.
    assert result.stdout == partial_stdout
    assert result.stderr == partial_stderr
    # Files counted from partial stdout (lines starting with '/').
    assert result.files_downloaded == 2
    # Rate-limit pattern in partial stderr → promoted to RATE_LIMITED so
    # _update_source_health stamps the platform cooldown.
    assert result.error_type == ErrorType.RATE_LIMITED
    assert "Rate-limited" in result.error_message


@pytest.mark.asyncio
async def test_validation_quarantines_truncated_files(tmp_path, monkeypatch):
    gdl = GalleryDLService(
        images_root=tmp_path / "images", rate_limit=3.0, validate_files=True,
    )
    bad = gdl.images_root / "alice" / "patreon" / "post" / "bad.jpg"
    bad.parent.mkdir(parents=True, exist_ok=True)
    bad.write_bytes(b"\xff\xd8\xff" + b"\x00" * 100)  # SOI but no EOI

    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(stdout=str(bad) + "\n", returncode=0),
    )
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.files_quarantined == 1
    assert len(result.quarantined_paths) == 1
    assert not bad.exists()
    assert result.error_type == ErrorType.VALIDATION_FAILED


@pytest.mark.asyncio
async def test_download_tier_limited_classified_as_success(gdl, monkeypatch):
    """Pure tier-gated stderr (paywall warnings only, no source-level
    errors). Should be TIER_LIMITED with success=True so the
    DownloadEvent doesn't surface as 'needs attention' for an operator
    who's just not paying the top tier. Pre-fix: this fell through to
    UNKNOWN_ERROR when the return code wasn't 1 or 4."""
    stderr = (
        "[patreon][warning] Not allowed to view post 159421258\n"
        "[patreon][warning] Not allowed to view post 159247879\n"
        "[patreon][warning] Not allowed to view post 158111214\n"
    )
    # Use a non-(1,4) return code to prove the gate widened.
    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(stdout="", stderr=stderr, returncode=128),
    )
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.error_type == ErrorType.TIER_LIMITED
    assert result.success is True
    assert "3 posts" in (result.error_message or "")


@pytest.mark.asyncio
async def test_download_tier_limited_with_per_item_video_failures(gdl, monkeypatch):
    """Real operator log shape (2026-05-31): hundreds of paywall warnings
    + a handful of per-item video errors caused by missing yt-dlp. The
    video errors are classified as per-item (not source-level), so the
    classifier should still pick TIER_LIMITED. Once yt-dlp is installed
    the video errors stop firing; until then, the source-health doesn't
    flap to 'needs attention' over paywalled content."""
    stderr = (
        "[patreon][warning] Not allowed to view post 159421258\n"
        "[patreon][warning] Not allowed to view post 159247879\n"
        "[downloader.ytdl][error] Cannot import yt-dlp or youtube-dl\n"
        "[download][error] Failed to download 02_video.mp4\n"
    )
    monkeypatch.setattr(
        "backend.app.services.gallery_dl.subprocess.run",
        lambda *a, **k: _proc(stdout="", stderr=stderr, returncode=1),
    )
    result = await gdl.download(
        url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
        source_config=SourceConfig(),
    )
    assert result.error_type == ErrorType.TIER_LIMITED
    assert result.success is True


def test_compute_run_stats(gdl):
    stats = gdl._compute_run_stats(
        return_code=0,
        stdout="/path/a.jpg\n/path/b.jpg\n# /path/c.jpg\n",
        stderr="[patreon][warning] foo\n[patreon][download][error] failed to download /x\n",
    )
    assert stats["downloaded_count"] == 2
    assert stats["skipped_count"] == 1
    assert stats["warning_count"] >= 1
    assert stats["per_item_failures"] == 1


def test_extract_errors_warnings(gdl):
    stderr = (
        "[patreon][debug] foo\n[patreon][error] bar\n"
        "[urllib3][debug] baz\n[patreon][warning] qux\n"
    )
    result = gdl._extract_errors_warnings(stderr)
    assert "[error]" in result
    assert "[warning]" in result
    assert "[debug]" not in result


def test_truncate_log_caps_large_text(gdl):
    huge = "line\n" * 200_000
    result = gdl._truncate_log(huge, max_bytes=10_000)
    assert len(result.encode()) < 11_000
    assert "lines elided" in result


def test_truncate_log_passes_short_text(gdl):
    text = "small\n"
    assert gdl._truncate_log(text) == text


# --- Plan #544: tick/backfill skip emission + PARTIAL classifier ------------


def test_build_config_emits_tick_skip_value(gdl):
    """Tick mode emits gallery-dl's exit:20 to short-circuit catch-up
    scans once 20 contiguous archived items are seen."""
    from backend.app.services.gallery_dl import TICK_SKIP_VALUE
    cfg = gdl._build_config_for_source(
        platform="subscribestar",
        source_config=SourceConfig(),
        artist_slug="alice",
        skip_value=TICK_SKIP_VALUE,
    )
    assert cfg["extractor"]["skip"] == "exit:20"


def test_build_config_emits_backfill_skip_value(gdl):
    """Backfill mode keeps gallery-dl's default skip=True so it walks the
    full post history."""
    from backend.app.services.gallery_dl import BACKFILL_SKIP_VALUE
    cfg = gdl._build_config_for_source(
        platform="subscribestar",
        source_config=SourceConfig(),
        artist_slug="alice",
        skip_value=BACKFILL_SKIP_VALUE,
    )
    assert cfg["extractor"]["skip"] is True


def test_categorize_partial_when_files_downloaded_then_nonzero_exit(gdl):
    """Non-zero exit + ≥1 file downloaded + no source-level error →
    PARTIAL. Downstream maps to status=ok because the next tick continues."""
    # stdout has a single downloaded file (line starting with `/`); no
    # error-class indicators in stderr; non-zero return code.
    etype, msg = gdl._categorize_error(
        return_code=1,
        stdout="/images/alice/patreon/post1/img.jpg\n",
        stderr="",
    )
    assert etype == ErrorType.PARTIAL
    assert "Downloaded 1 file" in msg


def test_categorize_unknown_when_no_files_and_no_pattern(gdl):
    """PARTIAL only kicks in when files were actually written. A non-zero
    exit with no downloads and no recognized pattern still falls through
    to UNKNOWN_ERROR — distinct from the partial-success case."""
    etype, _ = gdl._categorize_error(
        return_code=2,
        stdout="",
        stderr="some unrecognized noise\n",
    )
    assert etype == ErrorType.UNKNOWN_ERROR


def test_categorize_tier_limited_wins_over_partial(gdl):
    """A run that hit tier-limited warnings AND downloaded some files
    classifies as TIER_LIMITED (the more specific category), not PARTIAL."""
    stdout = "/images/alice/patreon/post1/img.jpg\n"
    stderr = "[patreon][warning] Not allowed to view post 123\n"
    etype, _ = gdl._categorize_error(
        return_code=1, stdout=stdout, stderr=stderr,
    )
    assert etype == ErrorType.TIER_LIMITED


# --- Cursor parsing (plan #689 / native ingester #697) ---------------------
# parse_last_cursor is retained post-cutover: the native Patreon ingester emits
# `Cursor: <token>` per page into DownloadResult.stdout and download_service
# checkpoints the last one. (The gallery-dl Patreon `files`/`cursor` config and
# the Mux yt-dlp Referer block were removed at the #697 cutover, along with the
# tests that pinned them.)


def test_parse_last_cursor_returns_last_match_across_streams():
    # One `Cursor: <token>` line per page; the last is the furthest-progressed
    # resume point.
    stderr = (
        "[patreon][debug] Cursor: 03:AAAA:bbb\n"
        "[urllib3] GET ...\n"
        "[patreon][debug] Cursor: 03:CCCC:ddd\n"
    )
    assert parse_last_cursor("", stderr) == "03:CCCC:ddd"


def test_parse_last_cursor_scans_stdout_too_and_none_when_absent():
    assert parse_last_cursor("Cursor: 99:ZZZ", "") == "99:ZZZ"
    assert parse_last_cursor("no marker here", "still nothing") is None
    assert parse_last_cursor("", "") is None