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