OverviewFilesCommitsIssuesDocsReleasesPeers

kodi-jacktook / plugin.video.jacktook / tests / unit / test_wyzie_subtitles.py

487 lines · python ref:
"""Subtitles from the source the player itself uses.

The streams we play carry no subtitles. Measured 2026-09-04 on No Other Choice
(tmdb 639988): the HLS master lists four muxed A/V variants and no
#EXT-X-MEDIA at all. The vidcore player gets them from a separate endpoint --
`vidcore.net/wyzie?id=<tmdb|imdb>[&season&episode]`, found in the 2026-07-21
deobfuscation -- which returns `[{display, language, url, encoding}]`, needs no
key and no browser, and answers for imdb ids as readily as tmdb ones (verified:
id=1396 and id=tt0903747 return byte-identical bodies). That makes it usable for
every source Jacktook plays, not only vidcore.

Run:  python3 -m pytest plugin.video.jacktook/tests/unit/test_wyzie_subtitles.py -q
"""

import os
from unittest.mock import MagicMock, patch

import pytest

from lib.clients.subtitle.wyzie import WyzieClient


@pytest.fixture
def client():
    return WyzieClient(MagicMock())


def _resp(payload, status=200):
    r = MagicMock()
    r.status_code = status
    r.json.return_value = payload
    return r


TRACKS = [
    {"display": "English", "language": "en", "url": "http://x/en", "encoding": "UTF-8"},
    {"display": "French", "language": "fr", "url": "http://x/fr", "encoding": "UTF-8"},
]


# ------------------------------------------------------------ asking for them

class TestQuery:
    def test_a_movie_is_asked_for_by_id_alone(self, client):
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(TRACKS)) as g:
            client.get_subtitles("movies", "639988")
        url = g.call_args[0][0]
        assert "id=639988" in url
        assert "season" not in url and "episode" not in url

    def test_an_episode_carries_season_and_episode(self, client):
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(TRACKS)) as g:
            client.get_subtitles("tv", "138502", season=1, episode=1)
        url = g.call_args[0][0]
        assert "id=138502" in url and "season=1" in url and "episode=1" in url

    def test_an_imdb_id_is_passed_through_unchanged(self, client):
        """Verified against the live endpoint: tt0903747 and 1396 return the
        same body, so no id translation is needed or wanted."""
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(TRACKS)) as g:
            client.get_subtitles("tv", "tt0903747", season=1, episode=1)
        assert "id=tt0903747" in g.call_args[0][0]

    def test_a_referer_is_sent(self, client):
        """The vidcore CDN 403s a request without one, and the /wyzie proxy is
        on the same host."""
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(TRACKS)) as g:
            client.get_subtitles("movies", "639988")
        headers = g.call_args.kwargs.get("headers") or {}
        assert "vidcore" in headers.get("Referer", "")


class TestNoResults:
    def test_a_404_is_no_subtitles_not_an_error(self, client):
        """Verified: X-Men '97 S2E8 answers 404 with a zero-length body while
        S1E1 returns 34 tracks. A title with none is ordinary."""
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(None, status=404)):
            assert client.get_subtitles("tv", "138502", season=2, episode=8) == []

    def test_a_network_failure_is_no_subtitles(self, client):
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   side_effect=OSError("no route to host")):
            assert client.get_subtitles("movies", "639988") == []

    def test_a_non_list_body_is_no_subtitles(self, client):
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp({"error": "nope"})):
            assert client.get_subtitles("movies", "639988") == []

    def test_entries_without_a_url_are_dropped(self, client):
        body = [{"display": "English", "language": "en", "url": "http://x/en"},
                {"display": "Broken", "language": "xx"}]
        with patch("lib.clients.subtitle.wyzie.requests.get",
                   return_value=_resp(body)):
            got = client.get_subtitles("movies", "639988")
        assert [t["language"] for t in got] == ["en"]


# ---------------------------------------------------------------- downloading

@pytest.fixture
def downloads(monkeypatch):
    calls = []

    def fake(url, timeout=None, headers=None):
        calls.append(url)
        r = MagicMock()
        r.status_code = 200
        r.content = b"1\n00:00:01,000 --> 00:00:02,000\nhello\n"
        return r

    monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", fake)
    return calls


