"""Integration tests for the notification system.""" 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_test"): 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 @pytest.mark.asyncio async def test_notifications_empty(client): await _register(client, "alice_test", email="a@x.com") token = await _login(client, "alice_test") r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {token}"}) assert r.status_code == 200 data = r.json() assert data["notifications"] == [] assert data["unread_count"] == 0 @pytest.mark.asyncio async def test_notification_on_role_change(client): _, admin_token = await _setup_admin(client) uid = await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 200 data = r.json() assert data["unread_count"] == 1 assert data["notifications"][0]["kind"] == "role_change" assert "moderator" in data["notifications"][0]["title"] @pytest.mark.asyncio async def test_notification_on_suspend(client): _, admin_token = await _setup_admin(client) uid = await _register(client, "bob_test", email="b@x.com") bob_token = await _login(client, "bob_test") await client.patch(f"/v1/admin/users/{uid}", json={"status": "suspended"}, headers={"Authorization": f"Bearer {admin_token}"}) await client.patch(f"/v1/admin/users/{uid}", json={"status": "active"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {bob_token}"}) assert r.status_code == 200 kinds = [n["kind"] for n in r.json()["notifications"]] assert "account_suspended" in kinds assert "account_restored" in kinds @pytest.mark.asyncio async def test_dismissing_one_deletes_it(client): _, admin_token = await _setup_admin(client) uid = await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) nid = r.json()["notifications"][0]["id"] # `/read` is the old path and still the one older clients call. It # dismisses, like DELETE — see api/notifications.py. r = await client.post(f"/v1/notifications/{nid}/read", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 200 r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) assert r.json()["unread_count"] == 0 assert r.json()["notifications"] == [], ( "a dismissed notification is deleted, not kept as a row nothing can " "ever show again") @pytest.mark.asyncio async def test_dismissing_all_deletes_them(client): _, admin_token = await _setup_admin(client) uid = await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) await client.patch(f"/v1/admin/users/{uid}", json={"role": "admin"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) assert r.json()["unread_count"] == 2 r = await client.post("/v1/notifications/read-all", headers={"Authorization": f"Bearer {alice_token}"}) assert r.status_code == 200 assert r.json()["removed"] == 2 r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) assert r.json()["unread_count"] == 0 assert r.json()["notifications"] == [], ( "read-all must not be the one route that still hoards rows") @pytest.mark.asyncio async def test_notification_unread_filter(client): _, admin_token = await _setup_admin(client) uid = await _register(client, "alice_test", email="a@x.com") alice_token = await _login(client, "alice_test") await client.patch(f"/v1/admin/users/{uid}", json={"role": "moderator"}, headers={"Authorization": f"Bearer {admin_token}"}) await client.patch(f"/v1/admin/users/{uid}", json={"role": "admin"}, headers={"Authorization": f"Bearer {admin_token}"}) r = await client.get("/v1/notifications", headers={"Authorization": f"Bearer {alice_token}"}) nid = r.json()["notifications"][0]["id"] await client.post(f"/v1/notifications/{nid}/read", headers={"Authorization": f"Bearer {alice_token}"}) r = await client.get("/v1/notifications?unread_only=true", headers={"Authorization": f"Bearer {alice_token}"}) assert len(r.json()["notifications"]) == 1 @pytest.mark.asyncio async def test_notification_requires_auth(client): r = await client.get("/v1/notifications") assert r.status_code in (401, 403, 422) @pytest.mark.asyncio async def test_hub_version_endpoint(client): r = await client.get("/v1/hub/version") assert r.status_code == 200 data = r.json() assert "hub" in data assert "mnp" in data assert "mhp" in data