diff options
Diffstat (limited to 'packages/meshbay-node/tests')
5 files changed, 306 insertions, 8 deletions
diff --git a/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py index 2dc31c9..a80645a 100644 --- a/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py +++ b/packages/meshbay-node/tests/test_added_root_does_not_wait_for_its_scan.py @@ -60,10 +60,10 @@ class _GatedIndexer(DirectoryIndexer): self.gate = asyncio.Event() self.at_gate = asyncio.Event() - async def _scan_root(self, root): + async def _scan_root(self, root, **kwargs): self.at_gate.set() await self.gate.wait() - return await super()._scan_root(root) + return await super()._scan_root(root, **kwargs) async def _indexer(one: Path, **kwargs) -> _GatedIndexer: diff --git a/packages/meshbay-node/tests/test_index_jobs_are_described.py b/packages/meshbay-node/tests/test_index_jobs_are_described.py new file mode 100644 index 0000000..dee6e39 --- /dev/null +++ b/packages/meshbay-node/tests/test_index_jobs_are_described.py @@ -0,0 +1,207 @@ +""" +What the operator's progress band is told about the indexing under way. + +`progress` said "scanning, this many bytes of that many" and nothing more. That +is one bar with no name on it. A node asked to add a second directory while the +first is still hashing does them one after the other — one scan lock, one +hashing thread — and an operator looking at a bar that jumps back to 0 % cannot +tell a second directory from a scan that started over. So it now says which +root, what kind of walk, how many files, and which roots wait their turn. +""" + +import asyncio +import os +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet + +pytestmark = pytest.mark.asyncio + +GROUP = "g" * 32 + + +def _set(*specs) -> RootSet: + return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name} + for s in specs]) + + +def _names(idx) -> list[str]: + return sorted(e.name for e in idx.index.entries) + + +async def _until(predicate, timeout: float = 3.0) -> bool: + deadline = asyncio.get_running_loop().time() + timeout + while not predicate(): + if asyncio.get_running_loop().time() > deadline: + return False + await asyncio.sleep(0.02) + return True + + +class _Held(DirectoryIndexer): + """Stops before hashing the first file under `hold_under`, and records walks.""" + hold_under: Path | None = None + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.gate = asyncio.Event() + self.at_gate = asyncio.Event() + self.walked: list[str] = [] + + async def _scan_root(self, root, *args, **kwargs): + self.walked.append(root.name) + return await super()._scan_root(root, *args, **kwargs) + + async def _hash_or_cached(self, root, file_path): + if self.hold_under is not None and file_path.is_relative_to(self.hold_under): + self.at_gate.set() + await self.gate.wait() + return await super()._hash_or_cached(root, file_path) + + +def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]: + one, r1, r2 = tmp_path / "one", tmp_path / "r1", tmp_path / "r2" + for d in (one, r1, r2): + d.mkdir() + (one / "a.txt").write_bytes(b"first root") + (r1 / "b.bin").write_bytes(os.urandom(2000)) + (r1 / "c.bin").write_bytes(os.urandom(3000)) + (r2 / "d.bin").write_bytes(os.urandom(4000)) + return one, r1, r2 + + +async def _drain(idx) -> None: + await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5) + + +async def test_the_root_under_way_and_the_ones_waiting_are_named(tmp_path): + one, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.walked.clear() + idx.hold_under = r1 + try: + await idx.retarget(_set(one, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + p = idx.progress + assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "scan", "r1", 1) + assert (p.files_done, p.files_total, p.total_bytes) == (0, 2, 5000) + assert p.queued == [] + + await idx.retarget(_set(one, r1, r2), wait=False) + assert p.queued == ["r2"] + + idx.gate.set() + await _drain(idx) + assert idx.walked == ["r1", "r2"], "the second root did not wait for the first" + assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", []) + assert _names(idx) == ["a.txt", "b.bin", "c.bin", "d.bin"] + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_root_removed_while_it_waited_leaves_the_queue(tmp_path): + one, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.walked.clear() + idx.hold_under = r1 + try: + await idx.retarget(_set(one, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + await idx.retarget(_set(one, r1, r2), wait=False) + await idx.retarget(_set(one, r1), wait=False) + assert idx.progress.queued == [], "a removed root is still announced as next" + + idx.gate.set() + await _drain(idx) + assert idx.walked == ["r1"] + assert idx.progress.queued == [] + finally: + idx.gate.set() + await idx.stop() + + +async def test_the_initial_scan_names_the_roots_still_to_come(tmp_path): + _, r1, r2 = _dirs(tmp_path) + idx = _Held(roots=_set(r1, r2), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + idx.hold_under = r1 + scan = asyncio.create_task(idx.initial_scan()) + try: + await asyncio.wait_for(idx.at_gate.wait(), 5) + assert (idx.progress.root, idx.progress.queued) == ("r1", ["r2"]) + idx.gate.set() + await asyncio.wait_for(scan, 5) + assert idx.progress.queued == [] + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path): + _, r1, r2 = _dirs(tmp_path) + + class _Broken(DirectoryIndexer): + async def _hash_or_cached(self, root, file_path): + raise RuntimeError("simulated failure mid-scan") + + idx = _Broken(roots=_set(r1, r2), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + with pytest.raises(RuntimeError): + await idx.initial_scan() + p = idx.progress + assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", []) + + +async def test_a_plug_waiting_its_turn_is_announced(tmp_path): + one, r1, _ = _dirs(tmp_path) + two = tmp_path / "two" + two.mkdir() + (two / "e.txt").write_bytes(b"removable") + idx = _Held(roots=_set(one, two), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None) + await idx.initial_scan() + idx.eject_root("two") + idx.hold_under = r1 + try: + await idx.retarget( + _set(one, {"path": str(two), "name": "two", "ejected": True}, r1), wait=False) + await asyncio.wait_for(idx.at_gate.wait(), 5) + plug = asyncio.create_task(idx.plug_root("two")) + assert await _until(lambda: idx.progress.queued == ["two"]) + + idx.gate.set() + await asyncio.wait_for(plug, 5) + assert idx.progress.queued == [] + assert idx.walked[-1] == "two" + finally: + idx.gate.set() + await idx.stop() + + +async def test_a_burst_is_described_as_watching(tmp_path): + one, _, _ = _dirs(tmp_path) + idx = DirectoryIndexer(roots=_set(one), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=None, + debounce_secs=0.01) + await idx.initial_scan() + idx._loop = asyncio.get_running_loop() + for i in range(2): + f = one / f"new{i}.bin" + f.write_bytes(os.urandom(1000)) + idx._schedule_update(f) + await asyncio.sleep(0) + + p = idx.progress + assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "watch", "", -1) + assert (p.files_done, p.files_total) == (0, 2) + + assert await _until(lambda: not p.scanning) + assert (p.kind, p.files_done, p.files_total) == ("", 2, 2) diff --git a/packages/meshbay-node/tests/test_index_no_cleartext.py b/packages/meshbay-node/tests/test_index_no_cleartext.py index f510884..e5e4555 100644 --- a/packages/meshbay-node/tests/test_index_no_cleartext.py +++ b/packages/meshbay-node/tests/test_index_no_cleartext.py @@ -151,9 +151,15 @@ def test_index_progress_stays_clear_and_stays_counters(): dicts = [n for n in ast.walk(tree) if isinstance(n, ast.Dict)] assert len(dicts) == 1, "more than one message built here — re-read this test" keys = {k.value for k in dicts[0].keys} + # `kind` is one of four fixed words and `root_pos`/`queued` are integers — + # the root names themselves stay on the loopback API. assert keys == {"type", "v", "group_id", - "scanning", "scanned_bytes", "total_bytes"}, ( + "scanning", "scanned_bytes", "total_bytes", + "files_done", "files_total", "kind", "root_pos", "queued"}, ( f"index_progress now carries {keys} — re-read decision D3 before shipping it") + values = {k.value: v for k, v in zip(dicts[0].keys, dicts[0].values)} + assert ast.unparse(values["queued"]) == "len(progress.queued)", ( + "the roots waiting must go out as a count, never as their names") assert "seal(" not in source assert "D3" in source, "the reason it is not sealed must stay next to the code" diff --git a/packages/meshbay-node/tests/test_index_progress.py b/packages/meshbay-node/tests/test_index_progress.py index 52cd78b..df51e85 100644 --- a/packages/meshbay-node/tests/test_index_progress.py +++ b/packages/meshbay-node/tests/test_index_progress.py @@ -36,19 +36,26 @@ def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * def test_indexing_status_defaults_idle_when_no_progress_tracked(): session = _session_with_progress(None) assert session._indexing_status() == { - "scanning": False, "scanned_bytes": 0, "total_bytes": 0} + "scanning": False, "scanned_bytes": 0, "total_bytes": 0, + "files_done": 0, "files_total": 0, "kind": "", "root_pos": -1, "queued": 0} def test_indexing_status_reflects_live_progress(): progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000, - current_dir="StarWars") + current_dir="Season 2", root="series", root_pos=1, + kind="scan", files_done=3, files_total=9, + queued=["archive", "photos"]) session = _session_with_progress(progress) status = session._indexing_status() - assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000} + assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000, + "files_done": 3, "files_total": 9, "kind": "scan", + "root_pos": 1, "queued": 2} assert "current_dir" not in status, \ "the directory name is operator-local detail, never sent to a member" + assert not {"series", "archive", "photos"} & {str(v) for v in status.values()}, \ + "a root name reached a member" # ── /api/groups/{id}/index-status (loopback) ──────────────────────────────── @@ -81,6 +88,36 @@ def test_index_status_route_reflects_indexer_progress(): } +# ── /api/index-status (loopback, every group) ─────────────────────────────── + +def test_every_group_is_described_including_one_still_being_attached(): + """The band reads one route for the whole node. A group in its initial scan + is in state["indexers"] and not yet in groups_ctx, and must be there.""" + busy, idle = MagicMock(), MagicMock() + busy.progress = IndexProgress( + scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024", + root="results", root_pos=1, kind="scan", files_done=1, files_total=4, + queued=["archive"]) + idle.progress = IndexProgress() + config = MagicMock() + named = MagicMock() + named.id, named.name = "a" * 32, "outputs" + config.groups = [named] + client = _ui_client({"config": config, + "indexers": {"a" * 32: busy, "b" * 32: idle}}) + + groups = {g["group_id"]: g for g in client.get("/api/index-status").json()["groups"]} + + assert groups["a" * 32] == { + "group_id": "a" * 32, "group_name": "outputs", "scanning": True, + "kind": "scan", "root": "results", "current_dir": "2024", + "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4, + "queued": ["archive"], + } + assert groups["b" * 32]["group_name"] == "b" * 8 + assert groups["b" * 32]["scanning"] is False + + # ── _push_index_progress / _progress_pusher ───────────────────────────────── def _daemon(tmp_path) -> NodeDaemon: @@ -123,6 +160,54 @@ async def test_push_index_progress_only_reaches_same_group_peers(tmp_path): @pytest.mark.asyncio +async def test_push_index_progress_carries_counters_never_root_names(tmp_path): + daemon = _daemon(tmp_path) + session = MagicMock() + session._group_id = "a" * 32 + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + daemon._push_index_progress("a" * 32, IndexProgress( + scanning=True, scanned_bytes=10, total_bytes=100, current_dir="2024", + root="results", root_pos=2, kind="rescan", files_done=5, files_total=50, + queued=["archive", "photos"])) + + msg = session._send.call_args[0][0] + assert (msg["kind"], msg["root_pos"], msg["queued"]) == ("rescan", 2, 2) + assert (msg["files_done"], msg["files_total"]) == (5, 50) + assert not {"results", "archive", "photos", "2024"} & {str(v) for v in msg.values()} + + +@pytest.mark.asyncio +async def test_progress_pusher_speaks_while_a_root_only_waits(tmp_path): + """Between one root's scan ending and the next one taking the lock, nothing + is scanning — but there is work coming, and the band must not drop it.""" + daemon = _daemon(tmp_path) + session = MagicMock() + session._group_id = "a" * 32 + mock_webrtc = MagicMock() + mock_webrtc._sessions = {"p1": session} + daemon._webrtc = mock_webrtc + + indexer = MagicMock() + indexer.group_id = "a" * 32 + indexer.progress = IndexProgress(scanning=False, queued=["archive"]) + + task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05)) + try: + await asyncio.sleep(0.12) + assert session._send.call_count >= 2 + assert session._send.call_args.args[0]["queued"] == 1 + finally: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + +@pytest.mark.asyncio async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path): daemon = _daemon(tmp_path) diff --git a/packages/meshbay-node/tests/test_root_work_outlives_the_session.py b/packages/meshbay-node/tests/test_root_work_outlives_the_session.py index 31de12e..1e31c31 100644 --- a/packages/meshbay-node/tests/test_root_work_outlives_the_session.py +++ b/packages/meshbay-node/tests/test_root_work_outlives_the_session.py @@ -149,11 +149,11 @@ async def test_a_plug_whose_caller_goes_away_does_not_empty_the_root(tmp_path): gate = None at_gate = None - async def _scan_root(self, root): + async def _scan_root(self, root, **kwargs): if self.gate is not None: self.at_gate.set() await self.gate.wait() - return await super()._scan_root(root) + return await super()._scan_root(root, **kwargs) idx = _Held(roots=_set(one, two), group_id=GROUP, sk_node=Ed25519PrivateKey.generate(), gek=None) |