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

"""
node_status, root_add, root_remove over MNP.

These test the D5 node management panel's server-side behaviour: the admin
identity check on node_status, the list_groups operation, and the root
add/remove flows through the MNP handlers.
"""

import base64
from dataclasses import asdict
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from conftest import one_root
from meshbay_common.adminop import (
    OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, OP_MEMBER_UNPIN,
    admin_transcript,
)
from meshbay_node.transport.quic_server import Denylist
from meshbay_common.crypto import pk_to_b64
from meshbay_common.join import ROLE_OPERATOR
from meshbay_common.protocol import MNP
from meshbay_node import ops
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import open_roster
from meshbay_node.roots import RootSet
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

GROUP = "g" * 32


@pytest.fixture
async def roster(tmp_path):
    r = await open_roster(tmp_path)
    yield r
    await r.close()


def _keypair():
    sk = Ed25519PrivateKey.generate()
    return sk, pk_to_b64(sk.public_key())


class _FakeBundleStore:
    def __init__(self):
        self.stored = []

    async def store(self, *args):
        self.stored.append(args)


class _FakeHub:
    class _S:
        user_id = "node-user"
    _session = _S()


def _last(session):
    return session.sent[-1] if session.sent else {}


async def _drain(session):
    for coro in session.spawned:
        await coro
    session.spawned.clear()


async def _session(
    tmp_path: Path, roster, *, operator: bool, node_user_id: str = "node-user",
) -> WebRTCPeerSession:
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)

    sk_op, pk_op = _keypair()
    if operator:
        await roster.pin_identity("grenet", "grenet", pk_op, pk_op, "code")
        await roster.set_member("", "grenet", ROLE_OPERATOR, "active", "local-cli")

    group_ctx = {"gek": b"\x01" * 32, "roots": roots, "index": index,
                 "join_policy": "invite"}
    denylist = Denylist()
    reload_called = []
    async def _reload():
        reload_called.append(True)

    state = {
        "groups_ctx": {GROUP: group_ctx},
        "roster": roster,
        "indexes": {GROUP: index},
        "bundle_store": _FakeBundleStore(),
        "pk_x25519_raw": b"\x02" * 32,
        "hub": _FakeHub(),
        "node_user_id": node_user_id,
        "denylist": denylist,
        "reload_fn": _reload,
    }

    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = {
        "roots": roots, "index": index, "sk_node": index.sk_node,
        "roster": roster, "groups": {GROUP: group_ctx},
        "has_admin_authority": operator,
        "daemon_state": state,
        "node_user_id": node_user_id,
    }
    session._group_id = GROUP
    session._user_id = "grenet" if operator else "mallory"
    session._username = session._user_id
    session._pk_user = ""
    session._uploads = {}
    session._admin_ops = {}
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    session.state = state
    session.sk_op = sk_op
    session.denylist = denylist
    session.reload_called = reload_called
    session.spawned = []
    session._spawn = session.spawned.append
    return session


async def _sign_and_exec(session, op: str, subject: str, sk, exec_fn,
                         group_id: str = GROUP):
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge", challenge
    transcript = admin_transcript(
        op=op, node_pk_b64=session._node_pk_b64(), group_id=group_id,
        subject=subject, nonce=base64.b64decode(challenge["nonce"]),
        ts=challenge["ts"])
    pending = session._admin_ops.get(challenge["op_id"]) or {
        "op": op, "subject": subject}
    await exec_fn(pending, transcript, sk.sign(transcript))


# ── _is_node_admin ──────────────────────────────────────────────────────────

