1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
|
"""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
|