1 files changed, 207 insertions, 0 deletions
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)