async def test_is_node_admin_matches_user_id(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    assert session._is_node_admin()


async def test_is_node_admin_rejects_different_user(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="someone-else")
    assert not session._is_node_admin()


async def test_is_node_admin_rejects_missing_node_user_id(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    del session._ctx["node_user_id"]
    assert not session._is_node_admin()


# ── node_status ─────────────────────────────────────────────────────────────

async def test_node_status_returns_groups_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._spawn(session._do_node_status({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == MNP.NODE_STATUS_ACK
    assert len(msg["groups"]) == 1
    assert msg["groups"][0]["id"] == GROUP


async def test_node_status_refused_for_non_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False,
                             node_user_id="grenet")
    session._spawn(session._do_node_status({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"
    assert "operator" in msg["detail"].lower()


async def test_node_status_refused_when_user_is_operator_but_ids_mismatch(
    tmp_path, roster,
):
    """A paired operator who is not the node owner cannot see node_status."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="someone-else")
    session._spawn(session._do_node_status({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"


async def test_node_status_catches_send_failure(tmp_path, roster):
    """If _send itself throws (e.g. msgpack encoding fails), the error must
    not silently vanish — it used to, because _send was outside the try block."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    original_send = session._send
    sent = []
    call_count = [0]

    def _exploding_send(msg):
        call_count[0] += 1
        if msg.get("type") == "node_status_ack":
            raise TypeError("msgpack cannot encode this")
        sent.append(msg)

    session._send = _exploding_send
    session._spawn(session._do_node_status({}))
    await _drain(session)
    # The try/except around _send should catch the error and send an error reply
    assert any(m.get("type") == "error" for m in sent)


# ── ops.list_groups ─────────────────────────────────────────────────────────

async def test_list_groups_returns_group_metadata(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
        "peers": {"p1": {"group_id": GROUP}, "p2": {"group_id": GROUP},
                  "p3": {"group_id": "other"}},
        "config": None,
    }
    result = await ops.list_groups(state)
    groups = result["groups"]
    assert len(groups) == 1
    g = groups[0]
    assert g["id"] == GROUP
    assert g["has_gek"] is True
    assert g["peers"] == 2
    assert isinstance(g["roots"], list)


async def test_list_groups_empty(tmp_path):
    state = {"groups_ctx": {}, "peers": {}, "config": None}
    result = await ops.list_groups(state)
    assert result["groups"] == []


# ── ops.add_root ────────────────────────────────────────────────────────────

async def test_add_root_creates_directory_and_returns_info(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()
    new_dir = tmp_path / "new_root"

    from meshbay_node.config import NodeConfig, GroupConfig, RootSpec
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
    ])

    conf = tmp_path / "node.toml"
    conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
                    f'  [[groups.roots]]\n  path = "{shared}"\n')
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.add_root(state, GROUP, str(new_dir))
    assert result["status"] == "added"
    assert new_dir.is_dir()
    assert len(result["roots"]) == 2


async def test_add_root_rejects_unknown_group(tmp_path):
    from meshbay_node.config import NodeConfig
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = []
    state = {"config": node_cfg, "groups_ctx": {}}
    with pytest.raises(ops.OpError, match="not configured"):
        await ops.add_root(state, "nonexistent", "/tmp/nope")


# ── ops.remove_root ─────────────────────────────────────────────────────────

async def test_remove_root_requires_at_least_one_remaining(tmp_path):
    shared = tmp_path / "shared"
    shared.mkdir()

    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(shared), name="shared", kind="generic", writable=True),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
                    f'  [[groups.roots]]\n  path = "{shared}"\n')
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    roots = one_root(shared)
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }
    with pytest.raises(ops.OpError, match="only root"):
        await ops.remove_root(state, GROUP, "shared")


async def test_removing_a_writable_root_is_allowed(tmp_path):
    """
    It used to be refused: with one designated upload root, removing it left
    the group with nowhere to put an upload and no way to say so. Several roots
    can be writable now, and a group with none is a valid read-only group — so
    the refusal would be protecting a state that is no longer special.
    """
    d1 = tmp_path / "incoming"
    d2 = tmp_path / "shared"
    d1.mkdir()
    d2.mkdir()

    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="incoming", kind="generic", writable=True),
        RootSpec(path=str(d2), name="shared", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "incoming"\n  writable = true\n\n'
        f'  [[groups.roots]]\n  path = "{d2}"\n  name = "shared"\n')
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }
    result = await ops.remove_root(state, GROUP, "incoming")
    assert result["status"] == "removed"
    assert [r["name"] for r in result["roots"]] == ["shared"]
    assert conf.read_text().count("[[groups.roots]]") == 1


async def test_update_root_rewrites_the_flags_in_node_toml(tmp_path):
    """
    The flags live in the operator's config file, so they survive a restart —
    and the file is hand-written and full of comments, so the change is a line
    edit rather than a round trip through a TOML writer that would discard
    every one of them.
    """
    d1 = tmp_path / "media"
    d1.mkdir()

    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="media", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n'
        f'  # the operator explained this one to themselves\n'
        f'  path = "{d1}"\n  name = "media"\n')
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.update_root(state, GROUP, "media",
                                   writable=True, removable=True)
    assert result["status"] == "updated"
    text = conf.read_text()
    assert "writable = true" in text
    assert "removable = true" in text
    assert "the operator explained this one to themselves" in text, (
        "the config file was rewritten instead of edited")

    # And the live root set agrees immediately, without waiting for a reload:
    # the loopback API reads it, and an operator who toggles a switch and sees
    # it snap back assumes the change did not take.
    assert roots.roots[0].writable is True
    assert roots.roots[0].removable is True

    # A second call that changes nothing must not append a duplicate line.
    await ops.update_root(state, GROUP, "media", writable=True, removable=True)
    assert conf.read_text().count("writable =") == 1


