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