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