async def test_update_root_replaces_a_legacy_upload_line(tmp_path):
    """
    A config written before the refactor says `upload = true`. Leaving it in
    place next to a new `writable` line would give the file two answers, and
    `RootSet.build` prefers `writable` — so the stale one would sit there
    contradicting the running node for as long as anyone read it.
    """
    d1 = tmp_path / "media"
    d1.mkdir()

    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="media", kind="generic", writable=True),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "media"\n'
        f'  upload = true\n')
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    await ops.update_root(state, GROUP, "media", writable=False)
    text = conf.read_text()
    assert "upload = true" not in text
    assert "writable = false" in text


async def test_remove_root_succeeds_with_two_roots(tmp_path):
    d1 = tmp_path / "dir1"
    d2 = tmp_path / "dir2"
    d1.mkdir()
    d2.mkdir()

    from meshbay_node.config import GroupConfig, RootSpec, NodeConfig
    cfg = GroupConfig(id=GROUP, name="test", roots=[
        RootSpec(path=str(d1), name="dir1", kind="generic", writable=True),
        RootSpec(path=str(d2), name="dir2", kind="generic", writable=False),
    ])
    node_cfg = NodeConfig.__new__(NodeConfig)
    node_cfg.groups = [cfg]

    conf = tmp_path / "node.toml"
    conf.write_text(
        f'[[groups]]\nid = "{GROUP}"\nname = "test"\n\n'
        f'  [[groups.roots]]\n  path = "{d1}"\n  name = "dir1"\n\n'
        f'  [[groups.roots]]\n  path = "{d2}"\n  name = "dir2"\n')
    roots = RootSet.build([asdict(r) for r in cfg.roots])
    index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
    state = {
        "config": node_cfg,
        "config_path": str(conf),
        "groups_ctx": {GROUP: {"index": index, "roots": roots, "gek": b"\x01" * 32}},
    }

    result = await ops.remove_root(state, GROUP, "dir2")
    assert result["status"] == "removed"
    assert len(cfg.roots) == 1
    assert cfg.roots[0].name == "dir1"


# ── root_add MNP handler ───────────────────────────────────────────────────

