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
|
"""
The indexing dock: what it shows, and who feeds it.
Adding a 900 GB directory left the operator nothing to look at once they left
the Settings panel — and nothing at all when the directory was added from
another machine, where that panel's bar never started. The dock sits above the
music bar on every page. What it shows is worked out in `index-dock-model.js`,
which has no imports so that it runs here under node as shipped; the wiring
around it is checked in the source, as the other SPA tests do.
"""
import json
import re
import shutil
import subprocess
from pathlib import Path
import pytest
from spa_source import transport_source
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
MODEL = STATIC / "index-dock-model.js"
DOCK = STATIC / "index-dock.js"
APP = STATIC / "app.js"
GROUP_PAGE = STATIC / "group-page.js"
TRANSPORT = STATIC / "transport.js"
pytestmark = pytest.mark.skipif(not MODEL.exists(), reason="SPA sources unavailable")
GB = 1024 ** 3
needs_node = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")
def _run(tmp_path, body: str):
(tmp_path / "package.json").write_text('{"type":"module"}')
(tmp_path / "model.js").write_text(MODEL.read_text(encoding="utf-8"))
script = tmp_path / "case.js"
script.write_text("import * as m from './model.js';\n" + body)
proc = subprocess.run(["node", str(script)], capture_output=True, text=True,
cwd=str(tmp_path))
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def _job(**over) -> dict:
job = {"groupId": "g1", "groupName": "outputs", "scanning": True, "kind": "scan",
"root": "results", "scannedBytes": 0, "totalBytes": 0, "filesDone": 0,
"filesTotal": 0, "queued": []}
job.update(over)
return job
def _frames(tmp_path, frames: list[tuple[int, list[dict]]], hide_at: int | None = None):
"""Feed dockRows a sequence of (time ms, jobs) and return each frame's rows."""
return _run(tmp_path, f"""
const frames = {json.dumps(frames)};
const hideAt = {json.dumps(hide_at)};
let memory = {{}};
const out = [];
for (const [now, jobs] of frames) {{
if (hideAt === now) memory = m.hideRow(memory, 'g1');
const r = m.dockRows(jobs, memory, now);
memory = r.memory;
out.push(r.rows.map((row) => [row.groupId, row.state]));
}}
console.log(JSON.stringify(out));
""")
# ── What is shown ───────────────────────────────────────────────────────────
@needs_node
def test_a_root_scan_shows_at_once(tmp_path):
assert _frames(tmp_path, [(0, [_job()])]) == [[["g1", "running"]]]
@needs_node
def test_a_small_burst_waits_five_seconds_and_a_large_one_does_not(tmp_path):
small = _job(kind="watch", root="", totalBytes=10_000)
assert _frames(tmp_path, [(0, [small]), (4000, [small]), (5000, [small])]) == [
[], [], [["g1", "running"]]]
large = _job(kind="reconcile", totalBytes=2 * GB)
assert _frames(tmp_path, [(0, [large])]) == [[["g1", "running"]]]
@needs_node
def test_a_short_burst_never_flashes_a_finished_row(tmp_path):
small = _job(kind="watch", root="", totalBytes=10_000)
idle = _job(scanning=False, kind="")
assert _frames(tmp_path, [(0, [small]), (1000, [idle])]) == [[], []]
@needs_node
def test_a_group_between_two_roots_stays_on_screen(tmp_path):
"""The gap between one root's walk ending and the next taking the lock."""
waiting = _job(scanning=False, kind="", root="", queued=["archive"])
assert _frames(tmp_path, [(0, [_job()]), (2000, [waiting])]) == [
[["g1", "running"]], [["g1", "running"]]]
@needs_node
def test_finished_says_so_for_four_seconds_then_goes(tmp_path):
idle = _job(scanning=False, kind="", root="")
assert _frames(tmp_path, [(0, [_job()]), (1000, [idle]), (4999, [idle]),
(5000, [idle])]) == [
[["g1", "running"]], [["g1", "done"]], [["g1", "done"]], []]
@needs_node
def test_a_hidden_row_stays_hidden_until_the_group_is_idle(tmp_path):
second = _job(root="archive")
idle = _job(scanning=False, kind="", root="")
frames = [(0, [_job()]), (1000, [_job()]), (2000, [second]), (3000, [idle]),
(9000, [idle]), (10000, [_job(root="photos")])]
assert _frames(tmp_path, frames, hide_at=1000) == [
[["g1", "running"]], [], [], [], [], [["g1", "running"]]]
# ── Where it comes from ─────────────────────────────────────────────────────
@needs_node
def test_the_loopback_answer_wins_and_the_hub_names_the_group(tmp_path):
out = _run(tmp_path, """
const local = { g1: m.fromLoopback({ group_id: 'g1', group_name: 'node name',
scanning: true, kind: 'scan', root: 'results', queued: ['archive'] }) };
const pushed = {
g1: m.fromPush('g1', { scanning: true, kind: 'rescan', root_pos: 0 }, [{ name: 'x' }]),
g2: m.fromPush('g2', { scanning: true }, []),
};
console.log(JSON.stringify(
m.mergeActivity(local, pushed, [{ id: 'g1', name: 'outputs' }])));
""")
by_id = {j["groupId"]: j for j in out}
assert (by_id["g1"]["kind"], by_id["g1"]["root"], by_id["g1"]["groupName"]) == (
"scan", "results", "outputs")
assert by_id["g1"]["queued"] == ["archive"]
assert by_id["g2"]["groupName"] == "g2"
@needs_node
def test_a_push_is_named_from_the_roots_table_it_does_not_carry(tmp_path):
out = _run(tmp_path, """
const roots = [{ name: 'outputs' }, { name: 'results' }];
console.log(JSON.stringify([
m.fromPush('g', { scanning: true, kind: 'scan', root_pos: 1, queued: 2 }, roots),
m.fromPush('g', { scanning: true, root_pos: 7 }, roots),
m.fromPush('g', { scanning: true, scanned_bytes: 5, total_bytes: 10 }, null),
]));
""")
assert (out[0]["root"], out[0]["queued"]) == ("results", 2)
assert out[1]["root"] == "", "a position past the table must name nothing"
assert (out[2]["kind"], out[2]["root"], out[2]["queued"]) == ("", "", 0), (
"a node older than the dock must read as a plain scan")
# ── Wiring ──────────────────────────────────────────────────────────────────
def test_the_dock_is_on_every_page_not_in_a_route():
app = APP.read_text(encoding="utf-8")
mount = app.index("<${IndexingDock}")
assert app.index("</main>") < mount < app.index("<${MusicPlayerBar}"), (
"the dock must sit outside the routed page, right before the music bar")
def test_only_an_operator_connection_feeds_the_dock():
page = GROUP_PAGE.read_text(encoding="utf-8")
calls = [m.start() for m in re.finditer(r"reportIndexPush\(groupId, (?!null)", page)]
assert len(calls) == 2, "the push handler and the handshake ack"
for at in calls:
guard = page[max(0, at - 200):at]
assert "transport.memberRole === 'operator'" in guard, (
"a member's connection must keep the sidebar dot only")
def test_leaving_a_group_clears_its_row():
page = GROUP_PAGE.read_text(encoding="utf-8")
cleanup = page[page.index(" return () => {\n cancelled = true;"):]
cleanup = cleanup[:cleanup.index("};")]
assert "reportIndexPush(groupId, null)" in cleanup
def test_the_transport_passes_the_new_counters_through():
source = transport_source()
block = source[source.index("msg.type === 'index_progress'"):]
block = block[:block.index("return;")]
for field in ("files_done", "files_total", "kind", "root_pos", "queued"):
assert f"{field}:" in block, f"{field} is dropped before it reaches the page"
def test_the_dock_polls_the_node_wide_route_not_one_group():
source = DOCK.read_text(encoding="utf-8")
assert "'/api/index-status'" in source
assert "/index-status`" not in source
|