OverviewFilesCommitsIssuesDocsReleasesPeers

stremio-vsembed / test_subtitles.py

381 lines · python ref:
"""Subtitles are fetched alongside a library copy, as sidecar files.

The HLS stream carries NONE. Measured 2026-09-04 on No Other Choice (tmdb
639988): the master lists four muxed A/V variants and no #EXT-X-MEDIA at all, so
there is nothing for `-c copy` to have dropped. The vidcore PLAYER gets them
from a separate endpoint -- `vidcore.net/wyzie?id=<tmdb>[&season&episode]`,
found in the 2026-07-21 deobfuscation -- which returns
`[{display, language, url, encoding}]` and needs no key and no browser. All 32
tracks for that movie fetch in 2.9s / 3.6 MB with 8 workers, every one plain
SRT, so there is no case for being selective.

They are written as sidecars next to the mp4 because that is what Jellyfin
indexes. Kodi cannot discover a sidecar over HTTP -- local copies are served
through `/mp4/...` with send_file -- so the Kodi side fetches its own; this half
is what makes the LIBRARY carry subtitles.

A title with no subtitles is not an error. vidcore answers 404 with a zero-length
body for those (verified: X-Men '97 S2E8), and an ingest must not fail over it.

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

import json
import os

import pytest

import addon


@pytest.fixture
def mp4(tmp_path):
    p = tmp_path / "No Other Choice (2025).mp4"
    p.write_bytes(b"not really an mp4")
    return str(p)


def _tracks(*pairs):
    return [{"display": d, "language": l, "url": f"http://x.invalid/{l}",
             "encoding": "UTF-8"} for l, d in pairs]


@pytest.fixture
def fetched(monkeypatch):
    """Record what was asked for; serve deterministic bytes."""
    calls = []

    def fake_get(url, timeout=None):
        calls.append(url)
        return b"1\n00:00:01,000 --> 00:00:02,000\n" + url.encode() + b"\n"

    monkeypatch.setattr(addon, "_fetch_subtitle", fake_get)
    return calls


# ---------------------------------------------------------------- the endpoint

def test_a_movie_is_asked_for_by_tmdb_id_alone(monkeypatch):
    seen = {}
    monkeypatch.setattr(addon, "_http_json",
                        lambda url, **k: seen.setdefault("url", url) and [])
    addon._wyzie_tracks("639988")
    assert "id=639988" in seen["url"]
    assert "season" not in seen["url"]
    assert "episode" not in seen["url"]


def test_an_episode_carries_season_and_episode(monkeypatch):
    seen = {}
    monkeypatch.setattr(addon, "_http_json",
                        lambda url, **k: seen.setdefault("url", url) and [])
    addon._wyzie_tracks("138502", season=1, episode=1)
    assert "id=138502" in seen["url"]
    assert "season=1" in seen["url"]
    assert "episode=1" in seen["url"]


def test_no_subtitles_is_an_empty_list_not_an_error(monkeypatch):
    """vidcore answers 404 with an empty body for a title Wyzie lacks."""
    def boom(url, **k):
        raise RuntimeError("HTTP 404")
    monkeypatch.setattr(addon, "_http_json", boom)
    assert addon._wyzie_tracks("138502", season=2, episode=8) == []


def test_a_malformed_answer_is_an_empty_list(monkeypatch):
    monkeypatch.setattr(addon, "_http_json", lambda url, **k: {"not": "a list"})
    assert addon._wyzie_tracks("639988") == []


def test_entries_without_a_url_are_dropped(monkeypatch):
    monkeypatch.setattr(addon, "_http_json", lambda url, **k: [
        {"display": "English", "language": "en", "url": "http://x/en"},
        {"display": "Broken", "language": "xx"},
    ])
    got = addon._wyzie_tracks("639988")
    assert [t["language"] for t in got] == ["en"]


# ------------------------------------------------------------------- the files

def test_sidecars_land_next_to_the_mp4_with_the_same_stem(mp4, fetched):
    paths = addon._write_sidecars(mp4, _tracks(("en", "English"), ("fr", "French")))
    assert len(paths) == 2
    for p in paths:
        assert os.path.dirname(p) == os.path.dirname(mp4)
        assert os.path.basename(p).startswith("No Other Choice (2025).")
        assert p.endswith(".srt")
        assert os.path.getsize(p) > 0


def test_the_language_code_is_the_suffix_jellyfin_reads(mp4, fetched):
    paths = addon._write_sidecars(mp4, _tracks(("en", "English")))
    assert os.path.basename(paths[0]) == "No Other Choice (2025).en.srt"


def test_opensubtitles_only_codes_are_mapped_to_ones_jellyfin_RESOLVES(mp4, fetched):
    """The targets are the codes Jellyfin's /Localization/Cultures answers to.

    Not a detail: the first version shipped `yue` and `zh-Hant`, which are
    perfectly good language tags and are simply absent from that table, so
    Jellyfin filed the Cantonese track as "Undefined". Measured, not reasoned.
    """
    paths = addon._write_sidecars(mp4, _tracks(
        ("pb", "Portuguese (BR)"), ("ea", "Spanish (LA)"),
        ("zt", "Chinese (traditional)"), ("zc", "Chinese (Cantonese)")))
    names = sorted(os.path.basename(p) for p in paths)
    assert names == sorted([
        "No Other Choice (2025).pt-BR.srt",     # -> Portuguese (Brazil)
        "No Other Choice (2025).es-419.srt",    # -> Spanish; Latin
        "No Other Choice (2025).zh-TW.srt",     # -> Chinese (Traditional)
        "No Other Choice (2025).zh-HK.srt",     # -> Chinese (Hong Kong)
    ])


def test_a_code_jellyfin_already_knows_is_left_alone(mp4, fetched):
    """`ze` is native to Jellyfin as "Chinese (Bilingual)". Mapping it to `zh`
    threw away a distinction the server was ready to make -- so the map must not
    contain it."""
    paths = addon._write_sidecars(mp4, _tracks(("ze", "Chinese bilingual")))
    assert os.path.basename(paths[0]) == "No Other Choice (2025).ze.srt"


def test_two_tracks_in_one_language_do_not_overwrite_each_other(mp4, fetched):
    paths = addon._write_sidecars(mp4, _tracks(("en", "English"), ("en", "English")))
    assert len(set(paths)) == 2, "the second track overwrote the first"
    assert all(os.path.exists(p) for p in paths)


def test_an_unknown_code_is_passed_through_rather_than_dropped(mp4, fetched):
    paths = addon._write_sidecars(mp4, _tracks(("qq", "Klingon")))
    assert os.path.basename(paths[0]) == "No Other Choice (2025).qq.srt"


def test_a_code_that_would_escape_the_directory_is_refused(mp4, fetched):
    """The code reaches a filename straight from a remote JSON body."""
    paths = addon._write_sidecars(mp4, _tracks(("../../etc/passwd", "Evil")))
    for p in paths:
        assert os.path.dirname(os.path.realpath(p)) == \
            os.path.realpath(os.path.dirname(mp4))


def test_one_failed_download_does_not_lose_the_others(mp4, monkeypatch):
    def flaky(url, timeout=None):
        if url.endswith("/fr"):
            raise RuntimeError("connection reset")
        return b"1\n00:00:01,000 --> 00:00:02,000\nhi\n"
    monkeypatch.setattr(addon, "_fetch_subtitle", flaky)
    paths = addon._write_sidecars(mp4, _tracks(
        ("en", "English"), ("fr", "French"), ("de", "German")))
    assert sorted(os.path.basename(p) for p in paths) == [
        "No Other Choice (2025).de.srt", "No Other Choice (2025).en.srt"]


def test_an_empty_body_is_not_written(mp4, monkeypatch):
    monkeypatch.setattr(addon, "_fetch_subtitle", lambda url, timeout=None: b"")
    assert addon._write_sidecars(mp4, _tracks(("en", "English"))) == []
    assert not [f for f in os.listdir(os.path.dirname(mp4)) if f.endswith(".srt")]


def test_no_tracks_writes_nothing_and_does_not_raise(mp4, fetched):
    assert addon._write_sidecars(mp4, []) == []


def test_a_sidecar_already_on_disk_is_not_refetched(mp4, fetched):
    addon._write_sidecars(mp4, _tracks(("en", "English")))
    assert len(fetched) == 1
    addon._write_sidecars(mp4, _tracks(("en", "English")))
    assert len(fetched) == 1, "refetched a subtitle already on disk"


# ------------------------------------------------------- wired into the ingest

@pytest.fixture
def ingest(monkeypatch, tmp_path):
    """A _run_ingest with everything outside the subtitle question stubbed."""
    state = {"tracks": [{"display": "English", "language": "en",
                         "url": "http://x.invalid/en"}],
             "sidecars": [], "remuxed": False, "have": False}

    monkeypatch.setattr(addon, "ingest_source", lambda: object())
    monkeypatch.setattr(addon, "_enough_space", lambda: True)
    monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: {
        "title": "No Other Choice", "year": "2025", "tmdb_id": "639988",
        "episode_title": ""})
    monkeypatch.setattr(addon, "LIBRARY_ROOT", str(tmp_path))
    monkeypatch.setattr(addon, "_library_has", lambda *a, **k: state["have"])
    monkeypatch.setattr(addon, "_resolve_vidcore",
                        lambda *a, **k: "http://x.invalid/master.m3u8")
    monkeypatch.setattr(addon, "_cap_variant", lambda url, h: url)

    def fake_remux(variant, dest):
        state["remuxed"] = True
        os.makedirs(os.path.dirname(dest), exist_ok=True)
        with open(dest, "wb") as f:
            f.write(b"video")

    monkeypatch.setattr(addon, "_remux_to_mp4", fake_remux)
    monkeypatch.setattr(addon, "_wyzie_tracks",
                        lambda *a, **k: state["tracks"])

    def fake_sidecars(dest, tracks):
        state["sidecars"].append((dest, list(tracks)))
        return [dest + ".en.srt"] if tracks else []

    monkeypatch.setattr(addon, "_write_sidecars", fake_sidecars)
    return state


def test_a_finished_copy_gets_its_subtitles(ingest):
    assert addon._run_ingest("movie", "tmdb:639988", 0, 0, 1080) == "done"
    assert len(ingest["sidecars"]) == 1
    dest, tracks = ingest["sidecars"][0]
    assert dest.endswith("No Other Choice (2025).mp4")
    assert [t["language"] for t in tracks] == ["en"]


def test_a_copy_already_in_the_library_is_backfilled(ingest):
    """The early return exists so an existing copy costs no browser. It should
    still gain subtitles -- that is the whole library, not just new fetches."""
    ingest["have"] = True
    assert addon._run_ingest("movie", "tmdb:639988", 0, 0, 1080) == "done"
    assert not ingest["remuxed"], "resolved and remuxed a copy we already hold"
    assert len(ingest["sidecars"]) == 1


def test_a_title_with_no_subtitles_still_succeeds(ingest):
    ingest["tracks"] = []
    assert addon._run_ingest("movie", "tmdb:639988", 0, 0, 1080) == "done"


def test_a_failed_subtitle_fetch_never_fails_the_ingest(ingest, monkeypatch):
    """The copy is complete and published before this runs. Nothing about a
    subtitle may take that away."""
    def boom(*a, **k):
        raise RuntimeError("wyzie is down")
    monkeypatch.setattr(addon, "_wyzie_tracks", boom)
    assert addon._run_ingest("movie", "tmdb:639988", 0, 0, 1080) == "done"


def test_a_failed_sidecar_write_never_fails_the_ingest(ingest, monkeypatch):
    def boom(*a, **k):
        raise OSError("read-only filesystem")
    monkeypatch.setattr(addon, "_write_sidecars", boom)
    assert addon._run_ingest("movie", "tmdb:639988", 0, 0, 1080) == "done"


def test_an_episode_passes_its_season_and_episode_through(ingest, monkeypatch):
    seen = {}
    monkeypatch.setattr(addon, "_wyzie_tracks",
                        lambda tmdb, season=0, episode=0: seen.update(
                            tmdb=tmdb, season=season, episode=episode) or [])
    monkeypatch.setattr(addon, "_library_meta", lambda *a, **k: {
        "title": "X-Men 97", "year": "2024", "tmdb_id": "138502",
        "episode_title": "Pilot"})
    addon._run_ingest("series", "tmdb:138502", 1, 1, 1080)
    assert seen == {"tmdb": "138502", "season": 1, "episode": 1}


# ------------------------------------------- OpenSubtitles, when wyzie has none
# 2026-10-18, Hope (tmdb 1058424): wyzie LISTED 9 tracks and answered 503 for
# every file, so the copy landed with no subtitles at all. OpenSubtitles'
# stremio addon (the host Jacktook already uses) had 4 English tracks, each
# named for the release it was cut for, and all 4 measured 0.13 s / 1.000
# against our copy. It is asked only when no English sidecar is left after
# wyzie and the sync, and what it gives goes through the same sync.

OS_BODY = {"subtitles": [
    {"id": "1", "url": "http://os.invalid/1", "lang": "eng"},
    {"id": "2", "url": "http://os.invalid/2", "lang": "kor"},
    {"id": "3", "url": "http://os.invalid/3", "lang": "eng"},
    {"id": "4", "url": "http://os.invalid/4", "lang": "eng"},
    {"id": "5", "url": "http://os.invalid/5", "lang": "eng"},
    {"id": "6", "lang": "eng"},
]}


@pytest.fixture
def os_http(monkeypatch):
    seen = []

    class R:
        status_code = 200

        def json(self):
            return OS_BODY

    def fake_get(url, **k):
        seen.append(url)
        return R()
    monkeypatch.setattr(addon.requests, "get", fake_get)
    return seen


def test_opensubs_asks_a_movie_by_imdb_id(os_http):
    addon._opensubs_tracks("movie", "tt27369017")
    assert os_http == [f"{addon.OPENSUBS_URL}/subtitles/movie/tt27369017.json"]


def test_opensubs_asks_an_episode_with_season_and_episode(os_http):
    addon._opensubs_tracks("series", "tt0903747", 2, 6)
    assert os_http == [f"{addon.OPENSUBS_URL}/subtitles/series/tt0903747:2:6.json"]


def test_opensubs_keeps_english_with_a_url_capped(os_http):
    tracks = addon._opensubs_tracks("movie", "tt27369017")
    assert [t["url"] for t in tracks] == [
        "http://os.invalid/1", "http://os.invalid/3", "http://os.invalid/4"]
    assert all(t["language"] == "en" for t in tracks)


def test_opensubs_failure_is_an_empty_list(monkeypatch):
    def boom(*a, **k):
        raise RuntimeError("down")
    monkeypatch.setattr(addon.requests, "get", boom)
    assert addon._opensubs_tracks("movie", "tt1") == []


REAL_OPENSUBS_FOR = addon._opensubs_for      # before conftest refuses it


@pytest.fixture
def fallback(mp4, monkeypatch, fetched):
    state = {"wyzie": [], "os": [{"language": "en", "url": "http://os.invalid/1"}],
             "asked": [], "synced": 0}
    monkeypatch.setattr(addon, "_wyzie_tracks", lambda *a, **k: state["wyzie"])
    monkeypatch.setattr(addon, "_opensubs_for", REAL_OPENSUBS_FOR)
    monkeypatch.setattr(addon, "_imdb_id_for", lambda kind, ident: "tt27369017")

    def fake_os(kind, imdb, season=0, episode=0):
        state["asked"].append((kind, imdb, season, episode))
        return state["os"]
    monkeypatch.setattr(addon, "_opensubs_tracks", fake_os)

    def fake_sync(dest):
        state["synced"] += 1
        return {}
    monkeypatch.setattr(addon, "_sync_subtitles", fake_sync)
    return state


def test_no_english_from_wyzie_asks_opensubtitles(mp4, fallback):
    meta = {"tmdb_id": "1058424"}
    addon._fetch_subtitles_for(mp4, meta, 0, 0, kind="movie")
    assert fallback["asked"] == [("movie", "tt27369017", 0, 0)]
    assert os.path.exists(os.path.splitext(mp4)[0] + ".en.srt")
    assert fallback["synced"] == 2, "the OpenSubtitles track was not measured"


def test_english_from_wyzie_never_asks_opensubtitles(mp4, fallback):
    fallback["wyzie"] = _tracks(("en", "English"))
    addon._fetch_subtitles_for(mp4, {"tmdb_id": "1058424"}, 0, 0, kind="movie")
    assert fallback["asked"] == []


def test_an_opensubtitles_failure_never_raises(mp4, fallback, monkeypatch):
    def boom(*a, **k):
        raise RuntimeError("tmdb down")
    monkeypatch.setattr(addon, "_imdb_id_for", boom)
    addon._fetch_subtitles_for(mp4, {"tmdb_id": "1058424"}, 0, 0, kind="movie")