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
|
"""
Cast receivers are listed as they answer, not at the end of the scan.
The scan runs its full length because a receiver coming back from a reset can
take several seconds to answer, but most answer within two; a picker that showed
nothing until the end was slow every time it was opened. These tests run the real
`CastChromecast` with mDNS replaced by a stub that answers on cue, and the clock
shortened, so what they measure is what the page's poll would see.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
CLIENT = Path(__file__).resolve().parents[2] / "meshbay-client"
CHROMECAST = CLIENT / "src" / "cast-chromecast.js"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not CHROMECAST.exists(),
reason="node or the desktop client sources are not available")
# Stubs the two dependencies at `require` time, and makes six seconds of scan
# last sixty milliseconds by scaling every timer the module arms.
SCRIPT = r"""
const Module = require('node:module');
const browsers = [];
const realLoad = Module._load;
Module._load = function (request, ...rest) {
if (request === 'bonjour-service') {
return { Bonjour: class {
find(_q, onUp) {
const b = { onUp, stopped: false, stop() { this.stopped = true; } };
browsers.push(b);
return b;
}
} };
}
if (request === 'castv2-client') return { Client: class {}, DefaultMediaReceiver: {} };
return realLoad.call(this, request, ...rest);
};
const realSetTimeout = global.setTimeout;
global.setTimeout = (fn, ms, ...a) => realSetTimeout(fn, ms / 100, ...a);
const wait = (ms) => new Promise((r) => realSetTimeout(r, ms));
const CastChromecast = require(process.argv[2]);
const tv = (id) => ({ name: id, txt: { id, fn: `TV ${id}` },
addresses: ['10.0.0.9'], port: 8009 });
(async () => {
const cc = new CastChromecast();
const out = {};
cc.startScan();
out.atStart = cc.devices();
browsers[0].onUp(tv('a'));
out.afterFirstAnswer = cc.devices();
await wait(100);
out.afterScan = cc.devices();
out.firstBrowserStopped = browsers[0].stopped;
// A second scan started over a first: the first's timer must not end it.
cc.startScan();
await wait(40);
cc.startScan();
await wait(40);
out.restartedStillScanning = cc.devices().scanning;
out.restartClearedOldDevices = cc.devices().devices.length === 0;
await wait(60);
out.restartedEnds = !cc.devices().scanning;
console.log(JSON.stringify(out));
})();
"""
@pytest.fixture(scope="module")
def run(tmp_path_factory):
script = tmp_path_factory.mktemp("cd") / "discover.cjs"
script.write_text(SCRIPT, encoding="utf-8")
proc = subprocess.run(
["node", str(script), str(CHROMECAST)],
capture_output=True, text=True, encoding="utf-8", timeout=60)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def test_a_receiver_is_listed_before_the_scan_ends(run):
assert run["atStart"] == {"devices": [], "scanning": True}
assert run["afterFirstAnswer"]["scanning"] is True
assert [d["name"] for d in run["afterFirstAnswer"]["devices"]] == ["TV a"]
def test_the_scan_ends_on_its_own_and_keeps_what_it_found(run):
assert run["afterScan"]["scanning"] is False
assert [d["id"] for d in run["afterScan"]["devices"]] == ["a"]
assert run["firstBrowserStopped"] is True
def test_a_new_scan_is_not_cut_short_by_the_one_it_replaced(run):
"""
The old code left the first scan's timer armed, and it stopped whichever
browser was current when it fired — the new one, two thirds early.
"""
assert run["restartedStillScanning"] is True
assert run["restartClearedOldDevices"] is True
assert run["restartedEnds"] is True
def test_the_local_player_is_silent_while_casting():
"""The local element keeps playing while casting — its playhead paces the
stream the relay forwards — so it must not keep its sound: a phone in the
room doubled the television's audio, screen off included. Whatever the
viewer had set comes back when the cast ends."""
from spa_source import STATIC
player = (STATIC / "video-player.js").read_text(encoding="utf-8")
effect = player.split("const mutedBeforeCastRef = useRef(null);", 1)[1]
effect = effect.split("}, [castActive]);", 1)[0]
assert "v.muted = true;" in effect
assert "mutedBeforeCastRef.current = v.muted;" in effect
assert "v.muted = mutedBeforeCastRef.current;" in effect
assert "pause()" not in effect, "pausing would stop the stream the receiver is playing"
def test_the_player_is_a_remote_while_casting():
"""Phone and television do not play in step, so while a receiver plays the
film the scrubber shows the receiver's position (stream time plus where
the stream started) and every seek goes through the ordinary seek, which
restarts the relay and reloads the receiver there. The copy-URL cast has
no receiver to read and keeps the local player."""
from spa_source import STATIC
player = (STATIC / "video-player.js").read_text(encoding="utf-8")
remote = player.split("// ── Remote ──", 1)[1].split("return html`", 1)[0]
assert "castDeviceName !== null && platform.cast.canControl" in remote
assert "streamStartRef.current + c.position" in remote
# Until the restart lands, the receiver's position is the old stream's.
assert "!castRestartPendingRef.current" in remote
assert "requestSeekRef.current(target)" in remote
assert "platform.cast.chromecastPause()" in remote
assert "platform.cast.chromecastPlay()" in remote
# A language change restarts the stream where the television is.
assert "remoteFilmPosRef.current ?? (v && v.currentTime)" in player
def test_every_locale_names_the_remote_controls():
from spa_source import STATIC
keys = ["cast.casting_to", "cast.seek", "cast.waiting", "cast.back30", "cast.fwd30", "cast.play", "cast.pause"]
for f in sorted((STATIC / "locales").glob("*.js")):
text = f.read_text(encoding="utf-8")
for k in keys:
assert f"'{k}':" in text, f"{f.name} lacks {k}"
|