ytstream/tests/test_strm.py

"""Materialising: .strm contents, NFO sidecars, naming, and idempotency."""

from __future__ import annotations

from datetime import date

import pytest

from ytstream import config, strm, videos

from conftest import add_video


@pytest.fixture()
def video(conn, channel):
    return add_video(conn, channel["id"], "dQw4w9WgXcQ",
                     title="A Video: With/Punctuation",
                     upload_date="2026-08-12", duration=1337)


@pytest.fixture()
def no_thumbs(monkeypatch):
    """Thumbnails are a network fetch; every test here runs without one."""
    monkeypatch.setattr(strm, "fetch_thumbnail", lambda *a, **k: False)


def test_strm_contains_only_the_proxy_url(conn, settings, media_root, channel,
                                          video, no_thumbs):
    result = strm.materialise(conn, settings, channel, video)

    path = media_root / result["rel_path"]
    assert path.read_text() == "http://127.0.0.1:8099/watch/dQw4w9WgXcQ"
    # No trailing newline, and nothing else in the file.
    assert path.read_bytes() == b"http://127.0.0.1:8099/watch/dQw4w9WgXcQ"


def test_strm_url_follows_the_configured_base(conn, settings, media_root, channel,
                                              video, no_thumbs):
    settings.set("proxy_base_url", "http://127.0.0.1:9999/")
    result = strm.materialise(conn, settings, channel, video)
    assert (media_root / result["rel_path"]).read_text().startswith(
        "http://127.0.0.1:9999/watch/"
    )


def test_layout_matches_the_naming_scheme(conn, settings, media_root, channel,
                                          video, no_thumbs):
    result = strm.materialise(conn, settings, channel, video)

    # Season = upload year, episode = MMDD*10 + ordinal.
    assert result["season"] == 2026
    assert result["episode"] == 8120
    assert result["rel_path"] == (
        "clabretro/Season 2026/"
        "clabretro - S2026E8120 - A Video With Punctuation [dQw4w9WgXcQ].strm"
    )


def test_nfo_is_written_alongside(conn, settings, media_root, channel, video,
                                  no_thumbs):
    result = strm.materialise(conn, settings, channel, video)
    nfo_path = (media_root / result["rel_path"]).with_suffix(".nfo")

    text = nfo_path.read_text()

    assert "<season>2026</season>" in text
    assert "<episode>8120</episode>" in text
    assert "<aired>2026-08-12</aired>" in text
    # durationinseconds is what stops a .strm episode showing a zero runtime
    # before it has ever been played.
    assert "<durationinseconds>1337</durationinseconds>" in text
    assert 'type="youtube"' in text


def test_nfo_has_no_streamdetails(conn, settings, media_root, channel, video,
                                  no_thumbs):
    """Pre-seeding them was measured to change nothing about Jellyfin probing."""
    result = strm.materialise(conn, settings, channel, video)
    text = (media_root / result["rel_path"]).with_suffix(".nfo").read_text()
    assert "streamdetails" not in text
    assert "fileinfo" not in text


def test_materialise_is_idempotent(conn, settings, media_root, channel, video,
                                   no_thumbs):
    first = strm.materialise(conn, settings, channel, video)
    before = (media_root / first["rel_path"]).read_bytes()

    second = strm.materialise(
        conn, settings, channel, videos.get(conn, "dQw4w9WgXcQ")
    )

    assert second["rel_path"] == first["rel_path"]
    assert (media_root / second["rel_path"]).read_bytes() == before


def test_row_is_marked_materialised(conn, settings, media_root, channel, video,
                                    no_thumbs):
    strm.materialise(conn, settings, channel, video)
    row = videos.get(conn, "dQw4w9WgXcQ")
    assert row["state"] == videos.MATERIALISED
    assert row["rel_path"]
    assert row["materialised_at"]


