summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-hub/tests
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-hub/tests')
-rw-r--r--packages/meshbay-hub/tests/test_admin.py259
1 files changed, 259 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_admin.py b/packages/meshbay-hub/tests/test_admin.py
new file mode 100644
index 0000000..51b233a
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_admin.py
@@ -0,0 +1,259 @@
+"""
+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_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"
generated by cgit v1.2.3 (git 2.51.0) at 2026-09-25 20:13:51 +0000