ytstream/tests/test_runner.py

"""Orchestration: ordering, the lock, and what turns the cron check red."""

from __future__ import annotations

import pytest

from ytstream import config, discovery, jellyfin, reap, runner, subsync, videos

from conftest import add_video


@pytest.fixture()
def quiet_jellyfin(monkeypatch):
    """Jellyfin refresh is best effort; record calls instead of making them."""
    calls = []
    monkeypatch.setattr(jellyfin.Jellyfin, "refresh", lambda self: calls.append(1))
    return calls


@pytest.fixture()
def no_thumbs(monkeypatch):
    from ytstream import strm

    monkeypatch.setattr(strm, "fetch_thumbnail", lambda *a, **k: False)


# ----------------------------------------------------------------------- lock


def test_lock_is_exclusive(tmp_path):
    path = tmp_path / "run.lock"
    with runner.exclusive_lock(path):
        with pytest.raises(runner.AlreadyRunning):
            with runner.exclusive_lock(path):
                pass  # pragma: no cover


def test_lock_is_released_after_use(tmp_path):
    path = tmp_path / "run.lock"
    with runner.exclusive_lock(path):
        pass
    with runner.exclusive_lock(path):
        pass


def test_lock_is_released_even_when_the_body_raises(tmp_path):
    path = tmp_path / "run.lock"
    with pytest.raises(ValueError):
        with runner.exclusive_lock(path):
            raise ValueError("boom")
    with runner.exclusive_lock(path):
        pass


# --------------------------------------------------------------- materialising


def test_materialise_all_writes_everything_listed(conn, settings, media_root,
                                                  channel, no_thumbs):
    for index in range(3):
        add_video(conn, channel["id"], f"vid{index:08d}")

    stats = runner.materialise_all(conn, settings)

    assert stats["materialised"] == 3
    assert stats["errors"] == 0
    assert videos.queue_depth(conn) == 0


def test_tvshow_nfo_is_written_once_per_channel(conn, settings, media_root,
                                                channel, no_thumbs):
    for index in range(3):
        add_video(conn, channel["id"], f"vid{index:08d}")

    stats = runner.materialise_all(conn, settings)

    assert stats["shows"] == 1
    assert (media_root / "clabretro" / "tvshow.nfo").exists()


def test_channel_with_nothing_in_the_window_creates_no_directory(
    conn, settings, media_root, channel, no_thumbs
):
    """52 of 117 measured channels are in this state, and an empty series in
    Jellyfin looks like a bug rather than a quiet channel."""
    add_video(conn, channel["id"], "vid00000001", state=videos.SKIPPED_OLD)

    runner.materialise_all(conn, settings)

    assert not (media_root / "clabretro").exists()


def test_materialise_all_honours_the_limit(conn, settings, media_root, channel,
                                           no_thumbs):
    for index in range(5):
        add_video(conn, channel["id"], f"vid{index:08d}")
    stats = runner.materialise_all(conn, settings, limit=2)
    assert stats["materialised"] == 2


def test_materialise_skips_a_video_whose_channel_vanished(conn, settings,
                                                         media_root, channel,
                                                         no_thumbs):
    add_video(conn, channel["id"], "vid00000001")
    # Simulate the channel being unsubscribed between claiming and writing, with
    # foreign keys off so the row survives to be found.
    conn.execute("PRAGMA foreign_keys = OFF")
    with conn:
        conn.execute("DELETE FROM channel WHERE id = ?", (channel["id"],))

    stats = runner.materialise_all(conn, settings)

    assert stats["materialised"] == 0
    assert stats["errors"] == 0


# ------------------------------------------------------------------ full cycle


def test_run_order_is_sync_poll_materialise_reap(conn, settings, media_root,
                                                 channel, monkeypatch,
                                                 quiet_jellyfin):
    order = []
    monkeypatch.setattr(subsync, "sync_all",
                        lambda *a: order.append("sync") or {"added": 0, "queued": 0,
                                                            "removed": 0, "refused": 0,
                                                            "sources": 0,
                                                            "pending_removal": 0})
    monkeypatch.setattr(discovery, "poll_all",
                        lambda *a: order.append("poll") or {"queued": 0})
    monkeypatch.setattr(runner, "materialise_all",
                        lambda *a, **k: order.append("materialise") or
                        {"materialised": 0, "shows": 0, "errors": 0})
    monkeypatch.setattr(reap, "run",
                        lambda *a: order.append("reap") or {"aged_out": 0})

    runner.run(conn, settings)

    assert order == ["sync", "poll", "materialise", "reap"]