def test_episode_ordinals_increment_within_a_day(conn, settings, media_root,
                                                 channel, no_thumbs):
    for index in range(3):
        row = add_video(conn, channel["id"], f"vid{index:08d}",
                        upload_date="2026-08-12")
        result = strm.materialise(conn, settings, channel, row)
        assert result["episode"] == 8120 + index


def test_ordinals_are_stable_when_an_earlier_video_ages_out(
    conn, settings, media_root, channel, no_thumbs
):
    """Aged-out rows keep their episode number, so later ordinals never shift."""
    first = add_video(conn, channel["id"], "vid00000001", upload_date="2026-08-12")
    strm.materialise(conn, settings, channel, first)
    videos.mark_aged_out(conn, "vid00000001")

    second = add_video(conn, channel["id"], "vid00000002", upload_date="2026-08-12")
    result = strm.materialise(conn, settings, channel, second)

    assert result["episode"] == 8121


def test_title_falls_back_to_the_video_id(conn, settings, media_root, channel,
                                          no_thumbs):
    """Backfilled rows carry no title until the feed supplies one."""
    row = add_video(conn, channel["id"], "vid00000001", title="",
                    upload_date="2026-08-12")
    result = strm.materialise(conn, settings, channel, row)
    assert "vid00000001" in result["rel_path"]


def test_write_show_creates_tvshow_nfo(media_root, channel):
    strm.write_show(channel)
    text = (media_root / "clabretro" / "tvshow.nfo").read_text()
    assert "<title>clabretro</title>" in text
    assert channel["channel_id"] in text


def test_show_nfo_uses_the_channel_title_not_a_video_title(conn, media_root, channel):
    """The channel/video join has `title` on both sides, and reading the wrong one
    renamed every series after whichever episode happened to be first."""
    add_video(conn, channel["id"], "vid00000001", title="Some Episode Title")
    strm.write_show(channel)
    text = (media_root / "clabretro" / "tvshow.nfo").read_text()
    assert "Some Episode Title" not in text


def test_remove_deletes_strm_and_sidecars(conn, settings, media_root, channel,
                                          video, no_thumbs):
    result = strm.materialise(conn, settings, channel, video)
    strm_path = media_root / result["rel_path"]
    stem = strm_path.name[: -len(".strm")]
    thumb = strm_path.with_name(stem + "-thumb.jpg")
    thumb.write_bytes(b"x" * 2000)

    removed = strm.remove(videos.get(conn, "dQw4w9WgXcQ"))

    assert removed == 3          # .strm, .nfo, -thumb.jpg
    assert not strm_path.exists()
    assert not thumb.exists()


def test_remove_leaves_files_it_does_not_own(conn, settings, media_root, channel,
                                             video, no_thumbs):
    result = strm.materialise(conn, settings, channel, video)
    season_dir = (media_root / result["rel_path"]).parent
    stranger = season_dir / "someone-elses-file.txt"
    stranger.write_text("not ours")

    strm.remove(videos.get(conn, "dQw4w9WgXcQ"))

    assert stranger.exists()


def test_remove_channel_tree(conn, media_root, channel):
    tree = media_root / "clabretro"
    (tree / "Season 2026").mkdir(parents=True)
    (tree / "tvshow.nfo").write_text("<tvshow/>")

    assert strm.remove_channel_tree(channel) is True
    assert not tree.exists()


def test_remove_channel_tree_refuses_the_media_root(conn, media_root):
    from conftest import add_channel

    row = add_channel(conn, "UC" + "q" * 22, "Blank", "")
    keep = media_root / "keep"
    keep.mkdir()

    assert strm.remove_channel_tree(row) is False
    assert keep.exists()


