simplex-orchestrate/group_test.py

"""Two-account integration test over the WebSocket interface:

  alice ──invitation──▶ bob          (1-time link connect → contacts)
  alice creates group, invites bob → bob joins → verify membership both sides.

Exercises the real async + SMP-network path (not just local commands).
Run (needs ./bin/simplex-chat):  .venv/bin/python group_test.py
"""

import asyncio
import json
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from supervisor.supervisor import Supervisor  # noqa: E402

events: dict[str, list[dict]] = {"alice": [], "bob": []}


async def on_event(name: str, resp: dict) -> None:
    events.setdefault(name, []).append(resp)


async def wait_event(name: str, pred, timeout: float, label: str) -> dict:
    start = time.time()
    i = 0
    while time.time() - start < timeout:
        while i < len(events[name]):
            e = events[name][i]; i += 1
            if pred(e):
                return e
        await asyncio.sleep(0.3)
    raise TimeoutError(f"[{name}] timed out waiting for {label}")


async def poll(coro_factory, pred, timeout: float, label: str):
    start = time.time()
    while time.time() - start < timeout:
        val = await coro_factory()
        if pred(val):
            return val
        await asyncio.sleep(1.0)
    raise TimeoutError(f"timed out polling for {label}")


def invitation_link(resp: dict) -> str:
    inv = resp.get("connLinkInvitation") or {}
    return inv.get("connShortLink") or inv.get("connFullLink", "")


async def main() -> int:
    sup = Supervisor(data_dir="data", base_port=5500, on_event=on_event)
    ok = False
    try:
        print("→ starting two accounts (alice, bob) …")
        await sup.start_cli("alice")
        await sup.start_cli("bob")
        for n in ("alice", "bob"):
            r = await sup.send(n, "/_start")
            print(f"  {n} /_start -> {r.get('type')}")

        uidA = (await sup.send("alice", "/user"))["user"]["userId"]
        uidB = (await sup.send("bob", "/user"))["user"]["userId"]
        print(f"  alice uid={uidA}  bob uid={uidB}")

        # 1. connect via 1-time invitation
        inv = await sup.send("alice", f"/_connect {uidA}")
        link = invitation_link(inv)
        print(f"→ alice invitation ({inv.get('type')}): {link[:55]}…")
        cb = await sup.send("bob", f"/connect {link}")
        print(f"→ bob /connect -> {cb.get('type')}")

        print("→ waiting for contactConnected (SMP handshake) …")
        await wait_event("alice", lambda e: e.get("type") == "contactConnected", 120, "alice contactConnected")
        print("  ✓ alice: contact connected")
        await wait_event("bob", lambda e: e.get("type") == "contactConnected", 120, "bob contactConnected")
        print("  ✓ bob: contact connected")

        # 2. alice creates a group and invites bob
        contacts = (await sup.send("alice", f"/_contacts {uidA}")).get("contacts", [])
        bob_cid = contacts[0]["contactId"]
        print(f"→ bob is contactId={bob_cid} on alice")
        gr = await sup.send("alice", f'/_group {uidA} ' + json.dumps({"displayName": "testgrp", "fullName": ""}))
        gid = gr["groupInfo"]["groupId"]
        print(f"→ group created gid={gid} ({gr.get('type')})")
        add = await sup.send("alice", f"/_add #{gid} {bob_cid} member")
        print(f"→ alice /_add bob -> {add.get('type')}")

        # 3. bob sees the pending invite, then joins
        async def bob_groups():
            return (await sup.send("bob", f"/_groups {uidB}")).get("groups", [])
        bgroups = await poll(bob_groups, lambda gs: len(gs) > 0, 120, "bob receives group invite")
        bg = bgroups[0]
        print(f"  ✓ bob sees group #{bg['groupId']} status={bg['membership']['memberStatus']}")
        j = await sup.send("bob", f"/_join #{bg['groupId']}")
        print(f"→ bob /_join -> {j.get('type')}")

        # 4. verify membership on both sides
        print("→ waiting for member to connect in group …")
        await wait_event(
            "alice",
            lambda e: e.get("type") in ("memberConnected", "joinedGroupMember", "connectedToGroupMember"),
            120, "alice memberConnected",
        )
        print("  ✓ alice: member connected event")

        async def alice_member_count():
            ag = (await sup.send("alice", f"/_groups {uidA}")).get("groups", [])
            return ag[0]["groupSummary"]["currentMembers"] if ag else 0
        count = await poll(alice_member_count, lambda c: c >= 2, 60, "group reaches 2 members")
        mem = (await sup.send("alice", f"/_members #{gid}"))["group"]["members"]
        print(f"  alice group members: {[(m['localDisplayName'], m['memberStatus']) for m in mem]}")
        print(f"  alice group currentMembers = {count}")
        ok = count >= 2
    finally:
        await sup.stop_all()

    print("\nRESULT:", "PASS — invite/join verified end to end" if ok else "INCOMPLETE")
    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))