1 files changed, 194 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_search_listed.py b/packages/meshbay-node/tests/test_search_listed.py
new file mode 100644
index 0000000..412d743
--- /dev/null
+++ b/packages/meshbay-node/tests/test_search_listed.py
@@ -0,0 +1,194 @@
+"""
+Whether a group's files are listed in members' cross-group Search.
+
+A presentation preference, stated as such everywhere (design §9.11): the node
+serves the same index to Search and to the group page, and cannot tell them
+apart. What is held here is the node half — stored on the node, absent means
+listed, changed only by a signed operator instruction, kept in step with the
+live context, and broadcast so every connected member's page moves with it.
+"""
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from meshbay_common.adminop import OP_SEARCH_LISTED
+from meshbay_common.protocol import MNP
+from meshbay_node import ops
+from meshbay_node.indexer.group_index import GroupIndex
+from meshbay_node.roster import Roster
+from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+GROUP = "g" * 32
+
+
+def _session(user_id: str = "op") -> WebRTCPeerSession:
+    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+    session._ctx = {"node_user_id": "op"}
+    session._group_id = GROUP
+    session._user_id = user_id
+    session._pk_user = ""
+    session.sent = []
+    session._send = session.sent.append
+    session.audited = []
+    session._audit = lambda *a, **k: session.audited.append(a)
+    return session
+
+
+def _challenges(session: WebRTCPeerSession) -> list:
+    issued = []
+    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))
+    return issued
+
+
+# ── Where it is stored ──────────────────────────────────────────────────────
+
+async def test_absent_means_listed_and_the_setting_survives_a_restart(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        assert await roster.search_listed(GROUP) is True, (
+            "absent must mean listed — an upgrade must not empty anyone's Search")
+        await roster.set_search_listed(GROUP, False, set_by="op")
+        assert await roster.search_listed(GROUP) is False
+    finally:
+        await roster.close()
+
+    reopened = Roster(db_path=tmp_path / "roster.db")
+    await reopened.open()
+    try:
+        assert await reopened.search_listed(GROUP) is False
+        assert await reopened.search_listed("other") is True, (
+            "one group's setting must not answer for another")
+    finally:
+        await reopened.close()
+
+
+async def test_the_op_updates_the_live_context(tmp_path):
+    """The handshake ack reads the context, so the op keeps the two in step."""
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate())
+        state = {"roster": roster, "node_user_id": "op",
+                 "groups_ctx": {GROUP: {"index": index}}}
+        out = await ops.set_search_listed(state, GROUP, False)
+        assert out == {"listed": False, "group_id": GROUP}
+        assert state["groups_ctx"][GROUP]["search_listed"] is False
+        assert await roster.search_listed(GROUP) is False
+    finally:
+        await roster.close()
+
+
+async def test_the_op_refuses_a_group_this_node_does_not_host(tmp_path):
+    roster = Roster(db_path=tmp_path / "roster.db")
+    await roster.open()
+    try:
+        state = {"roster": roster, "groups_ctx": {}}
+        with pytest.raises(ops.OpError):
+            await ops.set_search_listed(state, GROUP, False)
+    finally:
+        await roster.close()
+
+
+# ── Signed, and refused before a challenge otherwise ────────────────────────
+
+@pytest.mark.parametrize("payload", [{}, {"listed": "no"}, {"listed": 0}])
+async def test_a_malformed_request_is_refused(payload):
+    session = _session()
+    session._has_admin_authority = lambda: True
+    issued = _challenges(session)
+
+    session._do_search_listed(payload)
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+async def test_a_member_without_authority_is_refused():
+    session = _session("member-1")
+    session._has_admin_authority = lambda: False
+    issued = _challenges(session)
+
+    session._do_search_listed({"listed": False})
+
+    assert not issued
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+@pytest.mark.parametrize("listed,subject", [(True, "on"), (False, "off")])
+async def test_the_subject_names_the_outcome(listed, subject):
+    session = _session()
+    session._has_admin_authority = lambda: True
+    issued = _challenges(session)
+
+    session._do_search_listed({"listed": listed})
+
+    assert issued == [(OP_SEARCH_LISTED, subject)]
+
+
+async def test_a_bad_signature_changes_nothing():
+    session = _session()
+    ran = []
+
+    async def refuse(transcript, sig):
+        return False
+
+    async def run_op(fn, *args):
+        ran.append(args)
+
+    session._verify_admin_sig = refuse
+    session._run_op = run_op
+    session._broadcast_to_group = lambda notice: ran.append(notice)
+
+    await session._admin_exec_search_listed(
+        {"subject": "off"}, b"transcript", b"sig")
+
+    assert not ran
+    assert [m for m in session.sent if m.get("type") == "error"]
+
+
+def test_the_handshake_ack_carries_it_sealed_and_absent_reads_as_listed():
+    """
+    Read off the real builder rather than a hand-made config: the ack's
+    configuration is the dict `_complete_handshake` seals, and a client that
+    connects after the operator changed the setting learns it from there.
+    """
+    import ast
+    import inspect
+    import textwrap
+
+    source = textwrap.dedent(inspect.getsource(WebRTCPeerSession._complete_handshake))
+    tree = ast.parse(source)
+    config = next(
+        n.value for n in ast.walk(tree)
+        if isinstance(n, ast.Assign)
+        and any(isinstance(t, ast.Name) and t.id == "config" for t in n.targets))
+    assert isinstance(config, ast.Dict)
+    values = {k.value: v for k, v in zip(config.keys, config.values)
+              if isinstance(k, ast.Constant)}
+    assert "search_listed" in values, "the sealed ack no longer carries search_listed"
+    expr = ast.unparse(values["search_listed"])
+    assert "'search_listed', True" in expr, (
+        f"search_listed is built as {expr} — absent must read as listed")
+
+
+async def test_a_signed_change_is_applied_and_broadcast():
+    session = _session()
+    applied, broadcast = [], []
+
+    async def accept(transcript, sig):
+        return True
+
+    async def run_op(fn, *args):
+        applied.append((fn, args))
+
+    session._verify_admin_sig = accept
+    session._run_op = run_op
+    session._broadcast_to_group = broadcast.append
+
+    await session._admin_exec_search_listed(
+        {"subject": "off"}, b"transcript", b"sig")
+
+    assert applied == [(ops.set_search_listed, (GROUP, False))]
+    assert len(broadcast) == 1
+    assert broadcast[0]["type"] == MNP.SEARCH_LISTED_ACK
+    assert broadcast[0]["listed"] is False