Coverage for app/backend/src/tests/test_notifications.py: 99%
890 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import html
2import json
3import re
4from datetime import timedelta
5from unittest.mock import Mock, patch
6from urllib.parse import parse_qs, urlparse
8import grpc
9import pytest
10from google.protobuf import empty_pb2, timestamp_pb2
11from sqlalchemy import select, update
13from couchers.config import config
14from couchers.constants import DATETIME_INFINITY
15from couchers.context import make_background_user_context
16from couchers.crypto import b64decode
17from couchers.db import session_scope
18from couchers.jobs.handlers import check_expo_push_receipts
19from couchers.jobs.worker import process_job
20from couchers.models import (
21 DeviceType,
22 HostingStatus,
23 MeetupStatus,
24 Notification,
25 NotificationDelivery,
26 NotificationDeliveryType,
27 NotificationTopicAction,
28 PushNotificationDeliveryAttempt,
29 PushNotificationDeliveryOutcome,
30 PushNotificationPlatform,
31 PushNotificationSubscription,
32 User,
33)
34from couchers.notifications.background import handle_notification
35from couchers.notifications.expo_api import get_expo_push_receipts
36from couchers.notifications.notify import notify
37from couchers.notifications.send_raw_push_notification import PushNotificationError, send_raw_push_notification_v2
38from couchers.notifications.settings import get_topic_actions_by_delivery_type, reset_preference
39from couchers.notifications.web_push_api import decode_key, send_web_push
40from couchers.proto import (
41 api_pb2,
42 auth_pb2,
43 conversations_pb2,
44 editor_pb2,
45 events_pb2,
46 moderation_pb2,
47 notification_data_pb2,
48 notifications_pb2,
49)
50from couchers.proto.internal import jobs_pb2, unsubscribe_pb2
51from couchers.servicers.api import user_model_to_pb
52from couchers.utils import not_none, now
53from tests.fixtures.db import generate_user
54from tests.fixtures.misc import EmailCollector, PushCollector, process_jobs
55from tests.fixtures.sessions import (
56 api_session,
57 auth_api_session,
58 conversations_session,
59 notifications_session,
60 real_editor_session,
61)
64@pytest.mark.parametrize("enabled", [True, False])
65def test_SetNotificationSettings_preferences_respected_editable(db, enabled):
66 user, token = generate_user()
68 # enable a notification type and check it gets delivered
69 topic_action = NotificationTopicAction.badge__add
71 with notifications_session(token) as notifications:
72 notifications.SetNotificationSettings(
73 notifications_pb2.SetNotificationSettingsReq(
74 preferences=[
75 notifications_pb2.SingleNotificationPreference(
76 topic=topic_action.topic,
77 action=topic_action.action,
78 delivery_method="push",
79 enabled=enabled,
80 )
81 ],
82 )
83 )
85 with session_scope() as session:
86 notify(
87 session,
88 user_id=user.id,
89 topic_action=topic_action,
90 key="",
91 data=notification_data_pb2.BadgeAdd(
92 badge_id="volunteer",
93 badge_name="Active Volunteer",
94 badge_description="This user is an active volunteer for Couchers.org",
95 ),
96 )
98 process_job()
100 with session_scope() as session:
101 deliv = session.execute(
102 select(NotificationDelivery)
103 .join(Notification, Notification.id == NotificationDelivery.notification_id)
104 .where(Notification.user_id == user.id)
105 .where(Notification.topic_action == topic_action)
106 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
107 ).scalar_one_or_none()
109 if enabled:
110 assert deliv is not None
111 else:
112 assert deliv is None
115def test_SetNotificationSettings_preferences_not_editable(db):
116 user, token = generate_user()
118 # enable a notification type and check it gets delivered
119 topic_action = NotificationTopicAction.password_reset__start
121 with notifications_session(token) as notifications:
122 with pytest.raises(grpc.RpcError) as e:
123 notifications.SetNotificationSettings(
124 notifications_pb2.SetNotificationSettingsReq(
125 preferences=[
126 notifications_pb2.SingleNotificationPreference(
127 topic=topic_action.topic,
128 action=topic_action.action,
129 delivery_method="push",
130 enabled=False,
131 )
132 ],
133 )
134 )
135 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
136 assert e.value.details() == "That notification preference is not user editable."
139def test_unsubscribe(db, email_collector: EmailCollector):
140 # this is the ugliest test i've written
142 user, token = generate_user()
144 topic_action = NotificationTopicAction.badge__add
146 # first enable email notifs
147 with notifications_session(token) as notifications:
148 notifications.SetNotificationSettings(
149 notifications_pb2.SetNotificationSettingsReq(
150 preferences=[
151 notifications_pb2.SingleNotificationPreference(
152 topic=topic_action.topic,
153 action=topic_action.action,
154 delivery_method=method,
155 enabled=enabled,
156 )
157 for method, enabled in [("email", True), ("digest", False), ("push", False)]
158 ],
159 )
160 )
162 with session_scope() as session:
163 notify(
164 session,
165 user_id=user.id,
166 topic_action=topic_action,
167 key="",
168 data=notification_data_pb2.BadgeAdd(
169 badge_id="volunteer",
170 badge_name="Active Volunteer",
171 badge_description="This user is an active volunteer for Couchers.org",
172 ),
173 )
175 email = email_collector.pop_for_recipient(user.email, last=True)
177 # very ugly
178 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
179 for link in re.findall(r'<a href="(.*?)"', email.html): 179 ↛ 200line 179 didn't jump to line 200 because the loop on line 179 didn't complete
180 if "payload" not in link:
181 continue
182 print(link)
183 url_parts = urlparse(html.unescape(link))
184 params = parse_qs(url_parts.query)
185 print(params["payload"][0])
186 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
187 if payload.HasField("topic_action"): 187 ↛ 179line 187 didn't jump to line 179 because the condition on line 187 was always true
188 with auth_api_session() as (auth_api, metadata_interceptor):
189 assert (
190 auth_api.Unsubscribe(
191 auth_pb2.UnsubscribeReq(
192 payload=b64decode(params["payload"][0]),
193 sig=b64decode(params["sig"][0]),
194 )
195 ).response
196 == "You've been unsubscribed from email notifications of that type."
197 )
198 break
199 else:
200 raise Exception("Didn't find link")
202 with notifications_session(token) as notifications:
203 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
205 for group in res.groups:
206 for topic in group.topics:
207 for item in topic.items:
208 if topic == topic_action.topic and item == topic_action.action: 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 assert not item.email
211 with session_scope() as session:
212 notify(
213 session,
214 user_id=user.id,
215 topic_action=topic_action,
216 key="",
217 data=notification_data_pb2.BadgeAdd(
218 badge_id="volunteer",
219 badge_name="Active Volunteer",
220 badge_description="This user is an active volunteer for Couchers.org",
221 ),
222 )
224 assert email_collector.count_for_recipient(user.email) == 0
227def test_unsubscribe_do_not_email(db, email_collector: EmailCollector, moderator):
228 user, token = generate_user()
230 _, token2 = generate_user(complete_profile=True)
231 with api_session(token2) as api:
232 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
233 res = api.ListFriendRequests(empty_pb2.Empty())
234 fr_id = res.sent[0].friend_request_id
236 # Moderator approves the friend request, which triggers the notification email
237 moderator.approve_friend_request(fr_id)
239 email = email_collector.pop_for_recipient(user.email, last=True)
240 assert email.recipient == user.email
241 # very ugly
242 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
243 for link in re.findall(r'<a href="(.*?)"', email.html): 243 ↛ 264line 243 didn't jump to line 264 because the loop on line 243 didn't complete
244 if "payload" not in link:
245 continue
246 print(link)
247 url_parts = urlparse(html.unescape(link))
248 params = parse_qs(url_parts.query)
249 print(params["payload"][0])
250 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
251 if payload.HasField("do_not_email"):
252 with auth_api_session() as (auth_api, metadata_interceptor):
253 assert (
254 auth_api.Unsubscribe(
255 auth_pb2.UnsubscribeReq(
256 payload=b64decode(params["payload"][0]),
257 sig=b64decode(params["sig"][0]),
258 )
259 ).response
260 == "You will not receive any non-security emails, and your hosting status has been turned off. You may still receive the newsletter, and need to unsubscribe from it separately."
261 )
262 break
263 else:
264 raise Exception("Didn't find link")
266 _, token3 = generate_user(complete_profile=True)
267 with api_session(token3) as api:
268 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
269 res = api.ListFriendRequests(empty_pb2.Empty())
270 fr_id3 = res.sent[0].friend_request_id
272 # Approving this friend request should NOT send an email since user has do_not_email set
273 moderator.approve_friend_request(fr_id3)
275 assert email_collector.count_for_recipient(user.email) == 0
277 with session_scope() as session:
278 user_ = session.execute(select(User).where(User.id == user.id)).scalar_one()
279 assert user_.do_not_email
282def test_get_do_not_email(db):
283 _, token = generate_user()
285 with session_scope() as session:
286 user = session.execute(select(User)).scalar_one()
287 user.do_not_email = False
289 with notifications_session(token) as notifications:
290 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
291 assert not res.do_not_email_enabled
293 with session_scope() as session:
294 user = session.execute(select(User)).scalar_one()
295 user.do_not_email = True
296 user.hosting_status = HostingStatus.cant_host
297 user.meetup_status = MeetupStatus.does_not_want_to_meetup
299 with notifications_session(token) as notifications:
300 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
301 assert res.do_not_email_enabled
304def test_set_do_not_email(db):
305 _, token = generate_user()
307 with session_scope() as session:
308 user = session.execute(select(User)).scalar_one()
309 user.do_not_email = False
310 user.hosting_status = HostingStatus.can_host
311 user.meetup_status = MeetupStatus.wants_to_meetup
313 with notifications_session(token) as notifications:
314 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
316 with session_scope() as session:
317 user = session.execute(select(User)).scalar_one()
318 assert not user.do_not_email
320 with notifications_session(token) as notifications:
321 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
323 with session_scope() as session:
324 user = session.execute(select(User)).scalar_one()
325 assert user.do_not_email
326 assert user.hosting_status == HostingStatus.cant_host
327 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup
329 with notifications_session(token) as notifications:
330 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
332 with session_scope() as session:
333 user = session.execute(select(User)).scalar_one()
334 assert not user.do_not_email
337def test_list_notifications(db, push_collector: PushCollector, moderator):
338 user1, token1 = generate_user()
339 user2, token2 = generate_user()
341 with api_session(token2) as api:
342 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
343 res = api.ListFriendRequests(empty_pb2.Empty())
344 fr_id = res.sent[0].friend_request_id
346 # Moderator approves the friend request so the notification is sent
347 moderator.approve_friend_request(fr_id)
349 with notifications_session(token1) as notifications:
350 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
351 assert len(res.notifications) == 1
353 n = res.notifications[0]
355 assert n.topic == "friend_request"
356 assert n.action == "create"
357 assert n.key == str(user2.id)
358 assert n.title == f"Friend request from {user2.name}"
359 assert n.body == f"{user2.name} wants to be your friend."
360 assert n.icon.startswith("http://localhost:5001/img/thumbnail/")
361 assert n.url == f"http://localhost:3000/connections/friends/?from={user2.id}"
363 with conversations_session(token2) as c:
364 res = c.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[user1.id]))
365 group_chat_id = res.group_chat_id
366 moderator.approve_group_chat(group_chat_id)
367 for i in range(17):
368 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text=f"Test message {i}"))
370 process_jobs()
372 all_notifs = []
373 with notifications_session(token1) as notifications:
374 page_token = None
375 for _ in range(100): 375 ↛ 388line 375 didn't jump to line 388
376 res = notifications.ListNotifications(
377 notifications_pb2.ListNotificationsReq(
378 page_size=5,
379 page_token=page_token,
380 )
381 )
382 assert len(res.notifications) == 5 or not res.next_page_token
383 all_notifs += res.notifications
384 page_token = res.next_page_token
385 if not page_token:
386 break
388 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"{user2.name} wants to be your friend."]
389 assert bodys == [n.body for n in all_notifs]
392def test_notifications_seen(db, push_collector: PushCollector, moderator):
393 user1, token1 = generate_user()
394 user2, token2 = generate_user()
395 user3, token3 = generate_user()
396 user4, token4 = generate_user()
398 with api_session(token2) as api:
399 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
400 res = api.ListFriendRequests(empty_pb2.Empty())
401 fr_id2 = res.sent[0].friend_request_id
403 with api_session(token3) as api:
404 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
405 res = api.ListFriendRequests(empty_pb2.Empty())
406 fr_id3 = res.sent[0].friend_request_id
408 # Moderator approves the friend requests so notifications are sent
409 moderator.approve_friend_request(fr_id2)
410 moderator.approve_friend_request(fr_id3)
412 with notifications_session(token1) as notifications, api_session(token1) as api:
413 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
414 assert len(res.notifications) == 2
415 assert [n.is_seen for n in res.notifications] == [False, False]
416 notification_ids = [n.notification_id for n in res.notifications]
417 # should be listed desc time
418 assert notification_ids[0] > notification_ids[1]
420 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
422 with api_session(token4) as api:
423 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
424 res = api.ListFriendRequests(empty_pb2.Empty())
425 fr_id4 = res.sent[0].friend_request_id
427 # Moderator approves the friend request so notification is sent
428 moderator.approve_friend_request(fr_id4)
430 with notifications_session(token1) as notifications, api_session(token1) as api:
431 # mark everything before just the last one as seen (pretend we didn't load the last one yet in the api)
432 notifications.MarkAllNotificationsSeen(
433 notifications_pb2.MarkAllNotificationsSeenReq(latest_notification_id=notification_ids[0])
434 )
436 # last one is still unseen
437 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
439 # mark the first one unseen
440 notifications.MarkNotificationSeen(
441 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids[1], set_seen=False)
442 )
443 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
445 # mark the last one seen
446 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
447 assert len(res.notifications) == 3
448 assert [n.is_seen for n in res.notifications] == [False, True, False]
449 notification_ids2 = [n.notification_id for n in res.notifications]
451 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
453 notifications.MarkNotificationSeen(
454 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids2[0], set_seen=True)
455 )
457 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
458 assert len(res.notifications) == 3
459 assert [n.is_seen for n in res.notifications] == [True, True, False]
461 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
464def test_unseen_notification_count_excludes_ums_hidden(db, moderator):
465 user1, token1 = generate_user()
466 user2, token2 = generate_user()
468 with api_session(token2) as api:
469 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
470 res = api.ListFriendRequests(empty_pb2.Empty())
471 fr_id = res.sent[0].friend_request_id
473 # Before moderation the friend request is shadowed, so the resulting notification
474 # is not visible to the recipient and must not contribute to their unseen count.
475 with api_session(token1) as api:
476 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 0
478 moderator.approve_friend_request(fr_id)
480 with api_session(token1) as api:
481 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
484@pytest.mark.parametrize(
485 "visibility,author_sees,other_sees",
486 [
487 (moderation_pb2.MODERATION_VISIBILITY_VISIBLE, 1, 1),
488 (moderation_pb2.MODERATION_VISIBILITY_UNLISTED, 1, 1),
489 (moderation_pb2.MODERATION_VISIBILITY_SHADOWED, 1, 0),
490 (moderation_pb2.MODERATION_VISIBILITY_HIDDEN, 0, 0),
491 ],
492)
493def test_notifications_follow_the_visibility_of_their_content(db, moderator, visibility, author_sees, other_sees):
494 author, author_token = generate_user()
495 other, other_token = generate_user()
496 sender, sender_token = generate_user()
498 with conversations_session(author_token) as c:
499 group_chat_id = c.CreateGroupChat(
500 conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id, sender.id])
501 ).group_chat_id
502 moderator.approve_group_chat(group_chat_id)
504 with conversations_session(sender_token) as c:
505 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text="Test message"))
507 process_jobs()
509 moderator.set_group_chat_visibility(group_chat_id, visibility)
511 for token, expected in [(author_token, author_sees), (other_token, other_sees)]:
512 with api_session(token) as api:
513 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == expected
514 with notifications_session(token) as notifications:
515 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
516 assert len(res.notifications) == expected
519def test_GetVapidPublicKey(db):
520 _, token = generate_user()
522 with notifications_session(token) as notifications:
523 assert (
524 notifications.GetVapidPublicKey(empty_pb2.Empty()).vapid_public_key
525 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"
526 )
529def test_RegisterPushNotificationSubscription(db):
530 _, token = generate_user()
532 subscription_info = {
533 "endpoint": "https://updates.push.services.mozilla.com/wpush/v2/gAAAAABmW2_iYKVyZRJPhAhktbkXd6Bc8zjIUvtVi5diYL7ZYn8FHka94kIdF46Mp8DwCDWlACnbKOEo97ikaa7JYowGLiGz3qsWL7Vo19LaV4I71mUDUOIKxWIsfp_kM77MlRJQKDUddv-sYyiffOyg63d1lnc_BMIyLXt69T5SEpfnfWTNb6I",
534 "expirationTime": None,
535 "keys": {
536 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
537 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
538 },
539 }
541 with notifications_session(token) as notifications:
542 res = notifications.RegisterPushNotificationSubscription(
543 notifications_pb2.RegisterPushNotificationSubscriptionReq(
544 full_subscription_json=json.dumps(subscription_info),
545 )
546 )
549def test_RegisterPushNotificationSubscription_invalid_endpoint(db):
550 _, token = generate_user()
552 subscription_info = {
553 "endpoint": "https://permanently-removed.invalid/some-id",
554 "expirationTime": None,
555 "keys": {
556 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
557 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
558 },
559 }
561 with notifications_session(token) as notifications:
562 with pytest.raises(grpc.RpcError) as e:
563 notifications.RegisterPushNotificationSubscription(
564 notifications_pb2.RegisterPushNotificationSubscriptionReq(
565 full_subscription_json=json.dumps(subscription_info),
566 )
567 )
568 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
571def test_SendTestPushNotification(db, push_collector: PushCollector):
572 user, token = generate_user()
574 with notifications_session(token) as notifications:
575 notifications.SendTestPushNotification(empty_pb2.Empty())
577 assert push_collector.count_for_user(user.id) == 1
578 push = push_collector.pop_for_user(user.id, last=True)
579 assert push.content.title == "Push notifications test"
580 assert push.content.body == "If you see this, then it's working :)"
583def test_SendBlogPostNotification(db, email_collector: EmailCollector, push_collector: PushCollector):
584 super_user, super_token = generate_user(is_superuser=True)
586 user1, user1_token = generate_user()
587 # enabled email
588 user2, user2_token = generate_user()
589 # disabled push
590 user3, user3_token = generate_user()
592 topic_action = NotificationTopicAction.general__new_blog_post
594 with notifications_session(user2_token) as notifications:
595 notifications.SetNotificationSettings(
596 notifications_pb2.SetNotificationSettingsReq(
597 preferences=[
598 notifications_pb2.SingleNotificationPreference(
599 topic=topic_action.topic,
600 action=topic_action.action,
601 delivery_method="email",
602 enabled=True,
603 )
604 ],
605 )
606 )
608 with notifications_session(user3_token) as notifications:
609 notifications.SetNotificationSettings(
610 notifications_pb2.SetNotificationSettingsReq(
611 preferences=[
612 notifications_pb2.SingleNotificationPreference(
613 topic=topic_action.topic,
614 action=topic_action.action,
615 delivery_method="push",
616 enabled=False,
617 )
618 ],
619 )
620 )
622 with real_editor_session(super_token) as editor_api:
623 editor_api.SendBlogPostNotification(
624 editor_pb2.SendBlogPostNotificationReq(
625 title="Couchers.org v0.9.9 Release Notes",
626 blurb="Read about last major updates before v1!",
627 url="https://couchers.org/blog/2025/05/11/v0.9.9-release",
628 )
629 )
631 email = email_collector.pop_for_recipient(user2.email, last=True)
632 assert email.recipient == user2.email
633 assert "Couchers.org v0.9.9 Release Notes" in email.html
634 assert "Couchers.org v0.9.9 Release Notes" in email.plain
635 assert "Read about last major updates before v1!" in email.html
636 assert "Read about last major updates before v1!" in email.plain
637 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.html
638 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.plain
640 push = push_collector.pop_for_user(user1.id, last=True)
641 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
642 assert push.content.body == "Read about last major updates before v1!"
643 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
645 push = push_collector.pop_for_user(user2.id, last=True)
646 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
647 assert push.content.body == "Read about last major updates before v1!"
648 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
650 assert push_collector.count_for_user(user3.id) == 0
653def test_get_topic_actions_by_delivery_type(db):
654 user, token = generate_user()
656 # these are enabled by default
657 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults
658 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults
660 # these are disabled by default
661 assert NotificationDeliveryType.push not in NotificationTopicAction.event__create_any.defaults
662 assert NotificationDeliveryType.push not in NotificationTopicAction.discussion__create.defaults
664 with notifications_session(token) as notifications:
665 notifications.SetNotificationSettings(
666 notifications_pb2.SetNotificationSettingsReq(
667 preferences=[
668 notifications_pb2.SingleNotificationPreference(
669 topic=NotificationTopicAction.reference__receive_friend.topic,
670 action=NotificationTopicAction.reference__receive_friend.action,
671 delivery_method="push",
672 enabled=False,
673 ),
674 notifications_pb2.SingleNotificationPreference(
675 topic=NotificationTopicAction.event__create_any.topic,
676 action=NotificationTopicAction.event__create_any.action,
677 delivery_method="push",
678 enabled=True,
679 ),
680 ],
681 )
682 )
684 with session_scope() as session:
685 deliver = get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push)
686 assert NotificationTopicAction.reference__receive_friend not in deliver
687 assert NotificationTopicAction.host_request__accept in deliver
688 assert NotificationTopicAction.event__create_any in deliver
689 assert NotificationTopicAction.discussion__create not in deliver
690 assert NotificationTopicAction.account_deletion__start in deliver
693def test_reset_preference(db):
694 user, token = generate_user()
696 topic_action = NotificationTopicAction.event__create_any
697 # neither delivery type is on by default, so enabling both leaves two overrides to tell apart
698 assert NotificationDeliveryType.push not in topic_action.defaults
699 assert NotificationDeliveryType.email not in topic_action.defaults
701 def get_setting() -> notifications_pb2.NotificationItem:
702 with notifications_session(token) as notifications:
703 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
704 items: list[notifications_pb2.NotificationItem] = [
705 item
706 for group in res.groups
707 for topic in group.topics
708 if topic.topic == topic_action.topic
709 for item in topic.items
710 if item.action == topic_action.action
711 ]
712 (item,) = items
713 return item
715 with notifications_session(token) as notifications:
716 notifications.SetNotificationSettings(
717 notifications_pb2.SetNotificationSettingsReq(
718 preferences=[
719 notifications_pb2.SingleNotificationPreference(
720 topic=topic_action.topic,
721 action=topic_action.action,
722 delivery_method=delivery_method,
723 enabled=True,
724 )
725 for delivery_method in ["push", "email"]
726 ]
727 )
728 )
730 assert get_setting().push
731 assert get_setting().email
733 # there is no API for clearing an override back to its default, it's only reachable internally
734 with session_scope() as session:
735 reset_preference(session, user.id, topic_action, NotificationDeliveryType.push)
737 # only the push override should go, leaving the email one in place
738 assert not get_setting().push
739 assert get_setting().email
742def test_event_reminder_email_sent(db, email_collector: EmailCollector):
743 user, token = generate_user()
744 title = "Board Game Night"
746 # Saturday, July 5, 2025 at 4:40:00 AM UTC
747 start_time = timestamp_pb2.Timestamp(seconds=1751690400)
748 timezone = "Etc/GMT-2" # Etc/GMT-2 means GMT+2
749 expected_time_str = "Saturday, July 5, 6:40 AM"
751 with session_scope() as session:
752 user_in_session = session.get_one(User, user.id)
754 notify(
755 session,
756 user_id=user.id,
757 topic_action=NotificationTopicAction.event__reminder,
758 key="",
759 data=notification_data_pb2.EventReminder(
760 event=events_pb2.Event(
761 event_id=1, slug="board-game-night", title=title, start_time=start_time, timezone=timezone
762 ),
763 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)),
764 ),
765 )
767 email = email_collector.pop_for_recipient(user.email, last=True)
768 assert email.recipient == user.email
769 assert title in email.html
770 assert title in email.plain
771 assert expected_time_str in email.html
772 assert expected_time_str in email.plain
775def test_RegisterMobilePushNotificationSubscription(db):
776 user, token = generate_user()
778 with notifications_session(token) as notifications:
779 notifications.RegisterMobilePushNotificationSubscription(
780 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
781 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
782 device_name="My iPhone",
783 device_type="ios",
784 )
785 )
787 # Check subscription was created
788 with session_scope() as session:
789 sub = session.execute(
790 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
791 ).scalar_one()
792 assert sub.platform == PushNotificationPlatform.expo
793 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]"
794 assert sub.device_name == "My iPhone"
795 assert sub.device_type == DeviceType.ios
796 assert sub.disabled_at == DATETIME_INFINITY
799def test_RegisterMobilePushNotificationSubscription_android(db):
800 user, token = generate_user()
802 with notifications_session(token) as notifications:
803 notifications.RegisterMobilePushNotificationSubscription(
804 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
805 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
806 device_name="My Android",
807 device_type="android",
808 )
809 )
811 with session_scope() as session:
812 sub = session.execute(
813 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
814 ).scalar_one()
815 assert sub.platform == PushNotificationPlatform.expo
816 assert sub.device_type == DeviceType.android
819def test_RegisterMobilePushNotificationSubscription_no_device_type(db):
820 user, token = generate_user()
822 with notifications_session(token) as notifications:
823 notifications.RegisterMobilePushNotificationSubscription(
824 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
825 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]",
826 )
827 )
829 with session_scope() as session:
830 sub = session.execute(
831 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
832 ).scalar_one()
833 assert sub.platform == PushNotificationPlatform.expo
834 assert sub.device_name is None
835 assert sub.device_type is None
838def test_RegisterMobilePushNotificationSubscription_re_enable(db):
839 user, token = generate_user()
841 # Create a disabled subscription directly in the DB
842 with session_scope() as session:
843 sub = PushNotificationSubscription(
844 user_id=user.id,
845 platform=PushNotificationPlatform.expo,
846 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
847 device_name="Old Device",
848 device_type=DeviceType.ios,
849 )
850 sub.disabled_at = now()
851 session.add(sub)
852 session.flush()
853 sub_id = sub.id
855 # Re-register with the same token
856 with notifications_session(token) as notifications:
857 notifications.RegisterMobilePushNotificationSubscription(
858 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
859 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
860 device_name="New Device Name",
861 device_type="android",
862 )
863 )
865 # Check subscription was re-enabled and updated
866 with session_scope() as session:
867 sub = session.execute(
868 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
869 ).scalar_one()
870 assert sub.disabled_at == DATETIME_INFINITY
871 assert sub.device_name == "New Device Name"
872 assert sub.device_type == DeviceType.android
875def test_RegisterMobilePushNotificationSubscription_already_exists(db):
876 user, token = generate_user()
878 # Create an active subscription directly in the DB
879 with session_scope() as session:
880 sub = PushNotificationSubscription(
881 user_id=user.id,
882 platform=PushNotificationPlatform.expo,
883 token="ExponentPushToken[existingtoken]",
884 device_name="Existing Device",
885 device_type=DeviceType.ios,
886 )
887 session.add(sub)
889 # Try to register with the same token - should just return without error
890 with notifications_session(token) as notifications:
891 notifications.RegisterMobilePushNotificationSubscription(
892 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
893 token="ExponentPushToken[existingtoken]",
894 device_name="Different Name",
895 )
896 )
898 # Check subscription was NOT modified (already active)
899 with session_scope() as session:
900 sub = session.execute(
901 select(PushNotificationSubscription).where(
902 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]"
903 )
904 ).scalar_one()
905 assert sub.device_name == "Existing Device" # unchanged
908def test_SendTestMobilePushNotification(db, push_collector: PushCollector):
909 user, token = generate_user()
911 with notifications_session(token) as notifications:
912 notifications.SendTestMobilePushNotification(empty_pb2.Empty())
914 push = push_collector.pop_for_user(user.id, last=True)
915 assert push.content.title == "Mobile notifications test"
916 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉"
919def test_get_expo_push_receipts(db):
920 mock_response = Mock()
921 mock_response.status_code = 200
922 mock_response.json.return_value = {
923 "data": {
924 "ticket-1": {"status": "ok"},
925 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
926 }
927 }
929 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post:
930 result = get_expo_push_receipts(["ticket-1", "ticket-2"])
932 mock_post.assert_called_once()
933 call_args = mock_post.call_args
934 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts"
935 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}
937 assert result == {
938 "ticket-1": {"status": "ok"},
939 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
940 }
943def test_get_expo_push_receipts_empty(db):
944 result = get_expo_push_receipts([])
945 assert result == {}
948def test_check_expo_push_receipts_success(db):
949 """Test batch receipt checking with successful delivery."""
950 user, token = generate_user()
952 # Create a push subscription and delivery attempt (old enough to be checked)
953 with session_scope() as session:
954 sub = PushNotificationSubscription(
955 user_id=user.id,
956 platform=PushNotificationPlatform.expo,
957 token="ExponentPushToken[testtoken123]",
958 device_name="Test Device",
959 device_type=DeviceType.ios,
960 )
961 session.add(sub)
962 session.flush()
964 attempt = PushNotificationDeliveryAttempt(
965 push_notification_subscription_id=sub.id,
966 outcome=PushNotificationDeliveryOutcome.success,
967 status_code=200,
968 expo_ticket_id="test-ticket-id",
969 )
970 session.add(attempt)
971 session.flush()
972 # Make the attempt old enough to be checked (>15 min)
973 attempt.time = now() - timedelta(minutes=20)
974 attempt_id = attempt.id
975 sub_id = sub.id
977 # Mock the receipt API call
978 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
979 mock_post.return_value.status_code = 200
980 mock_post.return_value.json.return_value = {"data": {"test-ticket-id": {"status": "ok"}}}
982 check_expo_push_receipts(empty_pb2.Empty())
984 # Verify the attempt was updated
985 with session_scope() as session:
986 attempt = session.execute(
987 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
988 ).scalar_one()
989 assert attempt.receipt_checked_at is not None
990 assert attempt.receipt_status == "ok"
991 assert attempt.receipt_error_code is None
993 # Subscription should still be enabled
994 sub = session.execute(
995 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
996 ).scalar_one()
997 assert sub.disabled_at == DATETIME_INFINITY
1000def test_check_expo_push_receipts_device_not_registered(db, frozen_timewarp):
1001 """Test batch receipt checking with DeviceNotRegistered error disables subscription."""
1002 user, token = generate_user()
1004 # Create a push subscription and delivery attempt
1005 with session_scope() as session:
1006 sub = PushNotificationSubscription(
1007 user_id=user.id,
1008 platform=PushNotificationPlatform.expo,
1009 token="ExponentPushToken[devicegone]",
1010 device_name="Test Device",
1011 device_type=DeviceType.android,
1012 )
1013 session.add(sub)
1014 session.flush()
1016 attempt = PushNotificationDeliveryAttempt(
1017 push_notification_subscription_id=sub.id,
1018 outcome=PushNotificationDeliveryOutcome.success,
1019 status_code=200,
1020 expo_ticket_id="ticket-device-gone",
1021 )
1022 session.add(attempt)
1023 session.flush()
1024 # Make the attempt old enough to be checked
1025 attempt.time = now() - timedelta(minutes=20)
1026 attempt_id = attempt.id
1027 sub_id = sub.id
1029 # Mock the receipt API call with DeviceNotRegistered error
1030 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1031 mock_post.return_value.status_code = 200
1032 mock_post.return_value.json.return_value = {
1033 "data": {
1034 "ticket-device-gone": {
1035 "status": "error",
1036 "details": {"error": "DeviceNotRegistered"},
1037 }
1038 }
1039 }
1041 check_expo_push_receipts(empty_pb2.Empty())
1043 # Verify the attempt was updated and subscription disabled
1044 with session_scope() as session:
1045 attempt = session.execute(
1046 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1047 ).scalar_one()
1048 assert attempt.receipt_checked_at is not None
1049 assert attempt.receipt_status == "error"
1050 assert attempt.receipt_error_code == "DeviceNotRegistered"
1052 # Subscription should be disabled
1053 sub = session.execute(
1054 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1055 ).scalar_one()
1056 assert sub.disabled_at == now()
1059def test_check_expo_push_receipts_not_found(db):
1060 """Test batch receipt checking when ticket not found (expired)."""
1061 user, token = generate_user()
1063 with session_scope() as session:
1064 sub = PushNotificationSubscription(
1065 user_id=user.id,
1066 platform=PushNotificationPlatform.expo,
1067 token="ExponentPushToken[notfound]",
1068 )
1069 session.add(sub)
1070 session.flush()
1072 attempt = PushNotificationDeliveryAttempt(
1073 push_notification_subscription_id=sub.id,
1074 outcome=PushNotificationDeliveryOutcome.success,
1075 status_code=200,
1076 expo_ticket_id="unknown-ticket",
1077 )
1078 session.add(attempt)
1079 session.flush()
1080 # Make the attempt old enough to be checked
1081 attempt.time = now() - timedelta(minutes=15)
1082 attempt_id = attempt.id
1083 sub_id = sub.id
1085 # Mock empty receipt response (ticket not found)
1086 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1087 mock_post.return_value.status_code = 200
1088 mock_post.return_value.json.return_value = {"data": {}}
1090 check_expo_push_receipts(empty_pb2.Empty())
1092 with session_scope() as session:
1093 attempt = session.execute(
1094 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1095 ).scalar_one()
1096 assert attempt.receipt_checked_at is not None
1097 assert attempt.receipt_status == "not_found"
1099 # Subscription should still be enabled
1100 sub = session.execute(
1101 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1102 ).scalar_one()
1103 assert sub.disabled_at == DATETIME_INFINITY
1106def test_check_expo_push_receipts_skips_already_checked(db):
1107 """Test that already-checked receipts are not re-checked."""
1108 user, token = generate_user()
1110 # Create an attempt that was already checked
1111 with session_scope() as session:
1112 sub = PushNotificationSubscription(
1113 user_id=user.id,
1114 platform=PushNotificationPlatform.expo,
1115 token="ExponentPushToken[alreadychecked]",
1116 )
1117 session.add(sub)
1118 session.flush()
1120 attempt = PushNotificationDeliveryAttempt(
1121 push_notification_subscription_id=sub.id,
1122 outcome=PushNotificationDeliveryOutcome.success,
1123 status_code=200,
1124 expo_ticket_id="already-checked-ticket",
1125 receipt_checked_at=now(),
1126 receipt_status="ok",
1127 )
1128 session.add(attempt)
1129 session.flush()
1130 # Make the attempt old enough
1131 attempt.time = now() - timedelta(minutes=15)
1133 # Should not call the API since the only attempt is already checked
1134 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1135 check_expo_push_receipts(empty_pb2.Empty())
1136 mock_post.assert_not_called()
1139def test_SendDevPushNotification_success(db, push_collector: PushCollector):
1140 """Test SendDevPushNotification sends push with all specified parameters."""
1141 user, token = generate_user()
1143 # Enable dev APIs for this test
1144 config.ENABLE_DEV_APIS = True
1146 with notifications_session(token) as notifications:
1147 notifications.SendDevPushNotification(
1148 notifications_pb2.SendDevPushNotificationReq(
1149 title="Test Dev Title",
1150 body="Test dev notification body",
1151 icon="https://example.com/icon.png",
1152 url="https://example.com/action",
1153 key="test-key",
1154 ttl=3600,
1155 )
1156 )
1158 push = push_collector.pop_for_user(user.id, last=True)
1159 assert push.content.title == "Test Dev Title"
1160 assert push.content.body == "Test dev notification body"
1161 assert push.content.action_url == "https://example.com/action"
1162 assert push.content.icon_url == "https://example.com/icon.png"
1163 assert push.topic_action == "adhoc:testing"
1164 assert push.key == "test-key"
1165 assert push.ttl == 3600
1168def test_SendDevPushNotification_minimal(db, push_collector: PushCollector):
1169 """Test SendDevPushNotification with minimal parameters."""
1170 user, token = generate_user()
1172 config.ENABLE_DEV_APIS = True
1174 with notifications_session(token) as notifications:
1175 notifications.SendDevPushNotification(
1176 notifications_pb2.SendDevPushNotificationReq(
1177 title="Minimal Title",
1178 body="Minimal body",
1179 )
1180 )
1182 push = push_collector.pop_for_user(user.id, last=True)
1183 assert push.content.title == "Minimal Title"
1184 assert push.content.body == "Minimal body"
1185 assert push.topic_action == "adhoc:testing"
1188def test_SendDevPushNotification_disabled(db, push_collector: PushCollector):
1189 """Test SendDevPushNotification fails when ENABLE_DEV_APIS is disabled."""
1190 user, token = generate_user()
1192 # Ensure dev APIs are disabled (default in tests)
1193 config.ENABLE_DEV_APIS = False
1195 with notifications_session(token) as notifications:
1196 with pytest.raises(grpc.RpcError) as e:
1197 notifications.SendDevPushNotification(
1198 notifications_pb2.SendDevPushNotificationReq(
1199 title="Should Fail",
1200 body="This should not be sent",
1201 )
1202 )
1203 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1204 assert "Development APIs are not enabled" in not_none(e.value.details())
1206 assert push_collector.count_for_user(user.id) == 0
1209def test_SendDevPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1210 """Test SendDevPushNotification fails when push notifications are disabled."""
1211 user, token = generate_user()
1213 config.ENABLE_DEV_APIS = True
1214 config.PUSH_NOTIFICATIONS_ENABLED = False
1216 with notifications_session(token) as notifications:
1217 with pytest.raises(grpc.RpcError) as e:
1218 notifications.SendDevPushNotification(
1219 notifications_pb2.SendDevPushNotificationReq(
1220 title="Should Fail",
1221 body="This should not be sent",
1222 )
1223 )
1224 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1225 assert "Push notifications are currently disabled" in not_none(e.value.details())
1227 assert push_collector.count_for_user(user.id) == 0
1230def test_check_expo_push_receipts_skips_too_recent(db):
1231 """Test that too-recent receipts (<15 min) are not checked."""
1232 user, token = generate_user()
1234 # Create a recent attempt (not old enough to check)
1235 with session_scope() as session:
1236 sub = PushNotificationSubscription(
1237 user_id=user.id,
1238 platform=PushNotificationPlatform.expo,
1239 token="ExponentPushToken[recent]",
1240 )
1241 session.add(sub)
1242 session.flush()
1244 attempt = PushNotificationDeliveryAttempt(
1245 push_notification_subscription_id=sub.id,
1246 outcome=PushNotificationDeliveryOutcome.success,
1247 status_code=200,
1248 expo_ticket_id="recent-ticket",
1249 )
1250 session.add(attempt)
1251 session.flush()
1252 # Make the attempt only 5 minutes old (too recent)
1253 attempt.time = now() - timedelta(minutes=5)
1255 # Should not call the API since the attempt is too recent
1256 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1257 check_expo_push_receipts(empty_pb2.Empty())
1258 mock_post.assert_not_called()
1261def test_check_expo_push_receipts_batch(db):
1262 """Test that multiple receipts are checked in a single batch."""
1263 user, token = generate_user()
1265 # Create multiple delivery attempts
1266 attempt_ids = []
1267 with session_scope() as session:
1268 sub = PushNotificationSubscription(
1269 user_id=user.id,
1270 platform=PushNotificationPlatform.expo,
1271 token="ExponentPushToken[batch]",
1272 )
1273 session.add(sub)
1274 session.flush()
1276 for i in range(3):
1277 attempt = PushNotificationDeliveryAttempt(
1278 push_notification_subscription_id=sub.id,
1279 outcome=PushNotificationDeliveryOutcome.success,
1280 status_code=200,
1281 expo_ticket_id=f"batch-ticket-{i}",
1282 )
1283 session.add(attempt)
1284 session.flush()
1285 attempt.time = now() - timedelta(minutes=20)
1286 attempt_ids.append(attempt.id)
1288 # Mock the batch receipt API call
1289 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1290 mock_post.return_value.status_code = 200
1291 mock_post.return_value.json.return_value = {
1292 "data": {
1293 "batch-ticket-0": {"status": "ok"},
1294 "batch-ticket-1": {"status": "ok"},
1295 "batch-ticket-2": {"status": "ok"},
1296 }
1297 }
1299 check_expo_push_receipts(empty_pb2.Empty())
1301 # Should only call the API once for all tickets
1302 assert mock_post.call_count == 1
1304 # Verify all attempts were updated
1305 with session_scope() as session:
1306 for attempt_id in attempt_ids:
1307 attempt = session.execute(
1308 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1309 ).scalar_one()
1310 assert attempt.receipt_checked_at is not None
1311 assert attempt.receipt_status == "ok"
1314def test_DebugRedeliverPushNotification_success(db, push_collector: PushCollector):
1315 """Test DebugRedeliverPushNotification redelivers an existing notification."""
1316 user, token = generate_user()
1318 config.ENABLE_DEV_APIS = True
1320 # Create a notification for the user
1321 with session_scope() as session:
1322 notify(
1323 session,
1324 user_id=user.id,
1325 topic_action=NotificationTopicAction.badge__add,
1326 key="test-badge",
1327 data=notification_data_pb2.BadgeAdd(
1328 badge_id="volunteer",
1329 badge_name="Active Volunteer",
1330 badge_description="This user is an active volunteer for Couchers.org",
1331 ),
1332 )
1334 process_job()
1336 # Pop the initial push notification
1337 push_collector.pop_for_user(user.id, last=True)
1339 # Get the notification_id
1340 with session_scope() as session:
1341 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1342 notification_id = notification.id
1344 # Redeliver the notification
1345 with notifications_session(token) as notifications:
1346 notifications.DebugRedeliverPushNotification(
1347 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1348 )
1350 # Verify a new push was sent
1351 push = push_collector.pop_for_user(user.id, last=True)
1352 assert "Active Volunteer" in push.content.title
1353 assert push.topic_action == "badge:add"
1354 assert push.key == "test-badge"
1357def test_DebugRedeliverPushNotification_not_found(db, push_collector: PushCollector):
1358 """Test DebugRedeliverPushNotification fails when notification doesn't exist."""
1359 user, token = generate_user()
1361 config.ENABLE_DEV_APIS = True
1363 with notifications_session(token) as notifications:
1364 with pytest.raises(grpc.RpcError) as e:
1365 notifications.DebugRedeliverPushNotification(
1366 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=999999)
1367 )
1368 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1369 assert "notification not found" in not_none(e.value.details()).lower()
1371 assert push_collector.count_for_user(user.id) == 0
1374def test_DebugRedeliverPushNotification_wrong_user(db, push_collector: PushCollector):
1375 """Test DebugRedeliverPushNotification fails when notification belongs to another user."""
1376 user1, token1 = generate_user()
1377 user2, token2 = generate_user()
1379 config.ENABLE_DEV_APIS = True
1381 # Create a notification for user1
1382 with session_scope() as session:
1383 notify(
1384 session,
1385 user_id=user1.id,
1386 topic_action=NotificationTopicAction.badge__add,
1387 key="test-badge",
1388 data=notification_data_pb2.BadgeAdd(
1389 badge_id="volunteer",
1390 badge_name="Active Volunteer",
1391 badge_description="This user is an active volunteer for Couchers.org",
1392 ),
1393 )
1395 process_job()
1397 # Get the notification_id
1398 with session_scope() as session:
1399 notification = session.execute(select(Notification).where(Notification.user_id == user1.id)).scalar_one()
1400 notification_id = notification.id
1402 # user2 tries to redeliver user1's notification
1403 with notifications_session(token2) as notifications:
1404 with pytest.raises(grpc.RpcError) as e:
1405 notifications.DebugRedeliverPushNotification(
1406 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1407 )
1408 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1409 assert "notification not found" in not_none(e.value.details()).lower()
1411 assert push_collector.count_for_user(user2.id) == 0
1414def test_DebugRedeliverPushNotification_disabled(db, push_collector: PushCollector):
1415 """Test DebugRedeliverPushNotification fails when ENABLE_DEV_APIS is disabled."""
1416 user, token = generate_user()
1418 config.ENABLE_DEV_APIS = False
1420 with notifications_session(token) as notifications:
1421 with pytest.raises(grpc.RpcError) as e:
1422 notifications.DebugRedeliverPushNotification(
1423 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1424 )
1425 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1426 assert "Development APIs are not enabled" in not_none(e.value.details())
1428 assert push_collector.count_for_user(user.id) == 0
1431def test_DebugRedeliverPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1432 """Test DebugRedeliverPushNotification fails when push notifications are disabled."""
1433 user, token = generate_user()
1435 config.ENABLE_DEV_APIS = True
1436 config.PUSH_NOTIFICATIONS_ENABLED = False
1438 with notifications_session(token) as notifications:
1439 with pytest.raises(grpc.RpcError) as e:
1440 notifications.DebugRedeliverPushNotification(
1441 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1442 )
1443 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1444 assert "Push notifications are currently disabled" in not_none(e.value.details())
1446 assert push_collector.count_for_user(user.id) == 0
1449def test_handle_notification_email_delivery(db, email_collector: EmailCollector):
1450 """Test that email notifications are delivered when email preference is enabled."""
1451 user, token = generate_user()
1453 topic_action = NotificationTopicAction.badge__add
1455 # Enable email notifications for this topic
1456 with notifications_session(token) as notifications:
1457 notifications.SetNotificationSettings(
1458 notifications_pb2.SetNotificationSettingsReq(
1459 preferences=[
1460 notifications_pb2.SingleNotificationPreference(
1461 topic=topic_action.topic,
1462 action=topic_action.action,
1463 delivery_method="email",
1464 enabled=True,
1465 )
1466 ],
1467 )
1468 )
1470 with session_scope() as session:
1471 notify(
1472 session,
1473 user_id=user.id,
1474 topic_action=topic_action,
1475 key="test-badge",
1476 data=notification_data_pb2.BadgeAdd(
1477 badge_id="volunteer",
1478 badge_name="Active Volunteer",
1479 badge_description="This user is an active volunteer",
1480 ),
1481 )
1483 email = email_collector.pop_for_recipient(user.email, last=True)
1484 assert email.recipient == user.email
1486 with session_scope() as session:
1487 delivery = session.execute(
1488 select(NotificationDelivery)
1489 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1490 .where(Notification.user_id == user.id)
1491 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.email)
1492 ).scalar_one()
1493 assert delivery.delivered is not None
1496def test_handle_notification_push_delivery(db, push_collector: PushCollector):
1497 """Test that push notifications are delivered immediately when push preference is enabled."""
1498 user, token = generate_user()
1500 topic_action = NotificationTopicAction.badge__add
1502 with session_scope() as session:
1503 notify(
1504 session,
1505 user_id=user.id,
1506 topic_action=topic_action,
1507 key="test-badge",
1508 data=notification_data_pb2.BadgeAdd(
1509 badge_id="volunteer",
1510 badge_name="Active Volunteer",
1511 badge_description="This user is an active volunteer",
1512 ),
1513 )
1515 process_job()
1517 push = push_collector.pop_for_user(user.id, last=True)
1518 assert "Active Volunteer" in push.content.title
1520 with session_scope() as session:
1521 delivery = session.execute(
1522 select(NotificationDelivery)
1523 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1524 .where(Notification.user_id == user.id)
1525 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1526 ).scalar_one()
1527 assert delivery.delivered is not None
1530def test_handle_notification_digest_delivery(db):
1531 """Test that digest notifications are queued without a delivered timestamp."""
1532 user, token = generate_user()
1534 topic_action = NotificationTopicAction.badge__add
1536 # Enable only digest notifications for this topic
1537 with notifications_session(token) as notifications:
1538 notifications.SetNotificationSettings(
1539 notifications_pb2.SetNotificationSettingsReq(
1540 preferences=[
1541 notifications_pb2.SingleNotificationPreference(
1542 topic=topic_action.topic,
1543 action=topic_action.action,
1544 delivery_method="push",
1545 enabled=False,
1546 ),
1547 notifications_pb2.SingleNotificationPreference(
1548 topic=topic_action.topic,
1549 action=topic_action.action,
1550 delivery_method="digest",
1551 enabled=True,
1552 ),
1553 ],
1554 )
1555 )
1557 with session_scope() as session:
1558 notify(
1559 session,
1560 user_id=user.id,
1561 topic_action=topic_action,
1562 key="test-badge",
1563 data=notification_data_pb2.BadgeAdd(
1564 badge_id="volunteer",
1565 badge_name="Active Volunteer",
1566 badge_description="This user is an active volunteer",
1567 ),
1568 )
1570 process_job()
1572 # Verify digest NotificationDelivery was created WITHOUT delivered timestamp
1573 with session_scope() as session:
1574 delivery = session.execute(
1575 select(NotificationDelivery)
1576 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1577 .where(Notification.user_id == user.id)
1578 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.digest)
1579 ).scalar_one()
1580 assert delivery.delivered is None
1583def test_handle_notification_banned_user_no_email(db, email_collector: EmailCollector):
1584 """Test that banned users don't receive email notifications."""
1585 user, token = generate_user()
1587 topic_action = NotificationTopicAction.badge__add
1589 # Enable email notifications
1590 with notifications_session(token) as notifications:
1591 notifications.SetNotificationSettings(
1592 notifications_pb2.SetNotificationSettingsReq(
1593 preferences=[
1594 notifications_pb2.SingleNotificationPreference(
1595 topic=topic_action.topic,
1596 action=topic_action.action,
1597 delivery_method="email",
1598 enabled=True,
1599 )
1600 ],
1601 )
1602 )
1604 # Ban the user
1605 with session_scope() as session:
1606 session.execute(update(User).where(User.id == user.id).values(banned_at=now()))
1608 with session_scope() as session:
1609 notify(
1610 session,
1611 user_id=user.id,
1612 topic_action=topic_action,
1613 key="test-badge",
1614 data=notification_data_pb2.BadgeAdd(
1615 badge_id="volunteer",
1616 badge_name="Active Volunteer",
1617 badge_description="This user is an active volunteer",
1618 ),
1619 )
1621 # Email should not be sent to the banned user
1622 assert email_collector.count_for_recipient(user.email) == 0
1625def test_handle_notification_deleted_user_no_regular_email(db, email_collector: EmailCollector):
1626 """Test that deleted users don't receive non-account-deletion email notifications."""
1627 user, token = generate_user()
1629 topic_action = NotificationTopicAction.badge__add
1631 # Enable email notifications
1632 with notifications_session(token) as notifications:
1633 notifications.SetNotificationSettings(
1634 notifications_pb2.SetNotificationSettingsReq(
1635 preferences=[
1636 notifications_pb2.SingleNotificationPreference(
1637 topic=topic_action.topic,
1638 action=topic_action.action,
1639 delivery_method="email",
1640 enabled=True,
1641 )
1642 ],
1643 )
1644 )
1646 # Delete the user
1647 with session_scope() as session:
1648 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1650 with session_scope() as session:
1651 notify(
1652 session,
1653 user_id=user.id,
1654 topic_action=topic_action,
1655 key="test-badge",
1656 data=notification_data_pb2.BadgeAdd(
1657 badge_id="volunteer",
1658 badge_name="Active Volunteer",
1659 badge_description="This user is an active volunteer",
1660 ),
1661 )
1663 # Email should not be sent to deleted user for non-account-deletion notification
1664 assert email_collector.count_for_recipient(user.email) == 0
1667def test_handle_notification_deleted_user_receives_account_deletion_email(db, email_collector: EmailCollector):
1668 """Test that deleted users CAN receive account deletion notifications."""
1669 user, token = generate_user()
1671 topic_action = NotificationTopicAction.account_deletion__complete
1673 # Delete the user
1674 with session_scope() as session:
1675 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1677 with session_scope() as session:
1678 notify(
1679 session,
1680 user_id=user.id,
1681 topic_action=topic_action,
1682 key="",
1683 data=notification_data_pb2.AccountDeletionComplete(
1684 undelete_token="test-token",
1685 undelete_days=7,
1686 ),
1687 )
1689 # Email SHOULD be sent to deleted user for account deletion notification
1690 email = email_collector.pop_for_recipient(user.email, last=True)
1691 assert email.recipient == user.email
1694def test_handle_notification_do_not_email_respected(db, email_collector: EmailCollector):
1695 """Test that users with do_not_email set don't receive non-critical emails."""
1696 user, token = generate_user()
1698 topic_action = NotificationTopicAction.badge__add
1700 # Enable email notifications
1701 with notifications_session(token) as notifications:
1702 notifications.SetNotificationSettings(
1703 notifications_pb2.SetNotificationSettingsReq(
1704 preferences=[
1705 notifications_pb2.SingleNotificationPreference(
1706 topic=topic_action.topic,
1707 action=topic_action.action,
1708 delivery_method="email",
1709 enabled=True,
1710 )
1711 ],
1712 )
1713 )
1715 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1716 with session_scope() as session:
1717 session.execute(
1718 update(User)
1719 .where(User.id == user.id)
1720 .values(
1721 hosting_status=HostingStatus.cant_host,
1722 meetup_status=MeetupStatus.does_not_want_to_meetup,
1723 do_not_email=True,
1724 )
1725 )
1727 with session_scope() as session:
1728 notify(
1729 session,
1730 user_id=user.id,
1731 topic_action=topic_action,
1732 key="test-badge",
1733 data=notification_data_pb2.BadgeAdd(
1734 badge_id="volunteer",
1735 badge_name="Active Volunteer",
1736 badge_description="This user is an active volunteer",
1737 ),
1738 )
1740 # Email should not be sent when do_not_email is True
1741 assert email_collector.count_for_recipient(user.email) == 0
1744def test_handle_notification_critical_bypasses_do_not_email(db, email_collector: EmailCollector):
1745 """Test that critical notifications bypass do_not_email setting."""
1746 user, token = generate_user()
1748 topic_action = NotificationTopicAction.password__change
1750 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1751 with session_scope() as session:
1752 session.execute(
1753 update(User)
1754 .where(User.id == user.id)
1755 .values(
1756 hosting_status=HostingStatus.cant_host,
1757 meetup_status=MeetupStatus.does_not_want_to_meetup,
1758 do_not_email=True,
1759 )
1760 )
1762 with session_scope() as session:
1763 notify(
1764 session,
1765 user_id=user.id,
1766 topic_action=topic_action,
1767 key="",
1768 data=None,
1769 )
1771 # Critical email SHOULD be sent even with do_not_email=True
1772 email = email_collector.pop_for_recipient(user.email, last=True)
1773 assert email.recipient == user.email
1776def test_handle_notification_duplicate_delivery_skipped(db, push_collector: PushCollector):
1777 """Test that duplicate deliveries are skipped when NotificationDelivery already exists."""
1778 user, token = generate_user()
1780 topic_action = NotificationTopicAction.badge__add
1782 # Create notification manually
1783 with session_scope() as session:
1784 notification = Notification(
1785 user_id=user.id,
1786 topic_action=topic_action,
1787 key="test-badge",
1788 data=notification_data_pb2.BadgeAdd(
1789 badge_id="volunteer",
1790 badge_name="Active Volunteer",
1791 badge_description="This user is an active volunteer",
1792 ).SerializeToString(),
1793 )
1794 session.add(notification)
1795 session.flush()
1796 notification_id = notification.id
1798 # Manually create a push delivery (simulating it was already delivered)
1799 session.add(
1800 NotificationDelivery(
1801 notification_id=notification_id,
1802 delivery_type=NotificationDeliveryType.push,
1803 delivered=now(),
1804 )
1805 )
1807 # Try to handle the notification again
1808 handle_notification(jobs_pb2.HandleNotificationPayload(notification_id=notification_id))
1810 # No new push should be sent since delivery already exists
1811 assert push_collector.count_for_user(user.id) == 0
1813 # Verify only one delivery exists
1814 with session_scope() as session:
1815 delivery_count = len(
1816 session.execute(
1817 select(NotificationDelivery)
1818 .where(NotificationDelivery.notification_id == notification_id)
1819 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1820 )
1821 .scalars()
1822 .all()
1823 )
1824 assert delivery_count == 1
1827def test_handle_notification_deferred_when_content_not_visible(db, moderator):
1828 """Test that notifications linked to non-visible moderated content are deferred."""
1829 user1, token1 = generate_user(complete_profile=True)
1830 user2, token2 = generate_user(complete_profile=True)
1832 # Create a friend request (which creates a moderation state)
1833 # This also queues a notification via SendFriendRequest
1834 with api_session(token2) as api:
1835 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1837 # Process the queued job (handle_notification)
1838 process_job()
1840 # The notification should exist but have no deliveries because content is shadowed
1841 with session_scope() as session:
1842 notification = session.execute(
1843 select(Notification)
1844 .where(Notification.user_id == user1.id)
1845 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1846 ).scalar_one()
1848 deliveries = (
1849 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1850 .scalars()
1851 .all()
1852 )
1853 # No deliveries because content is not yet visible (shadowed)
1854 assert len(deliveries) == 0
1857def test_handle_notification_delivered_when_content_visible(db, moderator):
1858 """Test that notifications linked to visible moderated content are delivered."""
1859 user1, token1 = generate_user(complete_profile=True)
1860 user2, token2 = generate_user(complete_profile=True)
1862 # Create a friend request
1863 with api_session(token2) as api:
1864 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1865 res = api.ListFriendRequests(empty_pb2.Empty())
1866 fr_id = res.sent[0].friend_request_id
1868 # Process initial job (which is deferred because content is shadowed)
1869 process_job()
1871 # Approve the friend request so it becomes visible (this queues the notification job again)
1872 moderator.approve_friend_request(fr_id)
1874 # Process the notification job that was re-queued after approval
1875 process_jobs()
1877 # Notification should have been delivered
1878 with session_scope() as session:
1879 notification = session.execute(
1880 select(Notification)
1881 .where(Notification.user_id == user1.id)
1882 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1883 ).scalar_one()
1885 deliveries = (
1886 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1887 .scalars()
1888 .all()
1889 )
1890 # At least one delivery should exist
1891 assert len(deliveries) > 0
1894def test_notification_serializes_shadowed_actor(db, moderator):
1895 recipient, _ = generate_user(complete_profile=True)
1896 sender, sender_token = generate_user(complete_profile=True)
1898 with session_scope() as session:
1899 session.execute(update(User).where(User.id == sender.id).values(shadowed_at=now()))
1901 with api_session(sender_token) as api:
1902 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=recipient.id))
1904 process_job()
1906 with session_scope() as session:
1907 notification = session.execute(
1908 select(Notification)
1909 .where(Notification.user_id == recipient.id)
1910 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1911 ).scalar_one()
1912 data = notification_data_pb2.FriendRequestCreate.FromString(notification.data)
1913 assert data.other_user.user_id == sender.id
1914 assert not data.other_user.is_ghost
1917def test_handle_notification_multiple_delivery_types(
1918 db, email_collector: EmailCollector, push_collector: PushCollector
1919):
1920 """Test that multiple delivery types are processed for a single notification."""
1921 user, token = generate_user()
1923 topic_action = NotificationTopicAction.badge__add
1925 # Enable both email and push notifications
1926 with notifications_session(token) as notifications:
1927 notifications.SetNotificationSettings(
1928 notifications_pb2.SetNotificationSettingsReq(
1929 preferences=[
1930 notifications_pb2.SingleNotificationPreference(
1931 topic=topic_action.topic,
1932 action=topic_action.action,
1933 delivery_method="email",
1934 enabled=True,
1935 ),
1936 notifications_pb2.SingleNotificationPreference(
1937 topic=topic_action.topic,
1938 action=topic_action.action,
1939 delivery_method="push",
1940 enabled=True,
1941 ),
1942 notifications_pb2.SingleNotificationPreference(
1943 topic=topic_action.topic,
1944 action=topic_action.action,
1945 delivery_method="digest",
1946 enabled=True,
1947 ),
1948 ],
1949 )
1950 )
1952 with session_scope() as session:
1953 notify(
1954 session,
1955 user_id=user.id,
1956 topic_action=topic_action,
1957 key="test-badge",
1958 data=notification_data_pb2.BadgeAdd(
1959 badge_id="volunteer",
1960 badge_name="Active Volunteer",
1961 badge_description="This user is an active volunteer",
1962 ),
1963 )
1965 # Email should be sent
1966 email_collector.pop_for_recipient(user.email, last=True)
1968 # Push should be sent
1969 push = push_collector.pop_for_user(user.id, last=True)
1970 assert "Active Volunteer" in push.content.title
1972 # All three delivery types should have deliveries
1973 with session_scope() as session:
1974 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1976 deliveries = (
1977 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1978 .scalars()
1979 .all()
1980 )
1982 delivery_types = {d.delivery_type for d in deliveries}
1983 assert NotificationDeliveryType.email in delivery_types
1984 assert NotificationDeliveryType.push in delivery_types
1985 assert NotificationDeliveryType.digest in delivery_types
1987 # Email and push should have delivered timestamps
1988 for delivery in deliveries:
1989 if delivery.delivery_type in [NotificationDeliveryType.email, NotificationDeliveryType.push]:
1990 assert delivery.delivered is not None
1991 elif delivery.delivery_type == NotificationDeliveryType.digest: 1991 ↛ 1988line 1991 didn't jump to line 1988 because the condition on line 1991 was always true
1992 assert delivery.delivered is None
1995# a real uncompressed P-256 point, so the aes128gcm encryption in send_web_push actually runs
1996_P256DH_KEY = decode_key("BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0")
1999def _make_web_push_sub(user_id: int) -> int:
2000 with session_scope() as session:
2001 sub = PushNotificationSubscription(
2002 user_id=user_id,
2003 platform=PushNotificationPlatform.web_push,
2004 endpoint="https://updates.push.services.mozilla.com/wpush/v2/sometoken",
2005 auth_key=b"0123456789abcdef",
2006 p256dh_key=b"0123456789abcdef0123456789abcdef",
2007 full_subscription_info="{}",
2008 user_agent="Mozilla/5.0",
2009 )
2010 session.add(sub)
2011 session.flush()
2012 return sub.id
2015def test_web_push_bad_request_is_permanent_message_failure(db):
2016 """A rejected request won't be accepted on a retry, so don't retry it."""
2017 user, _ = generate_user()
2018 sub_id = _make_web_push_sub(user.id)
2020 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2021 mock_send.return_value = Mock(status_code=400, text="bad request", headers={})
2022 send_raw_push_notification_v2(
2023 jobs_pb2.SendRawPushNotificationPayloadV2(
2024 push_notification_subscription_id=sub_id,
2025 title="title",
2026 body="body",
2027 )
2028 )
2030 with session_scope() as session:
2031 attempt = session.execute(
2032 select(PushNotificationDeliveryAttempt).where(
2033 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2034 )
2035 ).scalar_one()
2036 assert attempt.outcome == PushNotificationDeliveryOutcome.permanent_message_failure
2037 assert attempt.status_code == 400
2039 # the subscription itself is still fine
2040 sub = session.execute(
2041 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
2042 ).scalar_one()
2043 assert sub.disabled_at == DATETIME_INFINITY
2046@pytest.mark.parametrize("status_code", [401, 403, 413, 429, 500, 502, 503])
2047def test_web_push_transient_error_is_retried(db, status_code):
2048 """Anything we don't recognise as permanent stays retryable."""
2049 user, _ = generate_user()
2050 sub_id = _make_web_push_sub(user.id)
2052 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2053 mock_send.return_value = Mock(status_code=status_code, text="try again later", headers={})
2054 with pytest.raises(PushNotificationError):
2055 send_raw_push_notification_v2(
2056 jobs_pb2.SendRawPushNotificationPayloadV2(
2057 push_notification_subscription_id=sub_id,
2058 title="title",
2059 body="body",
2060 )
2061 )
2063 with session_scope() as session:
2064 attempt = session.execute(
2065 select(PushNotificationDeliveryAttempt).where(
2066 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2067 )
2068 ).scalar_one()
2069 assert attempt.outcome == PushNotificationDeliveryOutcome.transient_failure
2070 assert attempt.status_code == status_code
2073@pytest.mark.parametrize("status_code", [404, 410])
2074def test_web_push_gone_disables_subscription(db, frozen_timewarp, status_code):
2075 user, _ = generate_user()
2076 sub_id = _make_web_push_sub(user.id)
2078 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2079 mock_send.return_value = Mock(status_code=status_code, text="gone", headers={})
2080 send_raw_push_notification_v2(
2081 jobs_pb2.SendRawPushNotificationPayloadV2(
2082 push_notification_subscription_id=sub_id,
2083 title="title",
2084 body="body",
2085 )
2086 )
2088 with session_scope() as session:
2089 attempt = session.execute(
2090 select(PushNotificationDeliveryAttempt).where(
2091 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2092 )
2093 ).scalar_one()
2094 assert attempt.outcome == PushNotificationDeliveryOutcome.permanent_subscription_failure
2096 sub = session.execute(
2097 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
2098 ).scalar_one()
2099 assert sub.disabled_at == now()
2102@pytest.mark.parametrize(("ttl", "expected"), [(0, "no-cache"), (3600, "cache")])
2103def test_web_push_sends_wns_cache_policy(ttl, expected):
2104 """WNS (Windows Notification Service, Edge's push backend) rejects a mismatched cache policy and ttl."""
2105 with patch("couchers.notifications.web_push_api.requests.post") as mock_post:
2106 send_web_push(
2107 b"data",
2108 "https://wns2-par02p.notify.windows.com/w/?token=abc",
2109 b"0123456789abcdef",
2110 _P256DH_KEY,
2111 "mailto:testing@couchers.org.invalid",
2112 config.PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY,
2113 ttl=ttl,
2114 )
2116 headers = mock_post.call_args.kwargs["headers"]
2117 assert headers["ttl"] == str(ttl)
2118 assert headers["x-wns-cache-policy"] == expected
2121def test_web_push_bad_request_reports_wns_error_to_sentry(db):
2122 """WNS explains a rejection in headers with an empty body, and the drop is silent otherwise."""
2123 user, _ = generate_user()
2124 sub_id = _make_web_push_sub(user.id)
2126 with (
2127 patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send,
2128 patch("couchers.notifications.send_raw_push_notification.sentry_sdk") as mock_sentry,
2129 ):
2130 mock_send.return_value = Mock(
2131 status_code=400,
2132 text="",
2133 headers={
2134 "X-WNS-Error-Description": "Ttl value conflicts with X-WNS-Cache-Policy.",
2135 "X-WNS-Status": "dropped",
2136 "Content-Length": "0",
2137 },
2138 )
2139 send_raw_push_notification_v2(
2140 jobs_pb2.SendRawPushNotificationPayloadV2(
2141 push_notification_subscription_id=sub_id,
2142 title="title",
2143 body="body",
2144 )
2145 )
2147 reported = mock_sentry.capture_exception.call_args[0][0]
2148 assert reported.response_headers["X-WNS-Error-Description"] == "Ttl value conflicts with X-WNS-Cache-Policy."
2149 # unrelated headers aren't worth reporting
2150 assert "Content-Length" not in reported.response_headers
2151 # the message carries them too, so they show up in the Sentry issue itself
2152 assert "Ttl value conflicts" in str(reported)
2154 scope = mock_sentry.new_scope.return_value.__enter__.return_value
2155 context = scope.set_context.call_args[0][1]
2156 assert context["status_code"] == 400
2157 assert context["response_headers"]["X-WNS-Status"] == "dropped"
2160def test_web_push_success_records_body_verbatim(db):
2161 """response holds the body exactly as received, never a wrapper around it."""
2162 user, _ = generate_user()
2163 sub_id = _make_web_push_sub(user.id)
2165 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2166 mock_send.return_value = Mock(status_code=201, text="ok", headers={"X-WNS-Status": "received"})
2167 send_raw_push_notification_v2(
2168 jobs_pb2.SendRawPushNotificationPayloadV2(
2169 push_notification_subscription_id=sub_id,
2170 title="title",
2171 body="body",
2172 )
2173 )
2175 with session_scope() as session:
2176 attempt = session.execute(
2177 select(PushNotificationDeliveryAttempt).where(
2178 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2179 )
2180 ).scalar_one()
2181 assert attempt.outcome == PushNotificationDeliveryOutcome.success
2182 assert attempt.response == "ok"