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