OverviewFilesCommitsIssuesDocsReleasesPeers

stremio-vsembed / test_subtitle_route.py

142 lines · python ref:
"""The library copy's OWN subtitles, for players that cannot see its folder.

2026-10-18: No Other Choice's English drifted in Kodi although the sync had
corrected it on 10-04. The corrected file only ever reached Jellyfin, which
reads the folder. Kodi plays the copy over HTTP through /mp4/ and Jacktook
fetched wyzie itself into a per-TMDB cache on the Pi -- proven by md5: the Pi's
English.srt (025fc304...) was byte-identical to the UNSYNCED original in
subs-original/, not the corrected sidecar (37840b79...).

So the subtitles that belong to a file are served by the same id the file is:
/subs/<kind>/<ident>.json lists them (Stremio's subtitle object: id, url,
lang), and each url serves the sidecar itself. Live, not baked into the stream
row, because Jacktook caches the row for 24h and a sync can land in between.

Like /web/file/, this must never fetch anything: /mp4/ queues the next episode
and starts an ingest, a subtitle lookup must not.

Run:  .venv/bin/python3 -m pytest test_subtitle_route.py -q
"""

import os

import pytest

import addon


TMDB = {
    "tv/1": {"name": "Show", "first_air_date": "2020-01-01", "seasons": []},
    "tv/1/season/2/episode/6": {"name": "Six", "runtime": 30},
    "tv/1/season/2/episode/7": {"name": "Seven", "runtime": 30},
    "movie/2": {"title": "Film", "release_date": "2021-05-05", "runtime": 30},
}


@pytest.fixture(autouse=True)
def isolated(tmp_path, monkeypatch):
    monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
    monkeypatch.setattr(addon, "INGEST_QUEUE_FILE", str(tmp_path / "queue.json"))
    monkeypatch.setattr(addon, "LIBRARY_VERIFIED_FILE", str(tmp_path / "verified.json"))
    monkeypatch.setattr(addon, "MP4_DIR", str(tmp_path / "mp4-cache"))
    monkeypatch.setattr(addon, "_duration_cache", {})
    monkeypatch.setattr(addon, "_library_meta_cache", {})
    monkeypatch.setattr(addon, "_mp4_jobs", {})
    monkeypatch.setattr(addon, "_tmdb_get", lambda path, params=None: TMDB[path])
    monkeypatch.setattr(addon, "_media_duration", lambda p: 1800.0)

    def refuse(*a, **k):
        raise AssertionError("a subtitle lookup started a download")
    monkeypatch.setattr(addon, "_start_mp4_job", refuse)
    monkeypatch.setattr(addon, "_queue_next_episode", refuse)
    monkeypatch.setattr(addon, "_ensure_ingest", refuse)
    monkeypatch.setattr(addon, "_resolve_vidcore", refuse)
    yield


@pytest.fixture
def client():
    return addon.app.test_client()


def _hold(kind, season=0, episode=0, tmdb="1", subs=("en",)):
    meta = addon._library_meta(kind, f"tmdb:{tmdb}", season, episode)
    path = addon._library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
                               season, episode, meta["episode_title"])
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "wb") as f:
        f.write(b"0123456789" * 100)
    stem = os.path.splitext(path)[0]
    for code in subs:
        with open(f"{stem}.{code}.srt", "w") as f:
            f.write(f"1\n00:00:01,000 --> 00:00:02,000\n{code}\n")
    return path


def test_a_held_movie_lists_its_sidecars(client):
    _hold("movie", tmdb="2", subs=("en", "ko"))
    r = client.get("/subs/movie/tmdb:2.json")
    assert r.status_code == 200
    subs = r.get_json()["subtitles"]
    assert sorted(s["lang"] for s in subs) == ["en", "ko"]
    for s in subs:
        assert s["id"] and s["url"].startswith("http://")


def test_each_listed_url_serves_that_sidecar_byte_for_byte(client):
    path = _hold("movie", tmdb="2", subs=("en", "pt-BR"))
    for s in client.get("/subs/movie/tmdb:2.json").get_json()["subtitles"]:
        body = client.get(s["url"].split("localhost", 1)[1]).get_data()
        with open(f"{os.path.splitext(path)[0]}.{s['lang']}.srt", "rb") as f:
            assert body == f.read()


def test_a_held_episode_lists_its_sidecars(client):
    _hold("series", 2, 6, subs=("en",))
    subs = client.get("/subs/series/tmdb:1:2:6.json").get_json()["subtitles"]
    assert [s["lang"] for s in subs] == ["en"]


def test_a_second_track_in_one_language_keeps_its_own_id(client):
    """`<stem>.en.2.srt` is a different file from `<stem>.en.srt`; both are
    offered, and each url serves its own."""
    path = _hold("movie", tmdb="2", subs=("en", "en.2"))
    subs = client.get("/subs/movie/tmdb:2.json").get_json()["subtitles"]
    assert sorted(s["id"] for s in subs) == ["en", "en.2"]
    assert all(s["lang"] == "en" for s in subs)
    two = next(s for s in subs if s["id"] == "en.2")
    body = client.get(two["url"].split("localhost", 1)[1]).get_data()
    with open(f"{os.path.splitext(path)[0]}.en.2.srt", "rb") as f:
        assert body == f.read()


def test_the_list_carries_each_file_identity(client):
    """So a client can tell a re-synced file from the one it cached."""
    path = _hold("movie", tmdb="2", subs=("en",))
    (s,) = client.get("/subs/movie/tmdb:2.json").get_json()["subtitles"]
    st = os.stat(f"{os.path.splitext(path)[0]}.en.srt")
    assert s["identity"] == [st.st_size, int(st.st_mtime)]


def test_not_held_is_an_empty_list_and_fetches_nothing(client):
    r = client.get("/subs/series/tmdb:1:2:7.json")
    assert r.status_code == 200
    assert r.get_json() == {"subtitles": []}


def test_held_without_sidecars_is_an_empty_list(client):
    _hold("movie", tmdb="2", subs=())
    assert client.get("/subs/movie/tmdb:2.json").get_json() == {"subtitles": []}


def test_a_name_that_is_not_one_of_its_sidecars_is_404(client):
    _hold("movie", tmdb="2", subs=("en",))
    assert client.get("/subs/movie/tmdb:2/de.srt").status_code == 404
    assert client.get("/subs/movie/tmdb:2/..%2F..%2Fverified.json.srt").status_code == 404


def test_bad_ids_are_404(client):
    assert client.get("/subs/movie/x.json").status_code == 404
    assert client.get("/subs/series/tmdb:1.json").status_code == 404
    assert client.get("/subs/anime/tmdb:1.json").status_code == 404