1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
"""
Choosing the audio track from the player.
A dubbed film carries several audio tracks and the node used to map the first
one unconditionally, so it played in whichever language was muxed first. Three
things have to hold on this side:
**The selector exists only where the node said there was something to choose.**
It is drawn from the `audio_tracks` list 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 an `audio_track` it would ignore. Ignoring
it would not degrade the stream; it would serve a different language in
silence, which is the failure this shape exists to make impossible.
**Changing track is the seek path.** One ffmpeg produces one audio track, so
there is nothing to switch inside a running stream: the node has to be asked
again, and the new stream opens with an init segment the SourceBuffer can only
accept after `reinitAt`'s abort/remove.
**The label is derived, not translated.** ffprobe reports ISO 639-2 in either
of its two variants, and the ten catalogues have no room for a language list —
`Intl.DisplayNames` does that, from a fold that is real logic and is run here
rather than read.
"""
import json
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()
def _player(app: str) -> str:
i = app.index("function VideoPlayer(")
nxt = app.find("\nfunction ", i + 1)
return app[i:nxt if nxt > 0 else len(app)]
def _lift(app: str, name: str) -> str:
"""One top-level function, as text, for node to execute."""
start = app.index(f"function {name}(")
depth, i, seen = 0, start, False
while i < len(app):
if app[i] == "{":
depth += 1
seen = True
elif app[i] == "}":
depth -= 1
if seen and depth == 0:
return app[start:i + 1]
i += 1
raise AssertionError(f"{name} never closes")
# ── The label, run rather than read ───────────────────────────────────────────
def _label_cases(tmp_path, app, cases, locale="en"):
script = tmp_path / "label.mjs"
src = "\n".join([
app[app.index("const _ISO639 = {"):app.index("};", app.index("const _ISO639 = {")) + 2],
# The language name is shared with `subtitleTrackLabel`, so it lives in
# a function of its own and has to be lifted alongside its caller. A
# lift that names one function stops exercising anything the moment
# logic moves out of it — here it throws, which is the good case; the
# bad one is a lift that still runs and no longer covers the rule.
_lift(app, "_languageName"),
_lift(app, "audioTrackLabel"),
])
script.write_text(
# The environment is modelled; the function under test is the shipped
# text above. `t` is only reached for a track with no usable language.
f"const getLocale = () => '{locale}';\n"
"const t = (k, p) => `${k}:${p.n}`;\n"
+ src
+ "\nconst out = JSON.parse(process.argv[2]).map(audioTrackLabel);\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_both_iso_639_2_variants_name_the_same_language(tmp_path, app):
"""Real files in one library use each: `fre` and `fra` are both French,
`ger` and `deu` both German. A fold that covers only one variant leaves
half a collection labelled with a three-letter code."""
out = _label_cases(tmp_path, app, [
{"i": 0, "lang": "fre", "ch": 2},
{"i": 1, "lang": "fra", "ch": 2},
{"i": 2, "lang": "ger", "ch": 2},
{"i": 3, "lang": "deu", "ch": 2},
])
assert out[0] == out[1], f"fre and fra must agree: {out}"
assert out[2] == out[3], f"ger and deu must agree: {out}"
assert "French" in out[0] and "German" in out[2], out
def test_the_container_title_wins_over_the_language_name(tmp_path, app):
"""Two tracks in one language are the same menu entry twice without it."""
out = _label_cases(tmp_path, app, [
{"i": 0, "lang": "fre", "title": "Surround", "ch": 6},
{"i": 1, "lang": "fre", "title": "Stereo", "ch": 2},
])
assert out[0] != out[1]
assert "Surround" in out[0] and "Stereo" in out[1]
def test_an_untitled_track_falls_back_to_its_channel_layout(tmp_path, app):
"""Still two distinguishable entries, and "5.1" needs no catalogue entry."""
out = _label_cases(tmp_path, app, [
{"i": 0, "lang": "fre", "ch": 6},
{"i": 1, "lang": "fre", "ch": 2},
])
assert out[0] != out[1], f"two French tracks must not read alike: {out}"
assert "5.1" in out[0] and "2.0" in out[1], out
def test_an_untagged_track_is_numbered_not_called_unknown(tmp_path, app):
"""`und`, or no tag at all — common enough in real files, and a number the
viewer can act on beats a word that tells them nothing."""
out = _label_cases(tmp_path, app, [
{"i": 0, "lang": "und", "ch": 2},
{"i": 1, "ch": 2},
])
assert out[0].startswith("video.audio_track_n:1"), out
assert out[1].startswith("video.audio_track_n:2"), out
def test_a_language_the_fold_does_not_know_shows_its_tag(tmp_path, app):
"""More useful than "Unknown": the tag is what the file actually says."""
out = _label_cases(tmp_path, app, [{"i": 0, "lang": "qaa", "ch": 2}])
assert "qaa" in out[0], out
def test_a_language_name_is_cased_for_a_menu_not_for_prose(tmp_path, app):
"""`Intl.DisplayNames` returns "français", which is right in a sentence and
wrong in a menu beside "AC3 5.1". The locales where this bites are the ones
this project ships most of."""
out = _label_cases(tmp_path, app, [
{"i": 0, "lang": "fre", "ch": 6},
{"i": 1, "lang": "eng", "ch": 6},
], locale="fr")
assert out == ["Français — 5.1", "Anglais — 5.1"], out
# ── The capability gate ───────────────────────────────────────────────────────
def test_the_selector_is_drawn_from_the_nodes_list_and_no_version(app):
"""The whole negotiation. A version comparison anywhere near this is the
bug: the floor is below the current version, so an older node is reachable,
and the only honest signal that it can switch track is that it enumerated."""
player = _player(app)
assert "audioTracks.length > 1" in player, \
"the selector must be gated on the list the node sent"
menu = player[player.index("audioTracks.length > 1"):]
menu = menu[:menu.index("platform.capabilities.lanCast")]
for forbidden in ("MNP_VERSION", "peerVersion", "3.2", "v_min"):
assert forbidden not in menu, (
f"the selector branches on {forbidden}, not on what the node said")
def test_no_track_is_ever_sent_to_a_node_that_did_not_offer_one():
"""`requestStream` must omit the field rather than default it.
A node that ignores `audio_track` does not degrade — it serves a different
language and says nothing. Sending 0 "harmlessly" to every node is what
makes that reachable, so the field has to be absent unless the caller was
given a real one.
"""
src = TRANSPORT.read_text()
fn = src[src.index("requestStream(fileId"):]
fn = fn[:fn.index("\n /** Room for")]
assert "Number.isInteger(audioTrack)" in fn, \
"an omitted track must not become 0 on the wire"
assert "req.audio_track = audioTrack" in fn
assert "audio_track: audioTrack" not in fn, \
"the field must be added conditionally, never built into the literal"
def test_changing_track_restarts_the_stream_where_the_film_already_was(app):
"""Setting the ref alone changes nothing: one ffmpeg carries one audio
track, so the node has to be asked again — and at the current position,
because the viewer changed language and not place."""
player = _player(app)
handler = player[player.index("audioTracks.length > 1"):]
handler = handler[:handler.index("platform.capabilities.lanCast")]
assert "audioTrackRef.current = track.i" in handler
assert "requestSeekRef.current" in handler, \
"a track change must go through the seek path"
assert "seek(v.currentTime)" in handler, \
"a track change must resume where the film already was"
def test_the_player_believes_the_node_about_which_track_is_playing(app):
"""`stream_init` reports the track actually used, which is not always the
one asked for — a list drawn before the file was replaced on disk can name
a track that is gone, and the node falls back to the first. Showing the
request instead of the answer would tick the wrong entry for the rest of
the film."""
player = _player(app)
assert "Number.isInteger(msg.audio_track)" in player
init = player[player.index("Number.isInteger(msg.audio_track)"):]
init = init[:init.index("}")]
assert "audioTrackRef.current = msg.audio_track" in init
|