""" Showing a subtitle track in the player. The node extracts one track whole, converts it to WebVTT and caches it under its own hash; this side asks for it by ordinal, pulls the blob through the ordinary chunk path and hangs a `` on the video element. Four things have to hold here: **The selector exists only where the node said there was something to show.** It is drawn from `subtitle_tracks` in `stream_init` and from nothing else — there is no version check in the player — so a node too old to enumerate them draws no selector and is never sent a `subtitle_req` it would answer "unknown message type" to. The list also carries only the tracks the node can convert, so what the menu offers is what will actually appear. **The ordinal travels untouched.** It counts every subtitle stream in the container, including the bitmap ones that are never listed, so it is not the track's position in the list this client received. Renumbering it here would map a different stream on the node and show the wrong language, or nothing. **A seek does not take the subtitles down.** The extraction is whole-file, so the cues are absolute: the `` outlives every restart of the MediaSource, including the one an audio-language change produces. **The reply is matched to the request by file *and* track.** Two extractions for one film can be in flight when the viewer changes their mind, and the reply that arrives first is not necessarily the one asked for first. """ import json import re import shutil import subprocess from pathlib import Path import pytest STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" APP = STATIC / "video-player.js" TRANSPORT = STATIC / "transport.js" pytestmark = pytest.mark.skipif( shutil.which("node") is None or not APP.exists(), reason="node or the SPA sources are not available") @pytest.fixture(scope="module") def app(): return APP.read_text() @pytest.fixture(scope="module") def transport(): return TRANSPORT.read_text() def _lift(src: str, name: str) -> str: """One top-level function, as text, for node to execute.""" start = src.index(f"function {name}(") depth, i, seen = 0, start, False while i < len(src): if src[i] == "{": depth += 1 seen = True elif src[i] == "}": depth -= 1 if seen and depth == 0: return src[start:i + 1] i += 1 raise AssertionError(f"{name} never closes") def _player(app: str) -> str: """The whole component, by position in the file. Not the brace matcher below: `function VideoPlayer({ entry, ... })` destructures its props, so counting from the first `{` closes on the parameter list and returns the signature alone — a lift that finds nothing it was asked about and asserts happily against an empty string. """ i = app.index("function VideoPlayer(") nxt = app.find("\nfunction ", i + 1) return app[i:nxt if nxt > 0 else len(app)] def _block(src: str, opener: str) -> str: """The body of one `x = (args) => {` assignment, braces matched.""" start = src.index(opener) depth, i, seen = 0, start, False while i < len(src): if src[i] == "{": depth += 1 seen = True elif src[i] == "}": depth -= 1 if seen and depth == 0: return src[start:i + 1] i += 1 raise AssertionError(f"{opener!r} never closes") # ── The label, run rather than read ─────────────────────────────────────────── def _label_cases(tmp_path, app, cases, locale="en"): script = tmp_path / "sublabel.mjs" src = "\n".join([ app[app.index("const _ISO639 = {"):app.index("};", app.index("const _ISO639 = {")) + 2], _lift(app, "_languageName"), _lift(app, "subtitleTrackLabel"), ]) script.write_text( f"const getLocale = () => '{locale}';\n" # Not every key is interpolated — the forced/SDH qualifiers take no # parameters — and a stub that assumes one throws where the real `t` # returns a string. A fixture narrower than production tests itself. "const t = (k, p) => (p ? `${k}:${p.n}` : k);\n" + src + "\nconst out = JSON.parse(process.argv[2]).map(subtitleTrackLabel);\n" "console.log(JSON.stringify(out));\n") proc = subprocess.run( ["node", str(script), json.dumps(cases)], capture_output=True, text=True) assert proc.returncode == 0, proc.stderr return json.loads(proc.stdout) def test_the_language_is_named_not_shown_as_a_tag(tmp_path, app): """Same fold as the audio tracks: ffprobe reports either ISO 639-2 variant and a library uses both.""" out = _label_cases(tmp_path, app, [ {"i": 0, "lang": "fre"}, {"i": 1, "lang": "fra"}, {"i": 2, "lang": "eng"}, ]) assert out[0] == out[1], f"fre and fra must agree: {out}" assert "French" in out[0] and "English" in out[2], out def test_the_container_title_tells_a_forced_track_from_a_full_one(tmp_path, app): """The difference the title carries is not cosmetic here. "Forced" and "SDH" are tagged with the same language as the ordinary track they sit beside. Dropping the title makes them one menu entry repeated, and picking the wrong one is the difference between a full translation and three lines of signage in a whole film. """ out = _label_cases(tmp_path, app, [ {"i": 0, "lang": "eng", "title": "Forced"}, {"i": 1, "lang": "eng", "title": "SDH"}, {"i": 2, "lang": "eng"}, ]) assert len({*out}) == 3, f"three tracks must give three entries: {out}" assert "Forced" in out[0] and "SDH" in out[1] def test_a_forced_track_is_named_as_one_in_the_reader_s_language(tmp_path, app): """The entry that was reported as a broken feature. A forced track shows signage and foreign dialogue only — on a real film, 77 seconds of text across 2h32 — so picking it and seeing nothing is its normal behaviour. It has to be possible to tell it from the full track beside it, which carries the same language tag, and the disposition says so where the container's English title tag is often simply absent. """ out = _label_cases(tmp_path, app, [ {"i": 0, "lang": "fre", "forced": True}, {"i": 1, "lang": "fre"}, {"i": 2, "lang": "eng", "sdh": True}, ]) assert out[0] != out[1], f"a forced track must not read like the full one: {out}" # The catalogues are stubbed here, so what is asserted is that the label # goes through `t` at all: the qualifier has to be translated, not the # English word a muxer typed into the container. assert out[0] == 'French — video.subtitles_forced', out assert out[1] == 'French', out assert out[2] == 'English — video.subtitles_sdh', out def test_the_disposition_wins_over_the_container_title(tmp_path, app): """A muxer's "Forced" is one spelling of many, and in one language.""" out = _label_cases(tmp_path, app, [ {"i": 0, "lang": "fre", "title": "FORCE VF", "forced": True}, ]) assert 'FORCE VF' not in out[0], out assert out[0] == 'French — video.subtitles_forced', out def test_an_untagged_track_is_numbered_not_called_unknown(tmp_path, app): """A file with no language tags still needs distinguishable entries.""" out = _label_cases(tmp_path, app, [{"i": 3, "lang": None}, {"i": 4, "lang": "und"}]) assert out == ["video.subtitle_track_n:4", "video.subtitle_track_n:5"], out # ── The shape of the feature, read from the source ──────────────────────────── def test_the_selector_comes_from_the_node_list_and_not_from_a_version(app): """Discovery from the answer. There is no version check in the player, and an empty list is the whole of "this node has no subtitles for you".""" assert "msg.subtitle_tracks" in app assert "subtitleTracks.length > 0 &&" in app, ( "the selector must be drawn from the node's list") for forbidden in ("MNP_VERSION", "v >= '3.3'", "'3.3'"): assert forbidden not in app, ( f"the player decided on {forbidden!r} instead of on the answer") def test_the_ordinal_is_passed_through_rather_than_renumbered(app): """`track.i` is the node's ordinal; the list index is not it.""" assert "requestSubtitle(entry.id, track.i)" in app, ( "the request must carry the ordinal the node published") assert not re.search(r"requestSubtitle\([^)]*\bindex\b", app), ( "a list position was sent where a stream ordinal belongs") def test_a_seek_does_not_take_the_subtitles_down(app): """The payoff of extracting whole-file rather than per-seek. `stream_init` arrives again on every seek and on every audio-language change. It may restate the available tracks; it must not clear the chosen one or revoke the blob, or every seek would re-extract and the track would blink out mid-film. """ body = _block(app, "transport.onStreamInit = (msg) => {") assert "setSubtitleTracks(" in body for forbidden in ("setSubtitleUrl(", "setSubtitleTrack(", "subtitleUrlRef.current ="): assert forbidden not in body, ( f"stream_init touches {forbidden!r}, so a seek disturbs the track") def test_the_blob_is_pulled_through_the_ordinary_chunk_path(app): """Not a new transfer mechanism — the same indirection as a poster.""" assert "pipelinedDownload(" in app assert "info.hash" in app, "the cache hash from the reply is what is fetched" def test_a_stale_reply_cannot_install_its_track(app): """Two extractions in flight, and the last answered is not the last asked. Without the generation the older reply overwrites the newer choice, and the menu then shows a tick against a language that is not on screen. """ body = _player(app) assert "subtitleGenRef.current !== gen" in body, ( "nothing stops an out-of-date extraction from installing its blob") assert "URL.revokeObjectURL(url)" in body, ( "the superseded blob must be released, not merely ignored") def test_the_reply_is_matched_on_the_file_and_the_track(transport): """One film's two tracks are exactly the pair that can be in flight together, so the file id alone cannot route the reply.""" assert "`subtitle:${obj.file_id}:${obj.track}`" in transport assert "`subtitle:${msg.file_id}:${msg.track}`" in transport def test_a_failed_extraction_does_not_take_the_film_down(app): """Subtitles are an addition to a film that is already playing. `setError` is the player's fatal path — it replaces the picture. A text track that could not be read must not reach it. """ # Matched on braces rather than on the dependency list. The deps are the # part of a callback most likely to change for reasons unrelated to what is # asserted here, and a slice keyed to them stops at `.index()` raising # rather than at the property being broken — a guard that reports the wrong # thing is barely better than one that reports nothing. body = _block(_player(app), "const selectSubtitle = useCallback(") assert "setError(" not in body, ( "a subtitle failure takes the whole player down") assert "setSubtitleError(true)" in body