def test_fetch_thumbnail_rejects_the_grey_placeholder(monkeypatch, tmp_path):
    """YouTube serves a tiny placeholder rather than a 404 for missing maxres."""
    import urllib.request

    class Response:
        status = 200

        def __init__(self, payload):
            self.payload = payload

        def read(self):
            return self.payload

        def __enter__(self):
            return self

        def __exit__(self, *exc):
            return False

    calls = []

    def fake_open(request, timeout=None):
        calls.append(request.full_url)
        # maxres returns the placeholder; hq returns something real.
        return Response(b"x" * 120 if "maxres" in request.full_url else b"y" * 5000)

    monkeypatch.setattr(urllib.request, "urlopen", fake_open)
    destination = tmp_path / "out-thumb.jpg"

    assert strm.fetch_thumbnail("dQw4w9WgXcQ", destination) is True
    assert destination.read_bytes() == b"y" * 5000
    assert len(calls) == 2


def test_fetch_thumbnail_is_skipped_when_one_already_exists(monkeypatch, tmp_path):
    import urllib.request

    monkeypatch.setattr(urllib.request, "urlopen",
                        lambda *a, **k: pytest.fail("should not fetch"))
    existing = tmp_path / "out-thumb.jpg"
    existing.write_bytes(b"cached")

    assert strm.fetch_thumbnail("dQw4w9WgXcQ", existing) is True


def test_untitled_video_does_not_get_its_id_written_back_as_a_title(
    conn, settings, media_root, channel, no_thumbs
):
    """Writing the fallback back to the database makes the row look titled, which
    permanently disables the title repair in discovery._record. That shipped once
    and left five of twenty episodes named after their video ids."""
    row = add_video(conn, channel["id"], "vid00000001", title="",
                    upload_date="2026-08-12")

    result = strm.materialise(conn, settings, channel, row)

    # The filename falls back to the id...
    assert "vid00000001]" in result["rel_path"]
    # ...but the row stays untitled, so a later feed poll can still repair it.
    assert videos.get(conn, "vid00000001")["title"] == ""


def test_episode_number_is_stable_across_rematerialising(
    conn, settings, media_root, channel, no_thumbs
):
    """`materialise --all` is the documented recovery from a Jellyfin metadata
    wipe. If it renumbered episodes, the recovery would create a second copy of
    every episode instead of repairing the first."""
    ids = []
    for index in range(3):
        row = add_video(conn, channel["id"], f"vid{index:08d}",
                        upload_date="2026-08-12")
        ids.append(strm.materialise(conn, settings, channel, row)["episode"])

    again = [strm.materialise(conn, settings, channel, videos.get(conn, f"vid{i:08d}"))
             ["episode"] for i in range(3)]

    assert again == ids == [8120, 8121, 8122]


def test_rematerialising_leaves_no_orphans(conn, settings, media_root, channel,
                                           no_thumbs):
    for index in range(3):
        add_video(conn, channel["id"], f"vid{index:08d}", upload_date="2026-08-12")
    for index in range(3):
        strm.materialise(conn, settings, channel, videos.get(conn, f"vid{index:08d}"))
    for index in range(3):
        strm.materialise(conn, settings, channel, videos.get(conn, f"vid{index:08d}"))

    assert len(list(media_root.rglob("*.strm"))) == 3
    assert len([p for p in media_root.rglob("*.nfo") if p.name != "tvshow.nfo"]) == 3


def test_a_renamed_episode_removes_its_old_files(conn, settings, media_root,
                                                 channel, no_thumbs):
    """The late-title path renames the file; the old one must not survive."""
    row = add_video(conn, channel["id"], "vid00000001", title="",
                    upload_date="2026-08-12")
    first = strm.materialise(conn, settings, channel, row)
    old = media_root / first["rel_path"]
    assert old.exists()

    with conn:
        conn.execute("UPDATE video SET title = ? WHERE video_id = ?",
                     ("Proper Title", "vid00000001"))
    second = strm.materialise(conn, settings, channel,
                              videos.get(conn, "vid00000001"))

    assert second["rel_path"] != first["rel_path"]
    assert not old.exists()
    assert (media_root / second["rel_path"]).exists()
    assert len(list(media_root.rglob("*.strm"))) == 1