1 files changed, 176 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_member_upload_policy.py b/packages/meshbay-node/tests/test_member_upload_policy.py
new file mode 100644
index 0000000..b1dc0cb
--- /dev/null
+++ b/packages/meshbay-node/tests/test_member_upload_policy.py
@@ -0,0 +1,176 @@
+"""
+The operator can close uploading to everyone but themselves.
+
+The point of these tests is the difference between a hidden button and a closed
+door. The interface stops offering the control, which is a courtesy to the
+people who are not trying; **the node refuses the upload**, which is the part
+that holds against someone who is. A member who kept an old tab open, or who
+speaks MNP directly, gets the same answer as everyone else.
+
+Two further things are worth holding:
+
+* the setting is changed by a **signed** operator instruction. A node that took
+  it from an unsigned message would let any member turn it back on, and the
+  control would be a suggestion;
+* it is stored on the **node**, not the hub. A hub that could decide who may
+  write to the operator's disk is a hub with authority over the node, which is
+  the thing this whole design is arranged to avoid.
+"""
+
+import base64
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.adminop import OP_MEMBER_UPLOAD
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+from conftest import one_root
+
+pytestmark = pytest.mark.asyncio
+
+
+def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
+             operator: str | None = None) -> WebRTCPeerSession:
+    shared_root = tmp_path / "shared"
+    shared_root.mkdir(exist_ok=True)
+    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
+    ctx = {
+        "roots": one_root(shared_root),
+        "index": index,
+        "sk_node": index.sk_node,
+        "member_upload": member_upload,
+        "node_user_id": operator,
+    }
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = ctx
+    session._group_id = None
+    session._user_id = user_id
+    session._pk_user = ""
+    session._uploads = {}
+    session.sent = []
+    session._send = session.sent.append
+    session._audit = lambda *a, **k: None
+    return session
+
+
+def _upload(session, filename="clip.mp4", body=b"bytes"):
+    session._do_file_upload({
+        "filename": filename, "chunk_index": 0, "total_chunks": 1,
+        "data": base64.b64encode(body).decode(),
+    })
+
+
+def _uploads_dir(session) -> Path:
+    return session._ctx["roots"].upload_root.path / "uploads"
+
+
+# ── The door, not the button ────────────────────────────────────────────────
+
+async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=False,
+                       operator="the-operator")
+    _upload(session)
+
+    assert not (_uploads_dir(session) / "clip.mp4").exists(), (
+        "the file was written even though uploading is off — the setting is "
+        "decorative and the hidden button was the whole control")
+    refusal = [m for m in session.sent if m.get("type") == "error"]
+    assert refusal and refusal[0].get("code") == "member_upload_off"
+
+
+async def test_the_operator_can_still_upload(tmp_path):
+    """Otherwise turning it off locks the operator out of their own node, and
+    the only way back is a config file and a restart."""
+    session = _session(tmp_path, "the-operator", member_upload=False,
+                       operator="the-operator")
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_members_upload_normally_when_it_is_on(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
+    """An existing node's context has no such key. The absence must read as
+    "allowed", or upgrading the node silently closes every group."""
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    del session._ctx["member_upload"]
+    _upload(session)
+
+    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"
+
+
+# ── Who may change it ───────────────────────────────────────────────────────
+
+async def test_changing_it_needs_a_signature(tmp_path):
+    """
+    The request only ever produces a challenge. Nothing is applied until a
+    signature over the transcript verifies — the same path as removing a member.
+    """
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_member_upload({"allowed": False})
+
+    assert issued == [(OP_MEMBER_UPLOAD, "off")]
+    assert session._ctx["member_upload"] is True, "applied before it was signed"
+
+
+async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
+    """The operator is shown the subject before signing. "member_upload" tells
+    them nothing; "off" tells them what they are about to do."""
+    session = _session(tmp_path, "op", member_upload=False, operator="op")
+    session._has_admin_authority = lambda: True
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+
+    session._do_member_upload({"allowed": True})
+
+    assert issued == [(OP_MEMBER_UPLOAD, "on")]
+
+
+async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
+    session = _session(tmp_path, "member-1", member_upload=True,
+                       operator="the-operator")
+    session._has_admin_authority = lambda: False
+
+    session._do_member_upload({"allowed": False})
+
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.member_upload_allowed("g1") is True, (
+            "absent must mean allowed, or an upgrade closes every group")
+        await roster.set_member_upload("g1", False, set_by="op")
+        assert await roster.member_upload_allowed("g1") is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.member_upload_allowed("g1") is False
+        assert await reopened.member_upload_allowed("g2") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()