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
189
190
191
192
193
194
195
196
197
198
199
|
"""
How notifications behave, which is mostly about what they do NOT do.
Three things were wrong and are pinned here: a busy chat produced one row per
message, muting a group did nothing whatsoever (the setting lived in the
browser's localStorage and nothing read it), and there was no way to clear the
list.
"""
import hashlib
import base64
import pytest
from sqlalchemy import select
from meshbay_hub.db.models import GroupMember, Notification, User
def _auth_key(password: str, username: str) -> str:
salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
return base64.b64encode(
hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
async def _user(client, username, password="a-long-enough-passphrase"):
await client.post("/v1/users/register", json={
"username": username, "email": f"{username}@example.com",
"auth_key": _auth_key(password, username)})
r = await client.post("/v1/users/login", json={
"username": username, "auth_key": _auth_key(password, username)})
return r.json()["access_token"]
@pytest.mark.asyncio
async def test_chat_keeps_one_notification_per_group(client, db_session):
"""Forty messages are one line saying when the conversation last spoke."""
from meshbay_hub.api.notifications import create_notification
token = await _user(client, "listener")
owner = await _user(client, "talker_test")
g = await client.post("/v1/groups", json={"name": "busy"},
headers={"Authorization": f"Bearer {owner}"})
gid = g.json()["group_id"]
await client.post(f"/v1/groups/{gid}/members/listener", json={},
headers={"Authorization": f"Bearer {owner}"})
uid = (await db_session.execute(
select(User.id).where(User.username == "listener"))).scalar_one()
for i in range(5):
await create_notification(db_session, uid, "chat_message",
f"talker posted in busy ({i})",
group_id=gid, aggregate=True)
await db_session.commit()
rows = (await db_session.execute(
select(Notification).where(Notification.user_id == uid,
Notification.kind == "chat_message"))).scalars().all()
assert len(rows) == 1, f"{len(rows)} rows for one conversation"
assert rows[0].title.endswith("(4)"), "it should carry the latest message"
assert rows[0].read is False, "a new message makes it unread again"
@pytest.mark.asyncio
async def test_muting_a_group_stops_notifications_being_created(client, db_session):
"""
The point of muting is that nothing is created — not that something is
created and then hidden by whoever happens to be rendering it.
"""
from meshbay_hub.api.notifications import create_notification
token = await _user(client, "quiet_test")
owner = await _user(client, "noisy_test")
g = await client.post("/v1/groups", json={"name": "loud"},
headers={"Authorization": f"Bearer {owner}"})
gid = g.json()["group_id"]
await client.post(f"/v1/groups/{gid}/members/quiet_test", json={},
headers={"Authorization": f"Bearer {owner}"})
r = await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 200, r.text
uid = (await db_session.execute(
select(User.id).where(User.username == "quiet_test"))).scalar_one()
await db_session.execute(
select(GroupMember).where(GroupMember.user_id == uid))
made = await create_notification(db_session, uid, "chat_message",
"noisy posted in loud",
group_id=gid, aggregate=True)
await db_session.commit()
assert made is None, "a muted group must not produce a notification"
rows = (await db_session.execute(
select(Notification).where(Notification.user_id == uid,
Notification.kind == "chat_message"))).scalars().all()
assert rows == []
@pytest.mark.asyncio
async def test_mute_is_reported_with_the_group_list(client):
"""So a second browser shows the same state, which localStorage never did."""
token = await _user(client, "settings-user")
g = await client.post("/v1/groups", json={"name": "mine"},
headers={"Authorization": f"Bearer {token}"})
gid = g.json()["group_id"]
await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
headers={"Authorization": f"Bearer {token}"})
listing = await client.get("/v1/groups/mine",
headers={"Authorization": f"Bearer {token}"})
assert listing.json()["groups"][0]["muted"] is True
@pytest.mark.asyncio
async def test_purge_clears_the_list(client, db_session):
from meshbay_hub.api.notifications import create_notification
token = await _user(client, "cluttered")
uid = (await db_session.execute(
select(User.id).where(User.username == "cluttered"))).scalar_one()
for i in range(3):
await create_notification(db_session, uid, "system", f"thing {i}")
await db_session.commit()
r = await client.delete("/v1/notifications",
headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 200
assert r.json()["removed"] == 3
after = await client.get("/v1/notifications",
headers={"Authorization": f"Bearer {token}"})
assert after.json()["notifications"] == []
assert after.json()["unread_count"] == 0
@pytest.mark.asyncio
async def test_only_your_own_notifications_are_purged(client, db_session):
from meshbay_hub.api.notifications import create_notification
mine = await _user(client, "self_test")
await _user(client, "other_test")
other_id = (await db_session.execute(
select(User.id).where(User.username == "other_test"))).scalar_one()
await create_notification(db_session, other_id, "system", "not yours")
await db_session.commit()
await client.delete("/v1/notifications",
headers={"Authorization": f"Bearer {mine}"})
left = (await db_session.execute(
select(Notification).where(Notification.user_id == other_id))).scalars().all()
assert len(left) == 1, "purging one account emptied another"
@pytest.mark.asyncio
async def test_you_are_not_notified_of_your_own_message(client, db_session):
"""
The node reports who wrote the message. The hub used to substitute the
node's own token subject — the operator's account — so the filter matched
the operator and nobody else: everyone was told about their own messages,
and the operator was told about no one's.
"""
from meshbay_hub.api import revocation as rev
from meshbay_hub.api.revocation import _handle_chat_notify
owner = await _user(client, "operator")
await _user(client, "chatty_test")
await _user(client, "quiet_test")
g = await client.post("/v1/groups", json={"name": "room"},
headers={"Authorization": f"Bearer {owner}"})
gid = g.json()["group_id"]
for name in ("chatty_test", "quiet_test"):
await client.post(f"/v1/groups/{gid}/members/{name}", json={},
headers={"Authorization": f"Bearer {owner}"})
ids = {u.username: u.id for u in (await db_session.execute(
select(User).where(User.username.in_(["operator", "chatty_test", "quiet_test"]))
)).scalars().all()}
# A node registered for this group, because a notification is now written
# only for a group the sending node actually hosts (AV3).
node_id = "notify-test-node"
rev._node_groups[node_id] = [gid]
try:
await _handle_chat_notify(gid, "chatty_test", ids["chatty_test"], node_id=node_id)
finally:
rev._node_groups.pop(node_id, None)
async def chat_rows(uid):
return (await db_session.execute(
select(Notification).where(Notification.user_id == uid,
Notification.kind == "chat_message")
)).scalars().all()
assert await chat_rows(ids["chatty_test"]) == [], "notified of their own message"
assert len(await chat_rows(ids["quiet_test"])) == 1
assert len(await chat_rows(ids["operator"])) == 1, \
"the operator reads the group too, and was the one being skipped"
|