def test_single_channel_run_skips_the_sync(conn, settings, channel, monkeypatch,
                                           quiet_jellyfin):
    """A targeted run is an operator action, not a mirror pass."""
    monkeypatch.setattr(subsync, "sync_all",
                        lambda *a: pytest.fail("sync should not run"))
    monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0})
    monkeypatch.setattr(reap, "run", lambda *a: {"aged_out": 0})

    result = runner.run(conn, settings, channel_pk=channel["id"])

    assert "sync" not in result


def test_run_records_last_run_at(conn, settings, channel, monkeypatch,
                                 quiet_jellyfin):
    monkeypatch.setattr(subsync, "sync_all", lambda *a: {"added": 0, "queued": 0,
                                                         "removed": 0, "refused": 0,
                                                         "sources": 0,
                                                         "pending_removal": 0})
    monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0})
    monkeypatch.setattr(reap, "run", lambda *a: {"aged_out": 0})

    runner.run(conn, settings)

    assert settings.raw("last_run_at")


def test_jellyfin_is_refreshed_only_when_the_tree_changed(
    conn, settings, media_root, channel, monkeypatch, quiet_jellyfin, no_thumbs
):
    monkeypatch.setattr(subsync, "sync_all", lambda *a: {"added": 0, "queued": 0,
                                                         "removed": 0, "refused": 0,
                                                         "sources": 0,
                                                         "pending_removal": 0})
    monkeypatch.setattr(discovery, "poll_all", lambda *a: {"queued": 0})

    runner.run(conn, settings)
    assert quiet_jellyfin == []

    add_video(conn, channel["id"], "vid00000001")
    runner.run(conn, settings)
    assert len(quiet_jellyfin) == 1


# ------------------------------------------------------------------ exit codes


def test_refused_sync_turns_the_check_red():
    """A silently-broken mirror is the worst outcome available: nothing looks
    wrong until someone asks why a channel never appeared."""
    assert runner.exit_code({"sync": {"refused": 1}}) == 1


def test_materialise_errors_turn_the_check_red():
    assert runner.exit_code({"materialise": {"errors": 2}}) == 1


def test_poll_failures_alone_do_not_turn_the_check_red():
    """Two of 119 measured channels fail permanently — terminated or private."""
    assert runner.exit_code({"poll": {"failed": 2}}) == 0


def test_clean_run_is_zero():
    assert runner.exit_code({"sync": {"refused": 0}, "poll": {"failed": 0},
                             "materialise": {"errors": 0}}) == 0


# ------------------------------------------------------------------- summarise


def test_summarise_mentions_a_refusal():
    text = runner.summarise({
        "sync": {"added": 0, "queued": 0, "removed": 0, "refused": 1},
        "poll": {"channels": 5, "queued": 0},
        "materialise": {"materialised": 0},
        "reap": {"aged_out": 0},
    })
    assert "SYNC_REFUSED" in text


def test_summarise_is_a_single_line():
    text = runner.summarise({
        "sync": {"added": 1, "queued": 2, "removed": 3, "refused": 0},
        "poll": {"channels": 119, "queued": 4, "shorts": 5, "live": 6, "failed": 2},
        "materialise": {"materialised": 7, "errors": 0},
        "reap": {"aged_out": 8},
    })
    assert "\n" not in text
    assert "channels=119" in text
    assert "materialised=7" in text
    assert "aged_out=8" in text


def test_run_prunes_channels_that_have_nothing_to_show(conn, settings, media_root,
                                                       monkeypatch):
    """End to end: a subscribed channel whose directory exists only because
    subscribe() wrote tvshow.nfo must not survive a run as an empty series."""
    from conftest import add_channel
    from ytstream import runner

    empty = add_channel(conn, "UC" + "s" * 22, "Dead Channel", "dead-channel")
    tree = media_root / "dead-channel"
    tree.mkdir()
    (tree / "tvshow.nfo").write_text("<tvshow/>")

    assert runner.prune_empty_channels(conn) == 1
    assert not tree.exists()


def test_run_keeps_a_channel_with_a_materialised_video(conn, settings, media_root):
    from conftest import add_channel, add_video
    from ytstream import runner, videos

    row = add_channel(conn, "UC" + "t" * 22, "Alive", "alive")
    tree = media_root / "alive"
    (tree / "Season 2026").mkdir(parents=True)
    (tree / "Season 2026" / "ep.strm").write_text("url")
    video = add_video(conn, row["id"], "vid00000009")
    videos.mark_materialised(conn, "vid00000009",
                             rel_path="alive/Season 2026/ep.strm", season=2026,
                             episode=1, upload_date="2026-08-01", duration=900,
                             title="t")

    assert runner.prune_empty_channels(conn) == 0
    assert tree.exists()