""" Safe eject, and the surprise unplug it exists to survive. `test_root_availability.py` pins the freeze: a root that goes away keeps its entries. This pins the half the operator drives — telling the node the drive is about to leave, and telling it the drive is back. The distinction that makes any of this work is that `ejected` and `is_live()` are separate answers. Between clicking Eject and physically unplugging, the directory is still readable; a design that recomputed availability from the filesystem alone would flip the root straight back to available and start serving files from a disk somebody has their hand on. The other property here is that the flag is *persisted*. It reached the roster in the first implementation and was never read back, so a restart — which is exactly what an operator does after noticing a drive fell off — silently undid the eject, and the next scan read an empty mount point as an erased library. """ import asyncio 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 from meshbay_node.roster import Roster pytestmark = pytest.mark.asyncio def _roots(*paths: Path, removable: bool = True) -> RootSet: return RootSet.build([ {"path": str(p), "removable": removable} for p in paths]) async def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: idx = DirectoryIndexer(roots=roots, group_id="g" * 32, sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) await idx.initial_scan() return idx def _names(idx: DirectoryIndexer) -> set[str]: return {e.name for e in idx.index.entries} # ── The two states are not the same question ───────────────────────────────── async def test_ejecting_hides_a_root_that_is_still_readable(tmp_path): """ The whole point of an eject button: the operator says the drive is leaving *before* it leaves. The directory is still there and still readable at this moment, so anything deriving availability from the filesystem would refuse to believe it. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films) idx = await _indexer(roots) idx.eject_root("Films") assert films.is_dir(), "the drive has not been unplugged yet" assert roots.roots[0].is_live() is True assert roots.roots[0].available is False assert idx.index.roots[0]["ejected"] is True assert idx.index.roots[0]["available"] is False async def test_an_eject_freezes_entries_rather_than_dropping_them(tmp_path): films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") (films / "b.mkv").write_bytes(b"b") idx = await _indexer(_roots(films)) idx.eject_root("Films") assert _names(idx) == {"a.mkv", "b.mkv"}, "eject deleted entries" async def test_reconciling_does_not_un_eject_a_root(tmp_path): """ The backstop runs every minute regardless. An ejected root whose directory is still readable must stay ejected, or the operator's eject lasts until the next tick. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films) idx = await _indexer(roots) idx.eject_root("Films") await idx.reconcile() assert roots.roots[0].ejected is True assert roots.roots[0].available is False async def test_plugging_back_relists_the_files(tmp_path): films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films) idx = await _indexer(roots) idx.eject_root("Films") await idx.plug_root("Films") assert roots.roots[0].ejected is False assert roots.roots[0].available is True assert _names(idx) == {"a.mkv"} async def test_what_changed_while_unplugged_is_picked_up_on_plug(tmp_path): """ A drive people take away comes back different. The plug pass has to see that, or the index describes a library that no longer exists on the disk the node is about to serve from. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films) idx = await _indexer(roots) idx.eject_root("Films") (films / "a.mkv").unlink() (films / "c.mkv").write_bytes(b"c") await idx.plug_root("Films") assert _names(idx) == {"c.mkv"} # ── The surprise unplug ────────────────────────────────────────────────────── async def test_a_removable_root_that_vanishes_is_auto_ejected(tmp_path): """ Nobody clicks Eject when they are in a hurry. A removable root whose path disappears is treated as ejected rather than merely unavailable, so it does not silently come back the moment the same mount point is readable again — which on a machine with automount is any other drive, or an empty stub. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films) idx = await _indexer(roots) (films / "a.mkv").unlink() films.rmdir() await idx.reconcile() assert roots.roots[0].ejected is True assert _names(idx) == {"a.mkv"}, "the library was treated as erased" async def test_a_non_removable_root_is_not_auto_ejected(tmp_path): """ The counter-property. Auto-eject requires the operator to have said the device is removable; an ordinary directory that briefly fails to stat must keep the old behaviour and come back on its own. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") roots = _roots(films, removable=False) idx = await _indexer(roots) (films / "a.mkv").unlink() films.rmdir() await idx.reconcile() assert roots.roots[0].ejected is False assert roots.roots[0].available is False films.mkdir() (films / "a.mkv").write_bytes(b"a") await idx.reconcile() assert roots.roots[0].available is True async def test_an_auto_eject_is_reported_so_it_can_be_persisted(tmp_path): """ The flag has to outlive the process. The first version of this set it in memory only, so restarting the node — which is what an operator does after noticing a drive fell off — cleared it, and the scan that followed read the empty mount point as a deletion of the whole library. """ films = tmp_path / "Films" films.mkdir() (films / "a.mkv").write_bytes(b"a") seen: list[tuple[str, bool]] = [] async def record(name: str, ejected: bool) -> None: seen.append((name, ejected)) roots = _roots(films) idx = await _indexer(roots, on_root_ejected=record) (films / "a.mkv").unlink() films.rmdir() await idx.reconcile() assert seen == [("Films", True)] # And only once, however many times the backstop runs afterwards. await idx.reconcile() await idx.reconcile() assert seen == [("Films", True)] # ── Restoring the flag ─────────────────────────────────────────────────────── async def test_a_root_built_as_ejected_starts_unavailable(tmp_path): """ What the daemon does with what the roster remembers. `available` must not be left at its default `True` here, or the group serves a drive that is not there for as long as it takes the first reconcile to run. """ films = tmp_path / "Films" films.mkdir() roots = RootSet.build([{"path": str(films), "removable": True, "ejected": True}]) assert roots.roots[0].ejected is True assert roots.roots[0].available is False async def test_the_roster_round_trips_the_ejected_set(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: assert await roster.ejected_roots("g1") == set() await roster.set_root_ejected("g1", "Films", True, set_by="op") await roster.set_root_ejected("g1", "Music", False, set_by="op") assert await roster.ejected_roots("g1") == {"films"} # Another group's drives are its own. assert await roster.ejected_roots("g2") == set() await roster.set_root_ejected("g1", "Films", False, set_by="op") assert await roster.ejected_roots("g1") == set() finally: await roster.close() async def test_the_ejected_key_is_case_folded(tmp_path): """ Root names are compared without regard to case everywhere else, and a key that did not fold would let `Films` and `films` disagree about the same drive — on Windows and macOS, the same directory. """ roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: await roster.set_root_ejected("g1", "FILMS", True, set_by="op") assert await roster.ejected_roots("g1") == {"films"} assert Roster.root_ejected_key("Films") == Roster.root_ejected_key("FILMS") finally: await roster.close()