async def test_root_add_issues_challenge(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_add({"group_id": GROUP, "path": "/tmp/test"})
    msg = _last(session)
    assert msg["type"] == "admin_challenge"


async def test_root_add_refuses_without_authority(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False)
    session._do_root_add({"group_id": GROUP, "path": "/tmp/test"})
    msg = _last(session)
    assert msg["type"] == "error"
    assert "authorized" in msg["detail"].lower()


async def test_root_add_refuses_missing_fields(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_add({"group_id": GROUP})
    assert _last(session)["type"] == "error"
    assert "Missing" in _last(session)["detail"]


# ── root_remove MNP handler ────────────────────────────────────────────────

async def test_root_remove_issues_challenge(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_remove({"group_id": GROUP, "root_name": "shared"})
    msg = _last(session)
    assert msg["type"] == "admin_challenge"


async def test_root_remove_refuses_without_authority(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False)
    session._do_root_remove({"group_id": GROUP, "root_name": "shared"})
    msg = _last(session)
    assert msg["type"] == "error"


async def test_root_remove_refuses_missing_fields(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_root_remove({"group_id": GROUP})
    assert _last(session)["type"] == "error"
    assert "Missing" in _last(session)["detail"]


# ── roster_read MNP handler ──────────────────────────────────────────────

async def test_roster_read_returns_members_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._spawn(session._do_roster_read({"group_id": GROUP}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "roster_read_ack"
    assert "members" in msg
    assert "identities" in msg


async def test_roster_read_refused_for_non_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False,
                             node_user_id="grenet")
    session._spawn(session._do_roster_read({"group_id": GROUP}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"


async def test_roster_read_filters_ghost_members(tmp_path, roster):
    """Members whose identity was deleted (revoked then unpinned) are filtered
    out by read_roster — the LEFT JOIN returns them with pk_ed25519 = NULL but
    they should never reach the UI."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    sk2, pk2 = _keypair()
    await roster.pin_identity("ghost", "ghost", pk2, pk2, "code")
    await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli")
    await roster._db.execute("DELETE FROM identities WHERE user_id = 'ghost'")
    await roster._db.commit()
    # Also add a real member so the roster isn't empty
    sk3, pk3 = _keypair()
    await roster.pin_identity("real", "real", pk3, pk3, "code")
    await roster.set_member(GROUP, "real", "member", "active", "local-cli")

    session._spawn(session._do_roster_read({"group_id": GROUP}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "roster_read_ack"
    ghost = [m for m in msg["members"] if m["user_id"] == "ghost"]
    assert len(ghost) == 0, "ghost members must be filtered out"
    real = [m for m in msg["members"] if m["user_id"] == "real"]
    assert len(real) == 1


async def test_unpin_fails_for_ghost_member(tmp_path, roster):
    """Unpinning a member with no identity gives a clear error, not a crash."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    await roster.set_member(GROUP, "ghost", "member", "revoked", "local-cli")
    # ghost has no identity row
    session._do_member_unpin({"user_id": "ghost"})
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge"

    await _sign_and_exec(session, OP_MEMBER_UNPIN, "ghost",
                         session.sk_op, session._admin_exec_member_unpin)
    msg = _last(session)
    assert msg["type"] == "error"
    assert "No such pinned identity" in msg["detail"]


async def test_unpin_succeeds_for_real_identity(tmp_path, roster):
    """Full unpin flow: challenge → sign → exec → identity deleted."""
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    sk2, pk2 = _keypair()
    await roster.pin_identity("target", "target", pk2, pk2, "code")
    await roster.set_member(GROUP, "target", "member", "active", "local-cli")

    session._do_member_unpin({"user_id": "target"})
    challenge = _last(session)
    assert challenge["type"] == "admin_challenge"

    await _sign_and_exec(session, OP_MEMBER_UNPIN, "target",
                         session.sk_op, session._admin_exec_member_unpin)
    msg = _last(session)
    assert msg["type"] == "member_unpin_ack"
    assert msg["user_id"] == "target"

    idents = await roster.list_identities()
    assert not any(i["user_id"] == "target" for i in idents)


# ── denylist_read MNP handler ────────────────────────────────────────────

async def test_denylist_read_returns_entries_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("bad-user")
    session._spawn(session._do_denylist_read({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "denylist_read_ack"
    assert msg["count"] == 1
    assert "bad-user" in msg["users"]


async def test_denylist_read_refused_for_non_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False,
                             node_user_id="grenet")
    session._spawn(session._do_denylist_read({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"


# ── denylist_clear MNP handler ───────────────────────────────────────────

async def test_denylist_clear_removes_entry_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("bad-user")
    session.denylist.deny_user("other-user")
    session._spawn(session._do_denylist_clear({"subject": "bad-user"}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "denylist_clear_ack"
    assert msg["removed"] == 1
    assert "bad-user" not in session.denylist.entries()["users"]
    assert "other-user" in session.denylist.entries()["users"]


async def test_denylist_clear_all_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session.denylist.deny_user("a")
    session.denylist.deny_user("b")
    session.denylist.deny_jti("j")
    session._spawn(session._do_denylist_clear({"subject": ""}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "denylist_clear_ack"
    assert msg["removed"] == 3


async def test_denylist_clear_refused_for_non_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False,
                             node_user_id="grenet")
    session._spawn(session._do_denylist_clear({"subject": "bad-user"}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"


# ── group_attach MNP handler ────────────────────────────────────────────

async def test_group_attach_issues_challenge(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"})
    msg = _last(session)
    assert msg["type"] == "admin_challenge"


async def test_group_attach_refused_without_authority(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False)
    session._do_group_attach({"name": "test-group", "shared_dir": "/tmp/share"})
    msg = _last(session)
    assert msg["type"] == "error"
    assert "authorized" in msg["detail"].lower()


async def test_group_attach_refuses_missing_fields(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._do_group_attach({"name": "test-group"})
    msg = _last(session)
    assert msg["type"] == "error"
    assert "Missing" in msg["detail"]


# ── node_reload MNP handler ─────────────────────────────────────────────

async def test_node_reload_runs_for_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=True,
                             node_user_id="grenet")
    session._spawn(session._do_node_reload({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "node_reload_ack"
    assert msg["status"] == "reloaded"
    assert len(session.reload_called) == 1


async def test_node_reload_refused_for_non_admin(tmp_path, roster):
    session = await _session(tmp_path, roster, operator=False,
                             node_user_id="grenet")
    session._spawn(session._do_node_reload({}))
    await _drain(session)
    msg = _last(session)
    assert msg["type"] == "error"