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

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