""" Integration tests for the admin/moderation panel API. """ 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 from meshbay_hub.api.deps import set_admin_usernames 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 _setup_admin(client, admin_name="admin"): user_id = await _register(client, admin_name, email=f"{admin_name}@x.com") set_admin_usernames([admin_name]) token = await _login(client, admin_name) return user_id, token # ── require_moderator ──────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_admin_stats_requires_moderator(client): await _register(client, "normie") token = await _login(client, "normie") r = await client.get("/v1/admin/stats", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 403 @pytest.mark.asyncio async def test_admin_stats_allowed_for_config_admin(client): _, token = await _setup_admin(client) r = await client.get("/v1/admin/stats", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert "users" in data assert "groups" in data assert "nodes" in data assert "online_nodes" in data @pytest.mark.asyncio async def test_admin_stats_allowed_for_db_moderator(client): _, admin_token = await _setup_admin(client, "boss") mod_id = await _register(client, "moduser") r = await client.patch(f"/v1/admin/users/{mod_id}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) assert r.status_code == 200 mod_token = await _login(client, "moduser") r = await client.get("/v1/admin/stats", headers={"Authorization": f"Bearer {mod_token}"}) assert r.status_code == 200 # ── User management ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_admin_list_users(client): _, token = await _setup_admin(client) await _register(client, "alice", email="a@x.com") await _register(client, "bob", email="b@x.com") r = await client.get("/v1/admin/users", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["total"] == 3 assert len(data["users"]) == 3 @pytest.mark.asyncio async def test_admin_list_users_search(client): _, token = await _setup_admin(client) await _register(client, "alice", email="a@x.com") await _register(client, "bob", email="b@x.com") r = await client.get("/v1/admin/users?q=ali", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["total"] == 1 assert data["users"][0]["username"] == "alice" @pytest.mark.asyncio async def test_admin_get_user_detail(client): _, token = await _setup_admin(client) uid = await _register(client, "alice", email="alice@example.com") r = await client.get(f"/v1/admin/users/{uid}", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["username"] == "alice" assert data["role"] == "user" assert data["status"] == "active" assert data["group_count"] == 0 assert "email" in data @pytest.mark.asyncio async def test_admin_suspend_unsuspend_user(client): _, token = await _setup_admin(client) uid = await _register(client, "alice", email="a@x.com") alice_token = await _login(client, "alice") r = await client.patch(f"/v1/admin/users/{uid}", json={"status": "suspended"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 assert r.json()["status"] == "suspended" r = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 403 r = await client.patch(f"/v1/admin/users/{uid}", json={"status": "active"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 assert r.json()["status"] == "active" r = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 200 @pytest.mark.asyncio async def test_admin_change_role(client): _, token = await _setup_admin(client) uid = await _register(client, "alice", email="a@x.com") r = await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 assert r.json()["role"] == "moderator" @pytest.mark.asyncio async def test_moderator_cannot_change_roles_or_revoke(client): """A moderator suspends and restores (reversible); it cannot promote anyone or hard-revoke, which would be a path from the moderation role to full instance control.""" _, admin_token = await _setup_admin(client, "boss") mod_id = await _register(client, "moduser") await client.patch(f"/v1/admin/users/{mod_id}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) mod_token = await _login(client, "moduser") mod_h = {"Authorization": f"Bearer {mod_token}"} victim = await _register(client, "victim", email="v@x.com") # No promoting an accomplice. r = await client.patch(f"/v1/admin/users/{victim}", json={"role": "admin"}, headers=mod_h) assert r.status_code == 403 # No hard revocation. r = await client.patch(f"/v1/admin/users/{victim}", json={"status": "revoked"}, headers=mod_h) assert r.status_code == 403 # No touching an admin's account. admin2 = await _register(client, "admin2", email="a2@x.com") await client.patch(f"/v1/admin/users/{admin2}", json={"role": "admin"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.patch(f"/v1/admin/users/{admin2}", json={"status": "suspended"}, headers=mod_h) assert r.status_code == 403 # Suspending a plain user is still fine. r = await client.patch(f"/v1/admin/users/{victim}", json={"status": "suspended"}, headers=mod_h) assert r.status_code == 200 @pytest.mark.asyncio async def test_admin_cannot_modify_self(client): admin_id, token = await _setup_admin(client) r = await client.patch(f"/v1/admin/users/{admin_id}", json={"role": "user"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 400 @pytest.mark.asyncio async def test_admin_invalid_role_rejected(client): _, token = await _setup_admin(client) uid = await _register(client, "alice", email="a@x.com") r = await client.patch(f"/v1/admin/users/{uid}", json={"role": "superuser"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 422 # ── Group management ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_admin_list_groups(client): _, token = await _setup_admin(client) await client.post("/v1/groups", json={ "name": "test-group", "visibility": "public", "join_policy": "open", }, headers={"Authorization": f"Bearer {token}"}) r = await client.get("/v1/admin/groups", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["total"] >= 1 assert data["groups"][0]["name"] == "test-group" assert "member_count" in data["groups"][0] @pytest.mark.asyncio async def test_admin_suspend_group(client): _, token = await _setup_admin(client) r = await client.post("/v1/groups", json={ "name": "bad-group", "visibility": "public", "join_policy": "open", }, headers={"Authorization": f"Bearer {token}"}) gid = r.json()["group_id"] r = await client.patch(f"/v1/admin/groups/{gid}", json={"status": "suspended"}, headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 assert r.json()["status"] == "suspended" # ── Audit logs ─────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_admin_logs(client): _, token = await _setup_admin(client) await _register(client, "alice", email="a@x.com") r = await client.get("/v1/admin/logs", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert len(data["logs"]) > 0 log = data["logs"][0] assert "event" in log assert "timestamp" in log @pytest.mark.asyncio async def test_admin_logs_filter_by_event(client): _, token = await _setup_admin(client) await _register(client, "alice", email="a@x.com") await _login(client, "alice") r = await client.get("/v1/admin/logs?event=login", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 for log in r.json()["logs"]: assert log["event"] == "login" # ── /v1/users/me ───────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_users_me(client): await _register(client, "alice", email="a@x.com") token = await _login(client, "alice") r = await client.get("/v1/users/me", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["username"] == "alice" assert data["role"] == "user" assert data["status"] == "active"