| … | … | @@ -0,0 +1,338 @@ |
| 1 | """Subtitles from the source the player itself uses. |
| 2 | |
| 3 | The streams we play carry no subtitles. Measured 2026-09-04 on No Other Choice |
| 4 | (tmdb 639988): the HLS master lists four muxed A/V variants and no |
| 5 | #EXT-X-MEDIA at all. The vidcore player gets them from a separate endpoint -- |
| 6 | `vidcore.net/wyzie?id=<tmdb|imdb>[&season&episode]`, found in the 2026-07-21 |
| 7 | deobfuscation -- which returns `[{display, language, url, encoding}]`, needs no |
| 8 | key and no browser, and answers for imdb ids as readily as tmdb ones (verified: |
| 9 | id=1396 and id=tt0903747 return byte-identical bodies). That makes it usable for |
| 10 | every source Jacktook plays, not only vidcore. |
| 11 | |
| 12 | Run: python3 -m pytest plugin.video.jacktook/tests/unit/test_wyzie_subtitles.py -q |
| 13 | """ |
| 14 | |
| 15 | import os |
| 16 | from unittest.mock import MagicMock, patch |
| 17 | |
| 18 | import pytest |
| 19 | |
| 20 | from lib.clients.subtitle.wyzie import WyzieClient |
| 21 | |
| 22 | |
| 23 | @pytest.fixture |
| 24 | def client(): |
| 25 | return WyzieClient(MagicMock()) |
| 26 | |
| 27 | |
| 28 | def _resp(payload, status=200): |
| 29 | r = MagicMock() |
| 30 | r.status_code = status |
| 31 | r.json.return_value = payload |
| 32 | return r |
| 33 | |
| 34 | |
| 35 | TRACKS = [ |
| 36 | {"display": "English", "language": "en", "url": "http://x/en", "encoding": "UTF-8"}, |
| 37 | {"display": "French", "language": "fr", "url": "http://x/fr", "encoding": "UTF-8"}, |
| 38 | ] |
| 39 | |
| 40 | |
| 41 | # ------------------------------------------------------------ asking for them |
| 42 | |
| 43 | class TestQuery: |
| 44 | def test_a_movie_is_asked_for_by_id_alone(self, client): |
| 45 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 46 | return_value=_resp(TRACKS)) as g: |
| 47 | client.get_subtitles("movies", "639988") |
| 48 | url = g.call_args[0][0] |
| 49 | assert "id=639988" in url |
| 50 | assert "season" not in url and "episode" not in url |
| 51 | |
| 52 | def test_an_episode_carries_season_and_episode(self, client): |
| 53 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 54 | return_value=_resp(TRACKS)) as g: |
| 55 | client.get_subtitles("tv", "138502", season=1, episode=1) |
| 56 | url = g.call_args[0][0] |
| 57 | assert "id=138502" in url and "season=1" in url and "episode=1" in url |
| 58 | |
| 59 | def test_an_imdb_id_is_passed_through_unchanged(self, client): |
| 60 | """Verified against the live endpoint: tt0903747 and 1396 return the |
| 61 | same body, so no id translation is needed or wanted.""" |
| 62 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 63 | return_value=_resp(TRACKS)) as g: |
| 64 | client.get_subtitles("tv", "tt0903747", season=1, episode=1) |
| 65 | assert "id=tt0903747" in g.call_args[0][0] |
| 66 | |
| 67 | def test_a_referer_is_sent(self, client): |
| 68 | """The vidcore CDN 403s a request without one, and the /wyzie proxy is |
| 69 | on the same host.""" |
| 70 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 71 | return_value=_resp(TRACKS)) as g: |
| 72 | client.get_subtitles("movies", "639988") |
| 73 | headers = g.call_args.kwargs.get("headers") or {} |
| 74 | assert "vidcore" in headers.get("Referer", "") |
| 75 | |
| 76 | |
| 77 | class TestNoResults: |
| 78 | def test_a_404_is_no_subtitles_not_an_error(self, client): |
| 79 | """Verified: X-Men '97 S2E8 answers 404 with a zero-length body while |
| 80 | S1E1 returns 34 tracks. A title with none is ordinary.""" |
| 81 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 82 | return_value=_resp(None, status=404)): |
| 83 | assert client.get_subtitles("tv", "138502", season=2, episode=8) == [] |
| 84 | |
| 85 | def test_a_network_failure_is_no_subtitles(self, client): |
| 86 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 87 | side_effect=OSError("no route to host")): |
| 88 | assert client.get_subtitles("movies", "639988") == [] |
| 89 | |
| 90 | def test_a_non_list_body_is_no_subtitles(self, client): |
| 91 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 92 | return_value=_resp({"error": "nope"})): |
| 93 | assert client.get_subtitles("movies", "639988") == [] |
| 94 | |
| 95 | def test_entries_without_a_url_are_dropped(self, client): |
| 96 | body = [{"display": "English", "language": "en", "url": "http://x/en"}, |
| 97 | {"display": "Broken", "language": "xx"}] |
| 98 | with patch("lib.clients.subtitle.wyzie.requests.get", |
| 99 | return_value=_resp(body)): |
| 100 | got = client.get_subtitles("movies", "639988") |
| 101 | assert [t["language"] for t in got] == ["en"] |
| 102 | |
| 103 | |
| 104 | # ---------------------------------------------------------------- downloading |
| 105 | |
| 106 | @pytest.fixture |
| 107 | def downloads(monkeypatch): |
| 108 | calls = [] |
| 109 | |
| 110 | def fake(url, timeout=None, headers=None): |
| 111 | calls.append(url) |
| 112 | r = MagicMock() |
| 113 | r.status_code = 200 |
| 114 | r.content = b"1\n00:00:01,000 --> 00:00:02,000\nhello\n" |
| 115 | return r |
| 116 | |
| 117 | monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", fake) |
| 118 | return calls |
| 119 | |
| 120 | |
| 121 | class TestDownload: |
| 122 | def test_files_land_in_the_folder_with_readable_names(self, client, downloads, tmp_path): |
| 123 | paths = client.download_subtitles_batch( |
| 124 | TRACKS, title="No Other Choice", folder_path=str(tmp_path)) |
| 125 | names = sorted(os.path.basename(p) for p in paths) |
| 126 | assert names == ["No Other Choice.English.srt", "No Other Choice.French.srt"] |
| 127 | assert all(os.path.getsize(p) > 0 for p in paths) |
| 128 | |
| 129 | def test_the_display_name_is_used_because_kodi_labels_by_filename( |
| 130 | self, client, downloads, tmp_path): |
| 131 | """wyzie's own label is better than anything we could map a code to: |
| 132 | it distinguishes 'Portuguese (BR)' from 'Portuguese', which pb/pt would |
| 133 | only do if we happened to know that mapping. |
| 134 | |
| 135 | The brackets do not survive slugify_title -- the repo's existing |
| 136 | sanitiser, used here rather than a second one that keeps them, because |
| 137 | the distinction is what matters and 'Portuguese BR' carries it.""" |
| 138 | tracks = [{"display": "Portuguese (BR)", "language": "pb", "url": "http://x/pb"}, |
| 139 | {"display": "Portuguese", "language": "pt", "url": "http://x/pt"}] |
| 140 | paths = client.download_subtitles_batch( |
| 141 | tracks, title="Film", folder_path=str(tmp_path)) |
| 142 | names = sorted(os.path.basename(p) for p in paths) |
| 143 | assert names == ["Film.Portuguese BR.srt", "Film.Portuguese.srt"] |
| 144 | assert len(set(names)) == 2 |
| 145 | |
| 146 | def test_two_tracks_with_the_same_label_do_not_collide( |
| 147 | self, client, downloads, tmp_path): |
| 148 | tracks = [dict(TRACKS[0]), dict(TRACKS[0], url="http://x/en2")] |
| 149 | paths = client.download_subtitles_batch( |
| 150 | tracks, title="Film", folder_path=str(tmp_path)) |
| 151 | assert len(set(paths)) == 2 |
| 152 | assert all(os.path.exists(p) for p in paths) |
| 153 | |
| 154 | def test_a_label_cannot_escape_the_folder(self, client, downloads, tmp_path): |
| 155 | """`display` arrives inside a remote JSON body and reaches a filename.""" |
| 156 | tracks = [{"display": "../../../etc/passwd", "language": "en", |
| 157 | "url": "http://x/en"}] |
| 158 | paths = client.download_subtitles_batch( |
| 159 | tracks, title="Film", folder_path=str(tmp_path)) |
| 160 | for p in paths: |
| 161 | assert os.path.dirname(os.path.realpath(p)) == os.path.realpath(str(tmp_path)) |
| 162 | |
| 163 | def test_one_failure_does_not_lose_the_rest(self, client, monkeypatch, tmp_path): |
| 164 | def flaky(url, timeout=None, headers=None): |
| 165 | r = MagicMock() |
| 166 | if url.endswith("/fr"): |
| 167 | raise OSError("connection reset") |
| 168 | r.status_code = 200 |
| 169 | r.content = b"1\n00:00:01,000 --> 00:00:02,000\nhi\n" |
| 170 | return r |
| 171 | monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", flaky) |
| 172 | paths = client.download_subtitles_batch( |
| 173 | TRACKS, title="Film", folder_path=str(tmp_path)) |
| 174 | assert [os.path.basename(p) for p in paths] == ["Film.English.srt"] |
| 175 | |
| 176 | def test_an_empty_body_is_not_written(self, client, monkeypatch, tmp_path): |
| 177 | def empty(url, timeout=None, headers=None): |
| 178 | r = MagicMock(); r.status_code = 200; r.content = b"" |
| 179 | return r |
| 180 | monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", empty) |
| 181 | assert client.download_subtitles_batch( |
| 182 | TRACKS, title="Film", folder_path=str(tmp_path)) == [] |
| 183 | |
| 184 | def test_a_file_already_on_disk_is_not_refetched(self, client, downloads, tmp_path): |
| 185 | client.download_subtitles_batch(TRACKS, title="Film", folder_path=str(tmp_path)) |
| 186 | assert len(downloads) == 2 |
| 187 | client.download_subtitles_batch(TRACKS, title="Film", folder_path=str(tmp_path)) |
| 188 | assert len(downloads) == 2, "refetched subtitles already on disk" |
| 189 | |
| 190 | def test_the_folder_is_created_if_missing(self, client, downloads, tmp_path): |
| 191 | target = str(tmp_path / "deep" / "nested") |
| 192 | paths = client.download_subtitles_batch( |
| 193 | TRACKS, title="Film", folder_path=target) |
| 194 | assert len(paths) == 2 and all(os.path.exists(p) for p in paths) |
| 195 | |
| 196 | def test_downloads_run_in_parallel(self, client, monkeypatch, tmp_path): |
| 197 | """32 tracks is the normal case, and this happens while a player waits. |
| 198 | Measured against the live endpoint: 2.9s at 8 workers, where the slowest |
| 199 | single request was 0.95s -- i.e. sequential would be ~20s, past the |
| 200 | point Kodi gives up on opening a stream.""" |
| 201 | import threading, time |
| 202 | peak = {"n": 0, "cur": 0} |
| 203 | lock = threading.Lock() |
| 204 | |
| 205 | def slow(url, timeout=None, headers=None): |
| 206 | with lock: |
| 207 | peak["cur"] += 1 |
| 208 | peak["n"] = max(peak["n"], peak["cur"]) |
| 209 | time.sleep(0.15) |
| 210 | with lock: |
| 211 | peak["cur"] -= 1 |
| 212 | r = MagicMock(); r.status_code = 200; r.content = b"x" |
| 213 | return r |
| 214 | |
| 215 | monkeypatch.setattr("lib.clients.subtitle.wyzie.requests.get", slow) |
| 216 | many = [{"display": f"Lang{i}", "language": f"l{i}", "url": f"http://x/{i}"} |
| 217 | for i in range(16)] |
| 218 | client.download_subtitles_batch(many, title="Film", folder_path=str(tmp_path)) |
| 219 | assert peak["n"] > 1, "downloads ran one at a time" |
| 220 | |
| 221 | |
| 222 | # ------------------------------------------------- wired into SubtitleManager |
| 223 | |
| 224 | from lib.clients.subtitle.submanager import SubtitleManager # noqa: E402 |
| 225 | |
| 226 | |
| 227 | def _manager(data=None): |
| 228 | data = data or {"title": "No Other Choice", "mode": "movies", |
| 229 | "ids": {"imdb_id": "tt31967670", "tmdb_id": "639988"}} |
| 230 | return SubtitleManager(data, MagicMock()) |
| 231 | |
| 232 | |
| 233 | def _settings(**overrides): |
| 234 | """get_setting stub. Wyzie on, everything else off, unless told otherwise.""" |
| 235 | values = {"wyzie_subtitle_enabled": True, "stremio_subtitle_enabled": False, |
| 236 | "deepl_enabled": False, "auto_subtitle_download": False} |
| 237 | values.update(overrides) |
| 238 | return lambda key, *a, **k: values.get(key, False) |
| 239 | |
| 240 | |
| 241 | class TestManagerPrefersWyzie: |
| 242 | def test_wyzie_is_asked_first_and_its_answer_is_used(self, tmp_path): |
| 243 | m = _manager() |
| 244 | m.wyzie_client.get_subtitles = MagicMock(return_value=TRACKS) |
| 245 | m.wyzie_client.download_subtitles_batch = MagicMock( |
| 246 | return_value=["/subs/a.srt"]) |
| 247 | m.opensub_client.get_subtitles = MagicMock(return_value=TRACKS) |
| 248 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 249 | got = m.fetch_subtitles(folder_path=str(tmp_path)) |
| 250 | assert got == ["/subs/a.srt"] |
| 251 | m.opensub_client.get_subtitles.assert_not_called() |
| 252 | |
| 253 | def test_it_falls_back_to_opensubtitles_when_wyzie_has_nothing(self, tmp_path): |
| 254 | """Not an either/or. Wyzie is keyless and usually richer, but a title it |
| 255 | lacks is exactly when the other source is worth having.""" |
| 256 | m = _manager() |
| 257 | m.wyzie_client.get_subtitles = MagicMock(return_value=[]) |
| 258 | m.opensub_client.get_subtitles = MagicMock(return_value=TRACKS) |
| 259 | m.opensub_client.download_subtitles_batch = MagicMock( |
| 260 | return_value=["/subs/os.srt"]) |
| 261 | with patch("lib.clients.subtitle.submanager.get_setting", |
| 262 | _settings(stremio_subtitle_enabled=True)): |
| 263 | got = m.fetch_subtitles(folder_path=str(tmp_path)) |
| 264 | assert got == ["/subs/os.srt"] |
| 265 | |
| 266 | def test_opensubtitles_is_not_called_when_it_is_switched_off(self, tmp_path): |
| 267 | m = _manager() |
| 268 | m.wyzie_client.get_subtitles = MagicMock(return_value=[]) |
| 269 | m.opensub_client.get_subtitles = MagicMock() |
| 270 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 271 | assert m.fetch_subtitles(folder_path=str(tmp_path)) is None |
| 272 | m.opensub_client.get_subtitles.assert_not_called() |
| 273 | |
| 274 | def test_wyzie_is_not_called_when_it_is_switched_off(self, tmp_path): |
| 275 | m = _manager() |
| 276 | m.wyzie_client.get_subtitles = MagicMock() |
| 277 | m.opensub_client.get_subtitles = MagicMock(return_value=None) |
| 278 | with patch("lib.clients.subtitle.submanager.get_setting", |
| 279 | _settings(wyzie_subtitle_enabled=False, |
| 280 | stremio_subtitle_enabled=True)): |
| 281 | m.fetch_subtitles(folder_path=str(tmp_path)) |
| 282 | m.wyzie_client.get_subtitles.assert_not_called() |
| 283 | |
| 284 | def test_a_title_with_only_a_tmdb_id_still_gets_subtitles(self, tmp_path): |
| 285 | """The fxl:/tmdb catalogs have no imdb id at all, and the old path gave |
| 286 | up on those outright ('no_imdb'). Wyzie takes either.""" |
| 287 | m = _manager({"title": "No Other Choice", "mode": "movies", |
| 288 | "ids": {"tmdb_id": "639988"}}) |
| 289 | m.wyzie_client.get_subtitles = MagicMock(return_value=TRACKS) |
| 290 | m.wyzie_client.download_subtitles_batch = MagicMock( |
| 291 | return_value=["/subs/a.srt"]) |
| 292 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 293 | got = m.fetch_subtitles(folder_path=str(tmp_path)) |
| 294 | assert got == ["/subs/a.srt"] |
| 295 | assert m.wyzie_client.get_subtitles.call_args[0][1] == "639988" |
| 296 | |
| 297 | def test_an_imdb_id_is_preferred_when_both_are_present(self, tmp_path): |
| 298 | m = _manager() |
| 299 | m.wyzie_client.get_subtitles = MagicMock(return_value=[]) |
| 300 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 301 | m.fetch_subtitles(folder_path=str(tmp_path)) |
| 302 | assert m.wyzie_client.get_subtitles.call_args[0][1] == "tt31967670" |
| 303 | |
| 304 | def test_no_id_of_either_kind_is_still_refused(self, tmp_path): |
| 305 | m = _manager({"title": "Mystery", "mode": "movies", "ids": {}}) |
| 306 | m.wyzie_client.get_subtitles = MagicMock() |
| 307 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 308 | assert m.fetch_subtitles(folder_path=str(tmp_path)) is None |
| 309 | assert m.last_fetch_status == "no_imdb" |
| 310 | m.wyzie_client.get_subtitles.assert_not_called() |
| 311 | |
| 312 | def test_an_episode_passes_season_and_episode(self, tmp_path): |
| 313 | m = _manager({"title": "X-Men 97", "mode": "tv", |
| 314 | "ids": {"imdb_id": "tt13623148"}, |
| 315 | "tv_data": {"season": 1, "episode": 1}}) |
| 316 | m.wyzie_client.get_subtitles = MagicMock(return_value=[]) |
| 317 | with patch("lib.clients.subtitle.submanager.get_setting", _settings()): |
| 318 | m.fetch_subtitles(folder_path=str(tmp_path)) |
| 319 | args = m.wyzie_client.get_subtitles.call_args[0] |
| 320 | assert args[2] == 1 and args[3] == 1 |
| 321 | |
| 322 | |
| 323 | def test_the_player_runs_subtitles_when_only_wyzie_is_enabled(): |
| 324 | """handle_subtitles gates the whole SubtitleManager path on |
| 325 | stremio_subtitle_enabled. Wyzie has to open that gate too, or enabling it |
| 326 | alone does nothing at all.""" |
| 327 | import re |
| 328 | from pathlib import Path |
| 329 | |
| 330 | # Read the file rather than the class: JacktookPLayer inherits from |
| 331 | # xbmc.Player, which the test harness stubs with a MagicMock, so the class |
| 332 | # object carries no source. test_player.py reads it this way for the same |
| 333 | # reason. |
| 334 | source = (Path(__file__).resolve().parents[2] / "lib" / "player.py").read_text() |
| 335 | body = re.search(r"def handle_subtitles\(self.*?\n def ", source, re.S) |
| 336 | assert body is not None, "handle_subtitles not found in player.py" |
| 337 | assert "wyzie_subtitle_enabled" in body.group(0), \ |
| 338 | "handle_subtitles never checks the wyzie setting" |