class TestDownload:
    def test_the_filename_is_the_language_and_nothing_else(self, client, downloads, tmp_path):
        """Kodi labels a subtitle by its filename, and the folder is already
        per-title, so the title in the name buys nothing and cost a lot:

        measured on the Pi, `data["title"]` for a stremio row is the ROW LABEL,
        not the film. Every one of the 32 tracks was written as
        `in library  seekable\nlocal copy.<Language>.srt` -- a newline inside a
        filename (slugify_title keeps \s), and a subtitle menu in which every
        entry read "in library seekable local copy".
        """
        paths = client.download_subtitles_batch(
            TRACKS, title="No Other Choice", folder_path=str(tmp_path))
        names = sorted(os.path.basename(p) for p in paths)
        assert names == ["English.srt", "French.srt"]
        assert all(os.path.getsize(p) > 0 for p in paths)

    def test_a_row_label_never_reaches_the_filename(self, client, downloads, tmp_path):
        paths = client.download_subtitles_batch(
            TRACKS, title="in library \u2014 seekable\nlocal copy",
            folder_path=str(tmp_path))
        for p in paths:
            n = os.path.basename(p)
            assert "\n" not in n and "library" not in n, repr(n)

    def test_control_characters_in_a_label_are_flattened(self, client, downloads, tmp_path):
        """`display` is remote text and lands in a filename."""
        tracks = [{"display": "Eng\nlish\ttext", "language": "en", "url": "http://x/en"}]
        paths = client.download_subtitles_batch(
            tracks, title="Film", folder_path=str(tmp_path))
        n = os.path.basename(paths[0])
        assert "\n" not in n and "\t" not in n, repr(n)

    def test_the_display_name_is_used_because_kodi_labels_by_filename(
            self, client, downloads, tmp_path):
        """wyzie's own label is better than anything we could map a code to:
        it distinguishes 'Portuguese (BR)' from 'Portuguese', which pb/pt would
        only do if we happened to know that mapping.

        The brackets do not survive slugify_title -- the repo's existing
        sanitiser, used here rather than a second one that keeps them, because
        the distinction is what matters and 'Portuguese BR' carries it."""
        tracks = [{"display": "Portuguese (BR)", "language": "pb", "url": "http://x/pb"},
                  {"display": "Portuguese", "language": "pt", "url": "http://x/pt"}]
        paths = client.download_subtitles_batch(
            tracks, title="Film", folder_path=str(tmp_path))
        names = sorted(os.path.basename(p) for p in paths)
        assert names == ["Portuguese BR.srt", "Portuguese.srt"]
        assert len(set(names)) == 2

    def test_two_tracks_with_the_same_label_do_not_collide(
            self, client, downloads, tmp_path):
        tracks = [dict(TRACKS[0]), dict(TRACKS[0], url="http://x/en2")]
        paths = client.download_subtitles_batch(
            tracks, title="Film", folder_path=str(tmp_path))
        assert len(set(paths)) == 2
        assert all(os.path.exists(p) for p in paths)

    def test_a_label_cannot_escape_the_folder(self, client, downloads, tmp_path):
        """`display` arrives inside a remote JSON body and reaches a filename."""
        tracks = [{"display": "../../../etc/passwd", "language": "en",
                   "url": "http://x/en"}]
        paths = client.download_subtitles_batch(
            tracks, title="Film", folder_path=str(tmp_path))
        for p in paths:
            assert os.path.dirname(os.path.realpath(p)) == os.path.realpath(str(tmp_path))

    def test_one_failure_does_not_lose_the_rest(self, client, monkeypatch, tmp_path):
        def flaky(url, timeout=None, headers=None):
            r = MagicMock()
            if url.endswith("/fr"):
                raise OSError("connection reset")
            r.status_code = 200
            r.content = b"1\n00:00:01,000 --> 00:00:02,000\nhi\n"
            return r
        monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", flaky)
        paths = client.download_subtitles_batch(
            TRACKS, title="Film", folder_path=str(tmp_path))
        assert [os.path.basename(p) for p in paths] == ["English.srt"]

    def test_an_empty_body_is_not_written(self, client, monkeypatch, tmp_path):
        def empty(url, timeout=None, headers=None):
            r = MagicMock(); r.status_code = 200; r.content = b""
            return r
        monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", empty)
        assert client.download_subtitles_batch(
            TRACKS, title="Film", folder_path=str(tmp_path)) == []

    def test_a_file_already_on_disk_is_not_refetched(self, client, downloads, tmp_path):
        client.download_subtitles_batch(TRACKS, title="Film", folder_path=str(tmp_path))
        assert len(downloads) == 2
        client.download_subtitles_batch(TRACKS, title="Film", folder_path=str(tmp_path))
        assert len(downloads) == 2, "refetched subtitles already on disk"

    def test_the_folder_is_created_if_missing(self, client, downloads, tmp_path):
        target = str(tmp_path / "deep" / "nested")
        paths = client.download_subtitles_batch(
            TRACKS, title="Film", folder_path=target)
        assert len(paths) == 2 and all(os.path.exists(p) for p in paths)

    def test_downloads_run_in_parallel(self, client, monkeypatch, tmp_path):
        """32 tracks is the normal case, and this happens while a player waits.
        Measured against the live endpoint: 2.9s at 8 workers, where the slowest
        single request was 0.95s -- i.e. sequential would be ~20s, past the
        point Kodi gives up on opening a stream."""
        import threading, time
        peak = {"n": 0, "cur": 0}
        lock = threading.Lock()

        def slow(url, timeout=None, headers=None):
            with lock:
                peak["cur"] += 1
                peak["n"] = max(peak["n"], peak["cur"])
            time.sleep(0.15)
            with lock:
                peak["cur"] -= 1
            r = MagicMock(); r.status_code = 200; r.content = b"x"
            return r

        monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", slow)
        many = [{"display": f"Lang{i}", "language": f"l{i}", "url": f"http://x/{i}"}
                for i in range(16)]
        client.download_subtitles_batch(many, title="Film", folder_path=str(tmp_path))
        assert peak["n"] > 1, "downloads ran one at a time"


