"""A library copy plays with ITS OWN subtitles, not wyzie's.
2026-10-18: No Other Choice's English drifted in Kodi although vsembed had
synced it to the copy's audio on 10-04. The synced file sat beside the mp4,
where only Jellyfin looks; Jacktook fetched wyzie itself into a per-TMDB cache
and reused it silently. Proven by md5: the Pi's English.srt (025fc304...) was
byte-identical to the UNSYNCED original, not the corrected sidecar (37840b79...).
A per-TMDB cache cannot be right anyway -- two copies of one film are two
different cuts. So when the stream is vsembed's `/mp4/<kind>/<ident>.mp4`, the
subtitles come from `/subs/<kind>/<ident>.json` on the same host: the copy's
own sidecars, after the server measured them against its audio. Anything else
-- a live stream, the server unreachable, a copy with none -- falls through to
wyzie exactly as before.
Run: python3 -m pytest plugin.video.jacktook/tests/unit/test_library_subtitles.py -q
"""
import os
from unittest.mock import MagicMock, patch
import pytest
from lib.clients.subtitle.library import LibrarySubtitleClient, library_list_url
HOST = "http://192.168.50.170:7220"
MOVIE = HOST + "/mp4/movie/tmdb:639988.mp4?h=1080"
EPISODE = HOST + "/mp4/series/tmdb:138502:1:1.mp4?h=1080"
def _resp(payload=None, status=200, body=b""):
r = MagicMock()
r.status_code = status
r.json.return_value = payload
r.content = body
return r
def _listing(*entries):
return {"subtitles": [
{"id": sid, "lang": sid.split(".")[0],
"url": "{0}/subs/movie/tmdb:639988/{1}.srt".format(HOST, sid),
"identity": identity}
for sid, identity in entries]}
class FakeServer:
"""Answers the list URL with `listing`, each file URL with its id."""
def __init__(self, listing):
self.listing = listing
self.files = []
def __call__(self, url, timeout=None, headers=None):
if url.endswith(".json"):
return _resp(self.listing)
self.files.append(url)
sid = url.rsplit("/", 1)[1][:-len(".srt")]
return _resp(body="1\n00:00:01,000 --> 00:00:02,000\n{0}\n".format(sid).encode())
# ------------------------------------------------------------ which streams
class TestListUrl:
def test_a_library_movie(self):
assert library_list_url(MOVIE) == HOST + "/subs/movie/tmdb:639988.json"
def test_a_library_episode(self):
assert library_list_url(EPISODE) == HOST + "/subs/series/tmdb:138502:1:1.json"
def test_an_imdb_keyed_copy(self):
assert library_list_url(HOST + "/mp4/movie/tt31967670.mp4") == \
HOST + "/subs/movie/tt31967670.json"
@pytest.mark.parametrize("url", [
HOST + "/vidcore/movie/tmdb:639988.m3u8",
HOST + "/hdhub/movie/tmdb:639988.mkv",
"https://cdn.example/mp4/movie/x/y.mp4",
"magnet:?xt=urn:btih:abc",
"",
None,
])
def test_anything_else_is_not_a_library_copy(self, url):
assert library_list_url(url) is None
# ----------------------------------------------------------------- fetching
@pytest.fixture
def client(tmp_path):
return LibrarySubtitleClient(str(tmp_path))
def _get(server):
return patch("lib.clients.subtitle.library.requests.get", side_effect=server)
class TestFetch:
def test_each_listed_track_is_written_and_returned(self, client):
server = FakeServer(_listing(("en", [99078, 1]), ("ko", [5, 1])))
with _get(server):
paths = client.get_subtitles(MOVIE)
assert sorted(os.path.basename(p) for p in paths) == ["en.srt", "ko.srt"]
for p in paths:
sid = os.path.basename(p)[:-len(".srt")]
with open(p, "rb") as f:
assert sid.encode() in f.read()
def test_a_second_track_in_one_language_is_kept_apart(self, client):
server = FakeServer(_listing(("en", [1, 1]), ("en.2", [2, 1])))
with _get(server):
paths = client.get_subtitles(MOVIE)
assert sorted(os.path.basename(p) for p in paths) == ["en.2.srt", "en.srt"]
def test_an_unchanged_track_is_not_downloaded_again(self, client):
server = FakeServer(_listing(("en", [99078, 1])))
with _get(server):
client.get_subtitles(MOVIE)
client.get_subtitles(MOVIE)
assert len(server.files) == 1
def test_a_track_the_server_resynced_is_downloaded_again(self, client):
"""The whole point: the server corrects a file in place, and the copy
on the Pi must follow it rather than outlive it."""
server = FakeServer(_listing(("en", [99078, 1])))
with _get(server):
client.get_subtitles(MOVIE)
server.listing = _listing(("en", [99100, 2]))
client.get_subtitles(MOVIE)
assert len(server.files) == 2
def test_a_track_the_server_no_longer_lists_is_not_offered(self, client):
server = FakeServer(_listing(("en", [1, 1]), ("de", [2, 1])))
with _get(server):
client.get_subtitles(MOVIE)
server.listing = _listing(("en", [1, 1]))
paths = client.get_subtitles(MOVIE)
assert [os.path.basename(p) for p in paths] == ["en.srt"]
def test_two_copies_do_not_share_a_folder(self, client):
server = FakeServer(_listing(("en", [1, 1])))
with _get(server):
a = client.get_subtitles(MOVIE)
b = client.get_subtitles(EPISODE)
assert os.path.dirname(a[0]) != os.path.dirname(b[0])
def test_one_failed_file_costs_only_that_track(self, client):
server = FakeServer(_listing(("en", [1, 1]), ("ko", [2, 1])))
def flaky(url, timeout=None, headers=None):
if url.endswith("/ko.srt"):
return _resp(status=503)
return server(url, timeout, headers)
with _get(flaky):
paths = client.get_subtitles(MOVIE)
assert [os.path.basename(p) for p in paths] == ["en.srt"]
def test_an_id_cannot_escape_the_folder(self, client, tmp_path):
server = FakeServer({"subtitles": [
{"id": "../../evil", "lang": "en", "url": HOST + "/subs/x/evil.srt",
"identity": [1, 1]}]})
with _get(server):
paths = client.get_subtitles(MOVIE)
for p in paths:
assert os.path.realpath(p).startswith(os.path.realpath(str(tmp_path)))
class TestFallsThrough:
"""None means 'not mine to answer' -- the caller carries on to wyzie."""
def test_not_a_library_stream(self, client):
with patch("lib.clients.subtitle.library.requests.get") as g:
assert client.get_subtitles(HOST + "/vidcore/movie/tmdb:1.m3u8") is None
g.assert_not_called()
def test_the_server_is_unreachable(self, client):
with patch("lib.clients.subtitle.library.requests.get",
side_effect=OSError("no route")):
assert client.get_subtitles(MOVIE) is None
def test_an_older_server_without_the_route(self, client):
with patch("lib.clients.subtitle.library.requests.get",
return_value=_resp(status=404)):
assert client.get_subtitles(MOVIE) is None
def test_a_copy_with_no_subtitles(self, client):
with _get(FakeServer({"subtitles": []})):
assert client.get_subtitles(MOVIE) is None
# ------------------------------------------------- wired into SubtitleManager
from lib.clients.subtitle.submanager import SubtitleManager # noqa: E402
def _settings(**overrides):
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)
def _manager(url):
return SubtitleManager({"title": "in library — seekable\nlocal copy",
"mode": "movies", "url": url,
"ids": {"tmdb_id": "", "imdb_id": "",
"original_id": "fxl:639988"}}, MagicMock())
class TestManager:
def test_a_library_copy_uses_its_own_and_never_asks_wyzie(self, tmp_path):
m = _manager(MOVIE)
m.library_client.get_subtitles = MagicMock(return_value=["/lib/en.srt"])
m.wyzie_client.get_subtitles = MagicMock()
with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
got = m.fetch_subtitles(folder_path=str(tmp_path))
assert got == ["/lib/en.srt"]
m.wyzie_client.get_subtitles.assert_not_called()
def test_it_wins_over_a_stale_per_title_cache(self, tmp_path):
"""The Pi's cache held the unsynced English for this exact title."""
(tmp_path / "English.srt").write_text("stale")
m = _manager(MOVIE)
m.library_client.get_subtitles = MagicMock(return_value=["/lib/en.srt"])
with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
assert m.fetch_subtitles(folder_path=str(tmp_path)) == ["/lib/en.srt"]
def test_nothing_from_the_library_falls_through_to_wyzie(self, tmp_path):
m = _manager(HOST + "/vidcore/movie/tmdb:639988.m3u8")
m.library_client.get_subtitles = MagicMock(return_value=None)
m.wyzie_client.get_subtitles = MagicMock(return_value=[{"url": "x"}])
m.wyzie_client.download_subtitles_batch = MagicMock(return_value=["/w/a.srt"])
with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
assert m.fetch_subtitles(folder_path=str(tmp_path)) == ["/w/a.srt"]
# ------------------------------------------- live streams (vsembed 0.33.0)
# A stream with no library copy behind it: the server samples the stream's
# own audio and places each English track (~30-60 s). The player starts on
# wyzie's raw tracks and swaps the synced one in when it is ready.
from lib.clients.subtitle.library import live_list_url, watch_live # noqa: E402
class TestLiveListUrl:
def test_a_vidcore_movie(self):
assert live_list_url(HOST + "/vidcore/movie/tmdb:1058424.m3u8?h=1080") == \
HOST + "/subs/movie/tmdb:1058424.json?src=vidcore&h=1080"
def test_an_hdhub_episode(self):
assert live_list_url(HOST + "/hdhub/series/tmdb:213375:1:1.mkv?h=720") == \
HOST + "/subs/series/tmdb:213375:1:1.json?src=hdhub&h=720"
def test_a_library_row_without_a_copy_yet(self):
assert live_list_url(MOVIE) == HOST + "/subs/movie/tmdb:639988.json?src=mp4&h=1080"
def test_no_height_asks_for_the_default(self):
assert live_list_url(HOST + "/vidcore/movie/tt27369017.m3u8") == \
HOST + "/subs/movie/tt27369017.json?src=vidcore&h=1080"
@pytest.mark.parametrize("url", [
"https://cdn.example/vidcore/movie/x.m3u8",
HOST + "/stream/movie/tt1.json",
"magnet:?xt=urn:btih:abc", "", None,
])
def test_anything_else_is_not_ours(self, url):
assert live_list_url(url) is None
LIVE = HOST + "/vidcore/movie/tmdb:1058424.m3u8?h=1080"
class LiveServer(FakeServer):
def __init__(self, statuses, listing):
FakeServer.__init__(self, listing)
self.statuses = list(statuses)
self.asked = 0
def __call__(self, url, timeout=None, headers=None):
if url.endswith(".json") or ".json?" in url:
self.asked += 1
status = self.statuses.pop(0) if len(self.statuses) > 1 else self.statuses[0]
body = dict(self.listing) if status == "ready" else {"subtitles": []}
body["status"] = status
return _resp(body)
return FakeServer.__call__(self, url, timeout, headers)
class TestPollLive:
def test_pending(self, client):
with _get(LiveServer(["pending"], _listing())):
assert client.poll_live(LIVE) == ("pending", None)
def test_ready_downloads_the_synced_tracks(self, client):
with _get(LiveServer(["ready"], _listing(("en", [1, 1])))):
status, paths = client.poll_live(LIVE)
assert status == "ready"
assert [os.path.basename(p) for p in paths] == ["en.synced.srt"]
def test_none(self, client):
with _get(LiveServer(["none"], _listing())):
assert client.poll_live(LIVE) == ("none", None)
def test_an_unreachable_server_is_none(self, client):
with patch("lib.clients.subtitle.library.requests.get", side_effect=OSError("x")):
assert client.poll_live(LIVE) == ("none", None)
def test_not_a_vsembed_stream_is_none_without_asking(self, client):
with patch("lib.clients.subtitle.library.requests.get") as g:
assert client.poll_live("https://cdn.example/x.m3u8") == ("none", None)
g.assert_not_called()
class TestWatchLive:
def test_waits_through_pending_then_returns_the_tracks(self, client):
server = LiveServer(["pending", "pending", "ready"], _listing(("en", [1, 1])))
waits = []
with _get(server):
paths = watch_live(client, LIVE, still_playing=lambda: True,
wait=lambda s: waits.append(s) or True)
assert [os.path.basename(p) for p in paths] == ["en.synced.srt"]
assert server.asked == 3 and len(waits) == 2
def test_gives_up_when_the_server_has_no_answer(self, client):
with _get(LiveServer(["pending", "none"], _listing())):
assert watch_live(client, LIVE, lambda: True, lambda s: True) is None
def test_stops_when_playback_stops(self, client):
server = LiveServer(["pending"], _listing())
with _get(server):
assert watch_live(client, LIVE, lambda: False, lambda s: True) is None
assert server.asked == 1
def test_stops_when_kodi_is_closing(self, client):
with _get(LiveServer(["pending"], _listing())):
assert watch_live(client, LIVE, lambda: True, lambda s: False) is None
def test_gives_up_at_the_deadline(self, client):
server = LiveServer(["pending"], _listing())
with _get(server):
assert watch_live(client, LIVE, lambda: True, lambda s: True,
deadline=30, interval=10) is None
assert server.asked == 4
class TestManagerSaysWhereSubtitlesCameFrom:
def test_from_the_library(self, tmp_path):
m = _manager(MOVIE)
m.library_client.get_subtitles = MagicMock(return_value=["/lib/en.srt"])
with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
m.fetch_subtitles(folder_path=str(tmp_path))
assert m.from_library is True
def test_not_from_the_library(self, tmp_path):
m = _manager(LIVE)
m.library_client.get_subtitles = MagicMock(return_value=None)
m.wyzie_client.get_subtitles = MagicMock(return_value=[{"url": "x"}])
m.wyzie_client.download_subtitles_batch = MagicMock(return_value=["/w/a.srt"])
with patch("lib.clients.subtitle.submanager.get_setting", _settings()):
m.fetch_subtitles(folder_path=str(tmp_path))
assert m.from_library is False