path: root/packages/meshbay-node/tests/test_root_eject.py

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