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
|
"""
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"
|