# ------------------------------------------------- wired into SubtitleManager

from lib.clients.subtitle.submanager import SubtitleManager  # noqa: E402


def _manager(data=None):
    data = data or {"title": "No Other Choice", "mode": "movies",
                    "ids": {"imdb_id": "tt31967670", "tmdb_id": "639988"}}
    return SubtitleManager(data, MagicMock())


def _settings(**overrides):
    """get_setting stub. Wyzie on, everything else off, unless told otherwise."""
    values = {"wyzie_subtitle_enabled": True, "stremio_subtitle_enabled": False,
              "deepl_enabled": False, "auto_subtitle_download": False}
    values.update(overrides)
    return lambda key, *a, **k: values.get(key, False)


class TestManagerPrefersWyzie:
    def test_wyzie_is_asked_first_and_its_answer_is_used(self, tmp_path):
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock(return_value=TRACKS)
        m.wyzie_client.download_subtitles_batch = MagicMock(
            return_value=["/subs/a.srt"])
        m.opensub_client.get_subtitles = MagicMock(return_value=TRACKS)
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            got = m.fetch_subtitles(folder_path=str(tmp_path))
        assert got == ["/subs/a.srt"]
        m.opensub_client.get_subtitles.assert_not_called()

    def test_it_falls_back_to_opensubtitles_when_wyzie_has_nothing(self, tmp_path):
        """Not an either/or. Wyzie is keyless and usually richer, but a title it
        lacks is exactly when the other source is worth having."""
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        m.opensub_client.get_subtitles = MagicMock(return_value=TRACKS)
        m.opensub_client.download_subtitles_batch = MagicMock(
            return_value=["/subs/os.srt"])
        with patch("lib.clients.subtitle.submanager.get_setting",
                   _settings(stremio_subtitle_enabled=True)):
            got = m.fetch_subtitles(folder_path=str(tmp_path))
        assert got == ["/subs/os.srt"]

    def test_opensubtitles_is_not_called_when_it_is_switched_off(self, tmp_path):
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        m.opensub_client.get_subtitles = MagicMock()
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            assert m.fetch_subtitles(folder_path=str(tmp_path)) is None
        m.opensub_client.get_subtitles.assert_not_called()

    def test_wyzie_is_not_called_when_it_is_switched_off(self, tmp_path):
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock()
        m.opensub_client.get_subtitles = MagicMock(return_value=None)
        with patch("lib.clients.subtitle.submanager.get_setting",
                   _settings(wyzie_subtitle_enabled=False,
                             stremio_subtitle_enabled=True)):
            m.fetch_subtitles(folder_path=str(tmp_path))
        m.wyzie_client.get_subtitles.assert_not_called()

    def test_a_title_with_only_a_tmdb_id_still_gets_subtitles(self, tmp_path):
        """The fxl:/tmdb catalogs have no imdb id at all, and the old path gave
        up on those outright ('no_imdb'). Wyzie takes either."""
        m = _manager({"title": "No Other Choice", "mode": "movies",
                      "ids": {"tmdb_id": "639988"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=TRACKS)
        m.wyzie_client.download_subtitles_batch = MagicMock(
            return_value=["/subs/a.srt"])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            got = m.fetch_subtitles(folder_path=str(tmp_path))
        assert got == ["/subs/a.srt"]
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "639988"

    def test_an_imdb_id_is_preferred_when_both_are_present(self, tmp_path):
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "tt31967670"

    def test_no_id_of_either_kind_is_still_refused(self, tmp_path):
        m = _manager({"title": "Mystery", "mode": "movies", "ids": {}})
        m.wyzie_client.get_subtitles = MagicMock()
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            assert m.fetch_subtitles(folder_path=str(tmp_path)) is None
        assert m.last_fetch_status == "no_imdb"
        m.wyzie_client.get_subtitles.assert_not_called()

    def test_an_episode_passes_season_and_episode(self, tmp_path):
        m = _manager({"title": "X-Men 97", "mode": "tv",
                      "ids": {"imdb_id": "tt13623148"},
                      "tv_data": {"season": 1, "episode": 1}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        args = m.wyzie_client.get_subtitles.call_args[0]
        assert args[2] == 1 and args[3] == 1


def test_the_player_runs_subtitles_when_only_wyzie_is_enabled():
    """handle_subtitles gates the whole SubtitleManager path on
    stremio_subtitle_enabled. Wyzie has to open that gate too, or enabling it
    alone does nothing at all."""
    import re
    from pathlib import Path

    # Read the file rather than the class: JacktookPLayer inherits from
    # xbmc.Player, which the test harness stubs with a MagicMock, so the class
    # object carries no source. test_player.py reads it this way for the same
    # reason.
    source = (Path(__file__).resolve().parents[2] / "lib" / "player.py").read_text()
    body = re.search(r"def handle_subtitles\(self.*?\n    def ", source, re.S)
    assert body is not None, "handle_subtitles not found in player.py"
    assert "wyzie_subtitle_enabled" in body.group(0), \
        "handle_subtitles never checks the wyzie setting"


class TestIdentFromOriginalId:
    """The catalogs that most need this carry no tmdb_id or imdb_id at all.

    Measured on the Pi, playing No Other Choice from the flixlab catalog, the
    playback payload was:

        "ids": {"tmdb_id": "", "tvdb_id": "", "imdb_id": "",
                "original_id": "fxl:639988"}

    -- empty strings, not missing keys, with the only real id under
    `original_id` behind a catalog prefix. So the fetch bailed with "no_imdb"
    for exactly the titles the tmdb-or-imdb endpoint was supposed to rescue.
    """

    def test_an_fxl_original_id_yields_the_bare_tmdb_id(self, tmp_path):
        m = _manager({"title": "No Other Choice", "mode": "movies",
                      "ids": {"tmdb_id": "", "imdb_id": "",
                              "original_id": "fxl:639988"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "639988"

    def test_a_tmdb_prefixed_original_id_is_also_stripped(self, tmp_path):
        m = _manager({"title": "X", "mode": "movies",
                      "ids": {"original_id": "tmdb:1396"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "1396"

    def test_an_imdb_original_id_is_left_intact(self, tmp_path):
        """tt-ids are what the endpoint wants; stripping is only for prefixes."""
        m = _manager({"title": "X", "mode": "movies",
                      "ids": {"original_id": "tt0903747"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "tt0903747"

    def test_a_real_imdb_id_still_wins_over_original_id(self, tmp_path):
        m = _manager({"title": "X", "mode": "movies",
                      "ids": {"imdb_id": "tt31967670", "original_id": "fxl:639988"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "tt31967670"

    def test_empty_strings_are_treated_as_absent_not_as_ids(self, tmp_path):
        """The keys are PRESENT and empty, so `.get(...)` returns "" and any
        check that only tests for None would sail straight past it."""
        m = _manager({"title": "X", "mode": "movies",
                      "ids": {"tmdb_id": "", "imdb_id": "", "original_id": ""}})
        m.wyzie_client.get_subtitles = MagicMock()
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            assert m.fetch_subtitles(folder_path=str(tmp_path)) is None
        assert m.last_fetch_status == "no_imdb"
        m.wyzie_client.get_subtitles.assert_not_called()

    def test_a_prefixed_non_numeric_id_is_not_mangled(self, tmp_path):
        m = _manager({"title": "X", "mode": "movies",
                      "ids": {"original_id": "kitsu:anime-42"}})
        m.wyzie_client.get_subtitles = MagicMock(return_value=[])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
            m.fetch_subtitles(folder_path=str(tmp_path))
        assert m.wyzie_client.get_subtitles.call_args[0][1] == "kitsu:anime-42"


class TestCachedSubtitlesAreReusedSilently:
    """A populated cache must not turn every replay into a modal prompt.

    The "use existing subtitles?" dialog was nearly unreachable before: the
    OpenSubtitles path needed an imdb id and shipped disabled, so the cache was
    usually empty. Wyzie writes ~32 files on the FIRST play of every title, so
    without this every subsequent play of anything stops on a yes/no box.

    Reusing them is also the right answer on the merits: wyzie returns every
    language it has, so "download them again" cannot produce anything the cache
    does not already hold.
    """

    def _cache(self, tmp_path, n=3):
        for name in ("English", "French", "German")[:n]:
            (tmp_path / (name + ".srt")).write_text("1\n00:00:01,000 --> 00:00:02,000\nx\n")
        return sorted(str(p) for p in tmp_path.glob("*.srt"))

    def test_cached_files_are_returned_without_a_dialog(self, tmp_path):
        cached = self._cache(tmp_path)
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock()
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()), \
             patch("lib.clients.subtitle.submanager.xbmcgui") as gui:
            got = m.fetch_subtitles(folder_path=str(tmp_path))
        assert sorted(got) == cached
        gui.Dialog.assert_not_called()
        m.wyzie_client.get_subtitles.assert_not_called()

    def test_the_dialog_still_appears_when_wyzie_is_off(self, tmp_path):
        """Without wyzie the cache came from OpenSubtitles, where re-downloading
        genuinely can produce a different (better) subtitle -- so the choice
        still belongs to the user there."""
        self._cache(tmp_path)
        m = _manager()
        m.opensub_client.get_subtitles = MagicMock(return_value=None)
        with patch("lib.clients.subtitle.submanager.get_setting",
                   _settings(wyzie_subtitle_enabled=False,
                             stremio_subtitle_enabled=True)), \
             patch("lib.clients.subtitle.submanager.xbmcgui") as gui:
            gui.Dialog.return_value.yesno.return_value = False
            m.fetch_subtitles(folder_path=str(tmp_path))
        gui.Dialog.assert_called()

    def test_an_empty_cache_still_goes_to_wyzie(self, tmp_path):
        m = _manager()
        m.wyzie_client.get_subtitles = MagicMock(return_value=TRACKS)
        m.wyzie_client.download_subtitles_batch = MagicMock(return_value=["/a.srt"])
        with patch("lib.clients.subtitle.submanager.get_setting", _settings()), \
             patch("lib.clients.subtitle.submanager.xbmcgui") as gui:
            got = m.fetch_subtitles(folder_path=str(tmp_path))
        assert got == ["/a.srt"]
        gui.Dialog.assert_not_called()