""" 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 import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from conftest import one_root from meshbay_node import ops from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.transport.quic_server import Denylist 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. """ public = [(n, f) for n, f in vars(ops).items() if inspect.iscoroutinefunction(f) and not n.startswith("_")] assert public, "no operations found — 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", "set_member_upload", "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 = inspect.getsource(ops) 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 (set_member_upload) ─────────────────────────────────────── async def test_set_member_upload_toggles_and_persists(tmp_path): from meshbay_node.roster import Roster state = _state(tmp_path) roster = Roster(db_path=tmp_path / "roster.db") await roster.open() state["roster"] = roster state["node_user_id"] = "operator" out = await ops.set_member_upload(state, "g" * 32, True) assert out["allowed"] is True assert state["groups_ctx"]["g" * 32]["member_upload"] is True out2 = await ops.set_member_upload(state, "g" * 32, False) assert out2["allowed"] is False assert state["groups_ctx"]["g" * 32]["member_upload"] is False # ── 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)