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

"""
One implementation, several front doors.

The point of `meshbay_node.ops` is not tidiness. C1 and C6 were both "a second
path into the node with its own weaker handshake", and two implementations of
`revoke` with two authorization checks is that shape one size down. So the tests
that matter here are the ones that would fail if a second implementation
appeared: the adapters must be thin, and the operations must not decide who may
call them.
"""

import asyncio
import inspect
from pathlib import Path, PureWindowsPath
from types import SimpleNamespace

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.transport.quic_server import Denylist
from node_source import ops_source

from conftest import one_root


def _state(tmp_path: Path) -> dict:
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    return {
        "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared),
                                  "gek": None}},
        "denylist": Denylist(path=tmp_path / "denylist.json"),
        "indexes": {"g" * 32: index},
    }


# ── The adapters stay thin ───────────────────────────────────────────────────

def test_operations_take_state_and_nothing_web_shaped():
    """
    An operation that knew about HTTP could not be called from MNP without a
    second copy. Every public operation therefore takes `state` first and
    returns plain data.
    """
    # Defined in `ops`, not merely visible there: an async helper imported into
    # it (`off_disk`, say) is not an operation, and reporting it as one says
    # nothing about a second implementation appearing. Every real operation is
    # written in the module or in the package it is split into, so nothing is
    # lost by asking.
    public = [(n, f) for n, f in vars(ops).items()
              if inspect.iscoroutinefunction(f) and not n.startswith("_")
              and (getattr(f, "__module__", "") == ops.__name__
                   or getattr(f, "__module__", "").startswith(ops.__name__ + "."))]
    assert len(public) > 30, "operations have gone missing — did the module move?"
    for name, fn in public:
        params = list(inspect.signature(fn).parameters)
        assert params and params[0] == "state", (
            f"{name} does not take state first — an adapter would have to "
            f"assemble something for it, which is where a second implementation "
            f"begins")


def test_the_http_adapter_adds_no_logic():
    """
    Each loopback handler should be a call into `ops` and nothing else. A
    handler that grew a check of its own would be a rule the MNP path does not
    have.
    """
    import meshbay_node.ui.app as ui
    source = inspect.getsource(ui)
    # Every endpoint that performs an operation routes through _op(...).
    for endpoint in ("operator_pair", "create_invite", "revoke_member",
                     "unpin_member", "init_gek", "attach_group", "delete_file",
                     "add_root", "remove_root", "update_root",
                     "eject_root", "plug_root", "reload_config"):
        start = source.index(f"async def {endpoint}(")
        body = source[start:start + 700]
        assert "_op(" in body.split("\n\n")[0] + body, (
            f"{endpoint} does not go through the shared adapter")
        assert "roster.set_status" not in body and "generate_gek" not in body, (
            f"{endpoint} performs the operation itself instead of calling ops")


def test_op_errors_carry_a_status_without_importing_http():
    source = ops_source()
    for forbidden in ("JSONResponse", "fastapi", "starlette", "HTTPException"):
        assert forbidden not in source, (
            f"ops imports {forbidden} — it must not know which adapter called it")


# ── Denylist (14.10) ─────────────────────────────────────────────────────────

async def test_denylist_reports_what_it_refuses(tmp_path):
    state = _state(tmp_path)
    state["denylist"].deny_user("alice")
    state["denylist"].deny_group("g" * 32)

    out = await ops.read_denylist(state)

    assert out["users"] == ["alice"]
    assert out["groups"] == ["g" * 32]
    assert out["count"] == 2


async def test_clearing_one_subject_leaves_the_rest(tmp_path):
    state = _state(tmp_path)
    state["denylist"].deny_user("alice")
    state["denylist"].deny_user("bob")

    out = await ops.clear_denylist(state, subject="alice")

    assert out["removed"] == 1
    assert (await ops.read_denylist(state))["users"] == ["bob"]


async def test_clearing_everything_says_how_much(tmp_path):
    """The count is the point: it tells the operator whether they undid one
    revocation or all of them."""
    state = _state(tmp_path)
    for name in ("a", "b", "c"):
        state["denylist"].deny_user(name)

    out = await ops.clear_denylist(state)

    assert out["removed"] == 3
    assert (await ops.read_denylist(state))["count"] == 0


async def test_denylist_survives_a_restart(tmp_path):
    """Finding H4: revocations used to live only in memory, so a restart
    silently un-revoked everyone."""
    state = _state(tmp_path)
    state["denylist"].deny_user("alice")

    reopened = Denylist(path=tmp_path / "denylist.json")
    assert reopened.is_denied("alice", jti="", group_id="")


# ── File deletion (14.11) ────────────────────────────────────────────────────

