"""Integration tests for group self-service: create, join, members.""" from datetime import UTC import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey from meshbay_common.crypto import pk_to_b64 def _gen_user_keys(): sk_ed = Ed25519PrivateKey.generate() sk_x = X25519PrivateKey.generate() return pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()) async def _register(client, username, email="test@x.com", password="testpass99"): pk_ed, pk_x = _gen_user_keys() r = await client.post("/v1/users/register", json={ "username": username, "email": email, "password": password, "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x, }) assert r.status_code == 201 return r.json()["user_id"] async def _login(client, username, password="testpass99"): r = await client.post("/v1/users/login", json={ "username": username, "password": password, }) assert r.status_code == 200 return r.json()["access_token"] async def _create_group(client, token, name="test-group", visibility="public", join_policy="open"): r = await client.post("/v1/groups", json={ "name": name, "visibility": visibility, "join_policy": join_policy, }, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 201 return r.json()["group_id"] async def _mark_hosted(db_session, *group_ids): """Pretend a node announced these groups, as /v1/nodes/ws would.""" from datetime import datetime from meshbay_hub.db.models import Group for gid in group_ids: (await db_session.get(Group, gid)).hosted_at = datetime.now(UTC) await db_session.commit() @pytest.mark.asyncio async def test_create_group(client): await _register(client, "alice_test", email="a@x.com") token = await _login(client, "alice_test") r = await client.post("/v1/groups", json={ "name": "my-group", "visibility": "public", "join_policy": "open", }, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 201 data = r.json() assert data["name"] == "my-group" assert "group_id" in data @pytest.mark.asyncio async def test_join_open_group(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") gid = await _create_group(client, alice_token, "open-group") await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") r = await client.post(f"/v1/groups/{gid}/join", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 200 assert r.json()["status"] == "joined" @pytest.mark.asyncio async def test_join_invite_group_rejected(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") # Private: invite-only is refused on a public group now, since a group # everyone can find and nobody can enter is a dead end. What is under test # here — /join refusing a group that is not open — is unchanged. gid = await _create_group(client, alice_token, "invite-group", visibility="private", join_policy="invite") await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") r = await client.post(f"/v1/groups/{gid}/join", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 403 @pytest.mark.asyncio async def test_join_already_member(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") gid = await _create_group(client, alice_token, "dup-group") r = await client.post(f"/v1/groups/{gid}/join", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 409 @pytest.mark.asyncio async def test_group_members(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") gid = await _create_group(client, alice_token, "team-group") await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") await client.post(f"/v1/groups/{gid}/join", headers={"Authorization": f"Bearer {bob_token}"}) r = await client.get(f"/v1/groups/{gid}/members", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 200 data = r.json() usernames = [m["username"] for m in data["members"]] assert "alice_test" in usernames assert "bob_test" in usernames assert data["admin_id"] is not None @pytest.mark.asyncio async def test_group_members_non_member_denied(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") gid = await _create_group(client, alice_token, "private-group", visibility="private", join_policy="invite") await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") r = await client.get(f"/v1/groups/{gid}/members", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 403 @pytest.mark.asyncio async def test_group_search(client, db_session): await _register(client, "alice_test", email="a@x.com") token = await _login(client, "alice_test") a = await _create_group(client, token, "alpha-team") b = await _create_group(client, token, "beta-team") # The directory shows groups a node has announced. Marked here so this test # exercises the search filter rather than the hosting one. await _mark_hosted(db_session, a, b) r = await client.get("/v1/groups?q=alpha") assert r.status_code == 200 names = [g["name"] for g in r.json()["groups"]] assert "alpha-team" in names assert "beta-team" not in names @pytest.mark.asyncio async def test_join_triggers_notification(client): await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") gid = await _create_group(client, alice_token, "notif-group") await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") await client.post(f"/v1/groups/{gid}/join", headers={"Authorization": f"Bearer {bob_token}"}) r = await client.get(f"/v1/groups/{gid}/members", headers={"Authorization": f"Bearer {alice_token}"}) assert len(r.json()["members"]) == 2 # ── Listed and open are one question ──────────────────────────────────────── @pytest.mark.asyncio async def test_a_private_group_cannot_be_open_to_everyone(client): """ A group anyone may join that nobody can find is a listing with the listing removed: it is absent from the directory, and joining goes through the node rather than a link, so nothing can reach it. It was accepted until now, and the create form offered it. """ await _register(client, "pat_test", email="pat@x.com") token = await _login(client, "pat_test") resp = await client.post("/v1/groups", json={ "name": "nowhere", "visibility": "private", "join_policy": "open", }, headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 422 assert "invite-only" in resp.json()["detail"] @pytest.mark.asyncio async def test_a_public_group_cannot_be_invite_only(client): """The other half, which was already refused — kept so that removing one check does not quietly remove both.""" await _register(client, "sam_test", email="sam@x.com") token = await _login(client, "sam_test") resp = await client.post("/v1/groups", json={ "name": "deadend", "visibility": "public", "join_policy": "invite", }, headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 422 @pytest.mark.asyncio async def test_the_two_combinations_that_mean_something_are_accepted(client): await _register(client, "robin_test", email="robin@x.com") token = await _login(client, "robin_test") for name, visibility, policy in (("closed", "private", "invite"), ("open-house", "public", "open")): resp = await client.post("/v1/groups", json={ "name": name, "visibility": visibility, "join_policy": policy, }, headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 201, f"{visibility}+{policy}: {resp.text}"