diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_root_pause.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_root_pause.py | 313 |
1 files changed, 313 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_root_pause.py b/packages/meshbay-node/tests/test_root_pause.py new file mode 100644 index 0000000..d9a4fcb --- /dev/null +++ b/packages/meshbay-node/tests/test_root_pause.py @@ -0,0 +1,313 @@ +""" +Pausing a root's indexing, and the walk that has to notice it. + +A scan of a large drive runs for hours, and nothing could stop one: the hashing +loop never looked at the root again once it had the file list. Removing the +directory let it read the whole drive anyway and threw the result away at the +end; ejecting it went on reading a drive the operator was about to unplug. + +A pause has to outlast the walk it stops. Reconcile indexes every file the index +lacks, so a walk that was merely cancelled was redone at the next tick; the flag +is what reconcile, the watcher and a restart all read. And it keeps what was +indexed: the index is rebuilt at every start, so a paused root is listed from the +hash cache rather than left empty. +""" + +import asyncio +from pathlib import Path +from types import SimpleNamespace + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_node import ops +from meshbay_node.config import GroupConfig, RootSpec +from meshbay_node.indexer import indexer as indexer_mod +from meshbay_node.indexer.cache import IndexCache +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet +from meshbay_node.roster import Roster + +pytestmark = pytest.mark.asyncio + +GID = "g" * 32 + + +def _films(tmp_path: Path, n: int = 6) -> Path: + films = tmp_path / "Films" + films.mkdir() + for i in range(n): + (films / f"f{i}.mkv").write_bytes(f"film {i}".encode()) + return films + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _count_reads(monkeypatch, after=None, at: int = 0) -> list[str]: + """Record every file actually read; run `after` once `at` have been.""" + read: list[str] = [] + real = indexer_mod._scan_file + + def counting(root, path): + read.append(path.name) + if after is not None and len(read) == at: + after() + return real(root, path) + + monkeypatch.setattr(indexer_mod, "_scan_file", counting) + return read + + +# ── The walk notices ───────────────────────────────────────────────────────── + +async def test_a_pause_stops_the_walk_at_the_next_file(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path))}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), at=2) + + await idx.initial_scan() + + assert len(read) == 2, f"read {len(read)} files after the pause" + assert len(_names(idx)) == 2, "the files hashed before the pause stay listed" + assert idx.progress.scanning is False + + +async def test_a_root_removed_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + films = _films(tmp_path) + other = tmp_path / "Other" + other.mkdir() + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def remove(): + idx.roots = RootSet.build([{"path": str(other)}]) + + read = _count_reads(monkeypatch, remove, at=1) + await idx.initial_scan() + + assert len(read) == 1, f"went on reading a removed root: {read}" + + +async def test_a_root_ejected_mid_walk_is_not_read_to_the_end(tmp_path, monkeypatch): + roots = RootSet.build([{"path": str(_films(tmp_path)), "removable": True}]) + idx = _indexer(roots) + read = _count_reads(monkeypatch, lambda: idx.eject_root("Films"), at=1) + + await idx.initial_scan() + + assert len(read) == 1, f"went on reading an ejected drive: {read}" + + +async def test_a_retarget_does_not_hide_the_pause_from_a_running_walk( + tmp_path, monkeypatch): + """The walk holds the Root it started with; a reload replaces every one.""" + films = _films(tmp_path) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + + def reload_then_pause(): + idx.roots = RootSet.build([{"path": str(films)}]) + idx.roots.roots[0].paused = True + + read = _count_reads(monkeypatch, reload_then_pause, at=1) + await idx.initial_scan() + + assert len(read) == 1 + + +# ── The pause outlasts the walk ────────────────────────────────────────────── + +async def test_reconcile_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + (films / "new.mkv").write_bytes(b"new film") + await idx.reconcile() + + assert "new.mkv" not in _names(idx), "reconcile redid the work the pause stopped" + + +async def test_the_watcher_leaves_a_paused_root_alone(tmp_path): + films = _films(tmp_path, n=1) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + await idx.pause_root("Films") + + new = films / "new.mkv" + new.write_bytes(b"new film") + await idx._update_entry(new, deleted=False) + + assert "new.mkv" not in _names(idx) + + +async def test_a_paused_root_is_listed_from_the_cache_and_nothing_is_read( + tmp_path, monkeypatch): + """ + After a restart the index is empty until the walk refills it. Skipping a + paused root would empty it for members; reading it would undo the pause. + """ + films = _films(tmp_path, n=3) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + await _indexer(RootSet.build([{"path": str(films)}]), cache=cache).initial_scan() + (films / "unhashed.mkv").write_bytes(b"never hashed") + + roots = RootSet.build([{"path": str(films), "paused": True}]) + idx = _indexer(roots, cache=cache) + read = _count_reads(monkeypatch) + await idx.initial_scan() + + assert read == [] + assert _names(idx) == {"f0.mkv", "f1.mkv", "f2.mkv"} + assert idx.progress.scanning is False + finally: + await cache.close() + + +async def test_a_pause_mid_rescan_keeps_the_entries_it_had(tmp_path, monkeypatch): + """A rescan drops the root's entries before it walks; stopping it there + would leave the operator with half a library.""" + films = _films(tmp_path, n=4) + cache = IndexCache(db_path=tmp_path / "cache.db") + await cache.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots, cache=cache) + await idx.initial_scan() + for f in films.iterdir(): + f.write_bytes(f.read_bytes() + b" edited") # every cache row stale + + read = _count_reads(monkeypatch, lambda: setattr(roots.roots[0], "paused", True), + at=1) + await idx._rescan_root(roots.roots[0]) + + assert len(read) == 1 + assert len(_names(idx)) == 1 + finally: + await cache.close() + + +# ── Resume ─────────────────────────────────────────────────────────────────── + +async def test_resume_indexes_what_is_missing_and_keeps_what_was_enriched(tmp_path): + films = _films(tmp_path, n=2) + idx = _indexer(RootSet.build([{"path": str(films)}])) + await idx.initial_scan() + for e in idx.index.entries: + e.artist = "Some Artist" + await idx.pause_root("Films") + (films / "new.mkv").write_bytes(b"new film") + + await idx.resume_root("Films") + await asyncio.gather(*idx._scan_tasks) + + assert "new.mkv" in _names(idx) + kept = [e for e in idx.index.entries if e.name != "new.mkv"] + assert all(e.artist == "Some Artist" for e in kept), \ + "resume replaced entries it should have left alone" + + +async def test_resume_returns_before_the_walk_ends(tmp_path): + films = _films(tmp_path, n=1) + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + await idx.initial_scan() + await idx.pause_root("Films") + + async with idx._scan_lock: + await asyncio.wait_for(idx.resume_root("Films"), timeout=1) + assert idx.progress.queued == ["Films"] + await asyncio.gather(*idx._scan_tasks) + assert idx.progress.queued == [] + + +# ── The flag is the operator's, and survives ───────────────────────────────── + +def _state(tmp_path: Path, roots: RootSet, roster: Roster) -> dict: + index = GroupIndex(group_id=GID, sk_node=Ed25519PrivateKey.generate()) + return { + "groups_ctx": {GID: {"index": index, "roots": roots, "gek": None}}, + "config": SimpleNamespace(groups=[SimpleNamespace(id=GID, roots=[])]), + "roster": roster, + "node_user_id": "operator", + } + + +async def test_pause_and_resume_persist_through_the_roster(tmp_path): + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build([{"path": str(films)}]), roster) + + out = await ops.pause_root(state, GID, "films") + assert out["status"] == "paused" + assert out["roots"][0]["paused"] is True + assert out["roots"][0]["available"] is True, "a paused root is still served" + assert await roster.paused_roots(GID) == {"films"} + assert (await ops.pause_root(state, GID, "Films"))["status"] == "already_paused" + + out = await ops.resume_root(state, GID, "Films") + assert out["status"] == "resumed" + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_removing_a_root_forgets_its_pause(tmp_path): + """A directory added back later under the same name starts indexed.""" + films = _films(tmp_path, n=1) + keep = tmp_path / "Keep" + keep.mkdir() + conf = tmp_path / "node.toml" + conf.write_text( + f'[[groups]]\nid = "{GID}"\nname = "g"\n\n' + f'[[groups.roots]]\npath = "{films.as_posix()}"\n\n' + f'[[groups.roots]]\npath = "{keep.as_posix()}"\n') + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + state = _state(tmp_path, RootSet.build( + [{"path": str(films)}, {"path": str(keep)}]), roster) + state["config_path"] = str(conf) + state["config"] = SimpleNamespace(groups=[GroupConfig( + id=GID, roots=[RootSpec(path=str(films)), RootSpec(path=str(keep))])]) + await ops.pause_root(state, GID, "Films") + + await ops.remove_root(state, GID, "Films") + + assert await roster.paused_roots(GID) == set() + finally: + await roster.close() + + +async def test_a_new_group_can_be_paused_during_its_first_scan(tmp_path): + """A group added while the node runs has an indexer, and no context until + that first scan ends — the scan most worth pausing.""" + films = _films(tmp_path, n=1) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + roots = RootSet.build([{"path": str(films)}]) + idx = _indexer(roots) + state = _state(tmp_path, roots, roster) + state["groups_ctx"] = {} + state["indexers"] = {GID: idx} + + out = await ops.pause_root(state, GID, "Films") + + assert out["status"] == "paused" + assert roots.roots[0].paused is True + finally: + await roster.close() |