async def test_deleting_a_file_removes_it_from_disk_and_index(tmp_path):
    state = _state(tmp_path)
    ctx = state["groups_ctx"]["g" * 32]
    target = ctx["roots"].roots[0].path / "gone.txt"
    target.write_text("x")
    from meshbay_common.protocol import IndexEntry
    ctx["index"].add_entry(IndexEntry(id="a" * 64, name="gone.txt", path="shared",
                                      size=1, type="other", added_at=0))

    out = await ops.delete_file(state, "g" * 32, "a" * 64)

    assert out["status"] == "deleted"
    assert not target.exists()
    assert ctx["index"].get_entry("a" * 64) is None


async def test_deleting_from_an_unavailable_root_is_refused(tmp_path):
    """
    A frozen root's files are still listed. Deleting one would either fail
    obscurely or — worse, once the drive returns — leave the index and the disk
    disagreeing.
    """
    state = _state(tmp_path)
    ctx = state["groups_ctx"]["g" * 32]
    from meshbay_common.protocol import IndexEntry
    ctx["index"].add_entry(IndexEntry(id="a" * 64, name="frozen.txt", path="shared",
                                      size=1, type="other", added_at=0))
    ctx["roots"].roots[0].available = False

    with pytest.raises(ops.OpError, match="frozen, not gone"):
        await ops.delete_file(state, "g" * 32, "a" * 64)

    assert ctx["index"].get_entry("a" * 64) is not None


async def test_deleting_an_unknown_file_says_so(tmp_path):
    state = _state(tmp_path)
    with pytest.raises(ops.OpError, match="No such file"):
        await ops.delete_file(state, "g" * 32, "f" * 64)


async def test_an_unhosted_group_offers_what_it_does_host(tmp_path):
    """A bare "no such group" leaves an operator guessing at a UUID."""
    state = _state(tmp_path)
    with pytest.raises(ops.OpError) as exc:
        await ops.delete_file(state, "z" * 32, "a" * 64)
    assert exc.value.status == 404
    assert exc.value.extra.get("available")


# ── Upload policy (per-root writable) ───────────────────────────────────────

async def test_the_group_wide_upload_switch_is_gone(tmp_path):
    """
    `set_member_upload` was the whole of the old policy, and it is deliberately
    not here any more — RO/RW on the root replaced it. A wrapper kept "for
    compatibility" would be a second way to decide who writes to the operator's
    disk, and two answers to that question is how C1 and C6 both happened.
    """
    assert not hasattr(ops, "set_member_upload")
    from meshbay_node.roster import Roster
    assert not hasattr(Roster, "set_member_upload")
    assert not hasattr(Roster, "member_upload_allowed")


async def test_eject_and_plug_persist_through_the_roster(tmp_path):
    """
    The state has to outlive the process: an operator ejects a drive, unplugs
    it, and restarts the node — and the rescan that follows must not read the
    empty mount point as an erased library.
    """
    from meshbay_node.roster import Roster
    state = _state(tmp_path)
    usb = tmp_path / "USB"
    usb.mkdir()
    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
        [{"path": str(usb), "removable": True, "writable": True}])
    state["config"] = SimpleNamespace(
        groups=[SimpleNamespace(id="g" * 32, roots=[])])
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    state["roster"] = roster
    state["node_user_id"] = "operator"
    try:
        out = await ops.eject_root(state, "g" * 32, "USB")
        assert out["status"] == "ejected"
        assert await roster.ejected_roots("g" * 32) == {"usb"}
        assert out["roots"][0]["ejected"] is True
        assert out["roots"][0]["available"] is False

        out = await ops.plug_root(state, "g" * 32, "USB")
        assert out["status"] == "plugged"
        assert await roster.ejected_roots("g" * 32) == set()
    finally:
        await roster.close()


async def test_a_root_that_is_not_removable_cannot_be_ejected(tmp_path):
    """
    Eject means "I am about to unplug this". On a directory that is not on a
    removable device it would hide a library with no way for the safety net to
    notice anything happened, and nothing to plug back in.
    """
    from meshbay_node.roster import Roster
    state = _state(tmp_path)
    fixed = tmp_path / "Fixed"
    fixed.mkdir()
    state["groups_ctx"]["g" * 32]["roots"] = RootSet.build(
        [{"path": str(fixed), "writable": True}])
    state["config"] = SimpleNamespace(
        groups=[SimpleNamespace(id="g" * 32, roots=[])])
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    state["roster"] = roster
    try:
        with pytest.raises(ops.OpError, match="removable"):
            await ops.eject_root(state, "g" * 32, "Fixed")
    finally:
        await roster.close()


async def test_plugging_a_drive_that_is_not_there_is_refused(tmp_path):
    """
    Clearing the flag while the device is still absent would restart the
    watchdog on a missing path and hand the next reconcile an empty directory —
    the deletion storm the eject was there to prevent, produced by the recovery.
    """
    from meshbay_node.roster import Roster
    state = _state(tmp_path)
    usb = tmp_path / "USB"
    usb.mkdir()
    roots = RootSet.build([{"path": str(usb), "removable": True}])
    roots.roots[0].ejected = True
    roots.roots[0].available = False
    state["groups_ctx"]["g" * 32]["roots"] = roots
    state["config"] = SimpleNamespace(
        groups=[SimpleNamespace(id="g" * 32, roots=[])])
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    state["roster"] = roster
    usb.rmdir()
    try:
        with pytest.raises(ops.OpError, match="device connected"):
            await ops.plug_root(state, "g" * 32, "USB")
        assert roots.roots[0].ejected is True
    finally:
        await roster.close()


# ── Reload ──────────────────────────────────────────────────────────────────

async def test_reload_config_calls_reload_fn(tmp_path):
    state = _state(tmp_path)
    called = []
    async def fake_reload():
        called.append(True)
    state["reload_fn"] = fake_reload

    out = await ops.reload_config(state)

    assert out["status"] == "reloaded"
    assert called


async def test_reload_config_without_fn_is_refused(tmp_path):
    state = _state(tmp_path)
    with pytest.raises(ops.OpError, match="Reload not available"):
        await ops.reload_config(state)


async def test_start_reload_returns_before_reload_fn_finishes(tmp_path):
    """The loopback route uses this one: a brand-new group's initial scan
    can take minutes, and the Electron bridge caps every loopback call at
    30s (main.js node:call) — start_reload must not block on it."""
    state = _state(tmp_path)
    release = asyncio.Event()
    called = []

    async def slow_reload():
        await release.wait()
        called.append(True)
    state["reload_fn"] = slow_reload

    out = await asyncio.wait_for(ops.start_reload(state), timeout=1.0)

    assert out["status"] == "reloading"
    assert not called, "start_reload must return before reload_fn finishes"

    release.set()
    await asyncio.sleep(0)  # let the still-running reload_fn task complete
    assert called, "reload_fn must still actually run, just not be waited on"


async def test_start_reload_without_fn_is_refused(tmp_path):
    state = _state(tmp_path)
    with pytest.raises(ops.OpError, match="Reload not available"):
        await ops.start_reload(state)


# ── node.toml editing survives Windows (LF, comments, backslash paths) ────────

def test_update_node_toml_forces_lf_and_keeps_standalone_comments(tmp_path):
    p = tmp_path / "node.toml"
    p.write_bytes("\r\n".join([
        "[hub]", 'url = "http://x"', "",
        "[node]",
        "# how long an invitation lives",
        "invite_ttl_hours = 168",
        "ui_port = 18000",
        "",
        "[[groups]]", 'id = "g1"',
    ]).encode())

    ops._update_node_toml(p, {"invite_ttl_hours": 24, "max_concurrent_streams": 4})

    raw = p.read_bytes()
    assert b"\r\n" not in raw, "must be rewritten LF-only, whatever it was read as"
    text = raw.decode("utf-8")
    assert "# how long an invitation lives" in text
    import tomllib
    node = tomllib.loads(text)["node"]
    assert node["invite_ttl_hours"] == 24
    assert node["max_concurrent_streams"] == 4


def test_a_backslash_path_written_into_node_toml_stays_parseable():
    """
    attach_group and add_root embed a directory into a TOML basic string. A raw
    Windows path there is a parse error, because backslash sequences are escapes
    (`\\U`, `\\a`, ...); the code writes `as_posix()` and pathlib reads `/` back
    on Windows.

    `PureWindowsPath`, not `Path`: on this suite's usual machine `Path` is a
    `PosixPath`, where a backslash is an ordinary filename character and
    `as_posix()` converts nothing — so the test modelled the wrong platform and
    failed everywhere except the one it was written for. Naming the flavour
    explicitly is what makes it the same assertion on all three.
    """
    import tomllib
    bs = chr(92)
    win_dir = f"C:{bs}Users{bs}alice{bs}Media"

    assert tomllib.loads(
        f'path = "{PureWindowsPath(win_dir).as_posix()}"\n'
    )["path"] == "C:/Users/alice/Media"

    with pytest.raises(tomllib.TOMLDecodeError):
        tomllib.loads(f'path = "{win_dir}"\n')   # the bug this guards against


def test_every_path_written_into_node_toml_goes_through_as_posix():
    """
    The half the round trip above cannot see.

    Proving `as_posix()` produces a parseable string says nothing about whether
    the code calls it, and this is a defect no Linux machine can reproduce: the
    config is written, parsed and served correctly here, and fails on the
    operator's Windows box. So the source is read for the shape instead —
    weak evidence, and the only kind available for a platform the suite does
    not run on.
    """
    import re
    source = ops_source()
    # Every f-string interpolation that lands on the right of a TOML `path =`.
    writes = re.findall(r'path\s*=\s*\\?"\{([^}]+)\}', source)
    assert writes, "no TOML path writer found — did the config writer move?"
    for expr in writes:
        assert "as_posix()" in expr, (
            f'node.toml path written as `{expr}` — a Windows path needs '
            f'as_posix(), or the file it lands in will not parse')