Coverage for app/backend/src/tests/test_notifications.py: 99%
904 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +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 BackgroundJob,
22 DeviceType,
23 HostingStatus,
24 MeetupStatus,
25 Notification,
26 NotificationDelivery,
27 NotificationDeliveryType,
28 NotificationTopicAction,
29 PushNotificationDeliveryAttempt,
30 PushNotificationDeliveryOutcome,
31 PushNotificationPlatform,
32 PushNotificationSubscription,
33 User,
34)
35from couchers.notifications.background import handle_notification
36from couchers.notifications.expo_api import get_expo_push_receipts
37from couchers.notifications.notify import PendingNotification, notify, notify_many
38from couchers.notifications.send_raw_push_notification import PushNotificationError, send_raw_push_notification_v2
39from couchers.notifications.settings import get_topic_actions_by_delivery_type, reset_preference
40from couchers.notifications.web_push_api import decode_key, send_web_push
41from couchers.proto import (
42 api_pb2,
43 auth_pb2,
44 conversations_pb2,
45 editor_pb2,
46 events_pb2,
47 moderation_pb2,
48 notification_data_pb2,
49 notifications_pb2,
50)
51from couchers.proto.internal import jobs_pb2, unsubscribe_pb2
52from couchers.servicers.api import user_model_to_pb
53from couchers.utils import not_none, now
54from tests.fixtures.db import generate_user
55from tests.fixtures.misc import EmailCollector, PushCollector, process_jobs
56from tests.fixtures.sessions import (
57 api_session,
58 auth_api_session,
59 conversations_session,
60 notifications_session,
61 real_editor_session,
62)
65@pytest.mark.parametrize("enabled", [True, False])
66def test_SetNotificationSettings_preferences_respected_editable(db, enabled):
67 user, token = generate_user()
69 # enable a notification type and check it gets delivered
70 topic_action = NotificationTopicAction.badge__add
72 with notifications_session(token) as notifications:
73 notifications.SetNotificationSettings(
74 notifications_pb2.SetNotificationSettingsReq(
75 preferences=[
76 notifications_pb2.SingleNotificationPreference(
77 topic=topic_action.topic,
78 action=topic_action.action,
79 delivery_method="push",
80 enabled=enabled,
81 )
82 ],
83 )
84 )
86 with session_scope() as session:
87 notify(
88 session,
89 user_id=user.id,
90 topic_action=topic_action,
91 key="",
92 data=notification_data_pb2.BadgeAdd(
93 badge_id="volunteer",
94 badge_name="Active Volunteer",
95 badge_description="This user is an active volunteer for Couchers.org",
96 ),
97 )
99 process_job()
101 with session_scope() as session:
102 deliv = session.execute(
103 select(NotificationDelivery)
104 .join(Notification, Notification.id == NotificationDelivery.notification_id)
105 .where(Notification.user_id == user.id)
106 .where(Notification.topic_action == topic_action)
107 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
108 ).scalar_one_or_none()
110 if enabled:
111 assert deliv is not None
112 else:
113 assert deliv is None
116def test_SetNotificationSettings_preferences_not_editable(db):
117 user, token = generate_user()
119 # enable a notification type and check it gets delivered
120 topic_action = NotificationTopicAction.password_reset__start
122 with notifications_session(token) as notifications:
123 with pytest.raises(grpc.RpcError) as e:
124 notifications.SetNotificationSettings(
125 notifications_pb2.SetNotificationSettingsReq(
126 preferences=[
127 notifications_pb2.SingleNotificationPreference(
128 topic=topic_action.topic,
129 action=topic_action.action,
130 delivery_method="push",
131 enabled=False,
132 )
133 ],
134 )
135 )
136 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
137 assert e.value.details() == "That notification preference is not user editable."
140def test_unsubscribe(db, email_collector: EmailCollector):
141 # this is the ugliest test i've written
143 user, token = generate_user()
145 topic_action = NotificationTopicAction.badge__add
147 # first enable email notifs
148 with notifications_session(token) as notifications:
149 notifications.SetNotificationSettings(
150 notifications_pb2.SetNotificationSettingsReq(
151 preferences=[
152 notifications_pb2.SingleNotificationPreference(
153 topic=topic_action.topic,
154 action=topic_action.action,
155 delivery_method=method,
156 enabled=enabled,
157 )
158 for method, enabled in [("email", True), ("digest", False), ("push", False)]
159 ],
160 )
161 )
163 with session_scope() as session:
164 notify(
165 session,
166 user_id=user.id,
167 topic_action=topic_action,
168 key="",
169 data=notification_data_pb2.BadgeAdd(
170 badge_id="volunteer",
171 badge_name="Active Volunteer",
172 badge_description="This user is an active volunteer for Couchers.org",
173 ),
174 )
176 email = email_collector.pop_for_recipient(user.email, last=True)
178 # very ugly
179 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
180 for link in re.findall(r'<a href="(.*?)"', email.html): 180 ↛ 201line 180 didn't jump to line 201 because the loop on line 180 didn't complete
181 if "payload" not in link:
182 continue
183 print(link)
184 url_parts = urlparse(html.unescape(link))
185 params = parse_qs(url_parts.query)
186 print(params["payload"][0])
187 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
188 if payload.HasField("topic_action"): 188 ↛ 180line 188 didn't jump to line 180 because the condition on line 188 was always true
189 with auth_api_session() as (auth_api, metadata_interceptor):
190 assert (
191 auth_api.Unsubscribe(
192 auth_pb2.UnsubscribeReq(
193 payload=b64decode(params["payload"][0]),
194 sig=b64decode(params["sig"][0]),
195 )
196 ).response
197 == "You've been unsubscribed from email notifications of that type."
198 )
199 break
200 else:
201 raise Exception("Didn't find link")
203 with notifications_session(token) as notifications:
204 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
206 for group in res.groups:
207 for topic in group.topics:
208 for item in topic.items:
209 if topic == topic_action.topic and item == topic_action.action: 209 ↛ 210line 209 didn't jump to line 210 because the condition on line 209 was never true
210 assert not item.email
212 with session_scope() as session:
213 notify(
214 session,
215 user_id=user.id,
216 topic_action=topic_action,
217 key="",
218 data=notification_data_pb2.BadgeAdd(
219 badge_id="volunteer",
220 badge_name="Active Volunteer",
221 badge_description="This user is an active volunteer for Couchers.org",
222 ),
223 )
225 assert email_collector.count_for_recipient(user.email) == 0
228def test_unsubscribe_do_not_email(db, email_collector: EmailCollector, moderator):
229 user, token = generate_user()
231 _, token2 = generate_user(complete_profile=True)
232 with api_session(token2) as api:
233 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
234 res = api.ListFriendRequests(empty_pb2.Empty())
235 fr_id = res.sent[0].friend_request_id
237 # Moderator approves the friend request, which triggers the notification email
238 moderator.approve_friend_request(fr_id)
240 email = email_collector.pop_for_recipient(user.email, last=True)
241 assert email.recipient == user.email
242 # very ugly
243 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
244 for link in re.findall(r'<a href="(.*?)"', email.html): 244 ↛ 265line 244 didn't jump to line 265 because the loop on line 244 didn't complete
245 if "payload" not in link:
246 continue
247 print(link)
248 url_parts = urlparse(html.unescape(link))
249 params = parse_qs(url_parts.query)
250 print(params["payload"][0])
251 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
252 if payload.HasField("do_not_email"):
253 with auth_api_session() as (auth_api, metadata_interceptor):
254 assert (
255 auth_api.Unsubscribe(
256 auth_pb2.UnsubscribeReq(
257 payload=b64decode(params["payload"][0]),
258 sig=b64decode(params["sig"][0]),
259 )
260 ).response
261 == "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."
262 )
263 break
264 else:
265 raise Exception("Didn't find link")
267 _, token3 = generate_user(complete_profile=True)
268 with api_session(token3) as api:
269 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
270 res = api.ListFriendRequests(empty_pb2.Empty())
271 fr_id3 = res.sent[0].friend_request_id
273 # Approving this friend request should NOT send an email since user has do_not_email set
274 moderator.approve_friend_request(fr_id3)
276 assert email_collector.count_for_recipient(user.email) == 0
278 with session_scope() as session:
279 user_ = session.execute(select(User).where(User.id == user.id)).scalar_one()
280 assert user_.do_not_email
283def test_get_do_not_email(db):
284 _, token = generate_user()
286 with session_scope() as session:
287 user = session.execute(select(User)).scalar_one()
288 user.do_not_email = False
290 with notifications_session(token) as notifications:
291 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
292 assert not res.do_not_email_enabled
294 with session_scope() as session:
295 user = session.execute(select(User)).scalar_one()
296 user.do_not_email = True
297 user.hosting_status = HostingStatus.cant_host
298 user.meetup_status = MeetupStatus.does_not_want_to_meetup
300 with notifications_session(token) as notifications:
301 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
302 assert res.do_not_email_enabled
305def test_set_do_not_email(db):
306 _, token = generate_user()
308 with session_scope() as session:
309 user = session.execute(select(User)).scalar_one()
310 user.do_not_email = False
311 user.hosting_status = HostingStatus.can_host
312 user.meetup_status = MeetupStatus.wants_to_meetup
314 with notifications_session(token) as notifications:
315 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
317 with session_scope() as session:
318 user = session.execute(select(User)).scalar_one()
319 assert not user.do_not_email
321 with notifications_session(token) as notifications:
322 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
324 with session_scope() as session:
325 user = session.execute(select(User)).scalar_one()
326 assert user.do_not_email
327 assert user.hosting_status == HostingStatus.cant_host
328 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup
330 with notifications_session(token) as notifications:
331 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
333 with session_scope() as session:
334 user = session.execute(select(User)).scalar_one()
335 assert not user.do_not_email
338def test_list_notifications(db, push_collector: PushCollector, moderator):
339 user1, token1 = generate_user()
340 user2, token2 = generate_user()
342 with api_session(token2) as api:
343 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
344 res = api.ListFriendRequests(empty_pb2.Empty())
345 fr_id = res.sent[0].friend_request_id
347 # Moderator approves the friend request so the notification is sent
348 moderator.approve_friend_request(fr_id)
350 with notifications_session(token1) as notifications:
351 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
352 assert len(res.notifications) == 1
354 n = res.notifications[0]
356 assert n.topic == "friend_request"
357 assert n.action == "create"
358 assert n.key == str(user2.id)
359 assert n.title == f"Friend request from {user2.name}"
360 assert n.body == f"{user2.name} wants to be your friend."
361 assert n.icon.startswith("http://localhost:5001/img/thumbnail/")
362 assert n.url == f"http://localhost:3000/connections/friends/?from={user2.id}"
364 with conversations_session(token2) as c:
365 res = c.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[user1.id]))
366 group_chat_id = res.group_chat_id
367 moderator.approve_group_chat(group_chat_id)
368 for i in range(17):
369 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text=f"Test message {i}"))
371 process_jobs()
373 all_notifs = []
374 with notifications_session(token1) as notifications:
375 page_token = None
376 for _ in range(100): 376 ↛ 389line 376 didn't jump to line 389
377 res = notifications.ListNotifications(
378 notifications_pb2.ListNotificationsReq(
379 page_size=5,
380 page_token=page_token,
381 )
382 )
383 assert len(res.notifications) == 5 or not res.next_page_token
384 all_notifs += res.notifications
385 page_token = res.next_page_token
386 if not page_token:
387 break
389 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"{user2.name} wants to be your friend."]
390 assert bodys == [n.body for n in all_notifs]
393def test_notifications_seen(db, push_collector: PushCollector, moderator):
394 user1, token1 = generate_user()
395 user2, token2 = generate_user()
396 user3, token3 = generate_user()
397 user4, token4 = generate_user()
399 with api_session(token2) as api:
400 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
401 res = api.ListFriendRequests(empty_pb2.Empty())
402 fr_id2 = res.sent[0].friend_request_id
404 with api_session(token3) as api:
405 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
406 res = api.ListFriendRequests(empty_pb2.Empty())
407 fr_id3 = res.sent[0].friend_request_id
409 # Moderator approves the friend requests so notifications are sent
410 moderator.approve_friend_request(fr_id2)
411 moderator.approve_friend_request(fr_id3)
413 with notifications_session(token1) as notifications, api_session(token1) as api:
414 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
415 assert len(res.notifications) == 2
416 assert [n.is_seen for n in res.notifications] == [False, False]
417 notification_ids = [n.notification_id for n in res.notifications]
418 # should be listed desc time
419 assert notification_ids[0] > notification_ids[1]
421 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
423 with api_session(token4) as api:
424 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
425 res = api.ListFriendRequests(empty_pb2.Empty())
426 fr_id4 = res.sent[0].friend_request_id
428 # Moderator approves the friend request so notification is sent
429 moderator.approve_friend_request(fr_id4)
431 with notifications_session(token1) as notifications, api_session(token1) as api:
432 # mark everything before just the last one as seen (pretend we didn't load the last one yet in the api)
433 notifications.MarkAllNotificationsSeen(
434 notifications_pb2.MarkAllNotificationsSeenReq(latest_notification_id=notification_ids[0])
435 )
437 # last one is still unseen
438 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
440 # mark the first one unseen
441 notifications.MarkNotificationSeen(
442 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids[1], set_seen=False)
443 )
444 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
446 # mark the last one seen
447 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
448 assert len(res.notifications) == 3
449 assert [n.is_seen for n in res.notifications] == [False, True, False]
450 notification_ids2 = [n.notification_id for n in res.notifications]
452 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
454 notifications.MarkNotificationSeen(
455 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids2[0], set_seen=True)
456 )
458 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
459 assert len(res.notifications) == 3
460 assert [n.is_seen for n in res.notifications] == [True, True, False]
462 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
465def test_unseen_notification_count_excludes_ums_hidden(db, moderator):
466 user1, token1 = generate_user()
467 user2, token2 = generate_user()
469 with api_session(token2) as api:
470 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
471 res = api.ListFriendRequests(empty_pb2.Empty())
472 fr_id = res.sent[0].friend_request_id
474 # Before moderation the friend request is shadowed, so the resulting notification
475 # is not visible to the recipient and must not contribute to their unseen count.
476 with api_session(token1) as api:
477 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 0
479 moderator.approve_friend_request(fr_id)
481 with api_session(token1) as api:
482 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
485@pytest.mark.parametrize(
486 "visibility,author_sees,other_sees",
487 [
488 (moderation_pb2.MODERATION_VISIBILITY_VISIBLE, 1, 1),
489 (moderation_pb2.MODERATION_VISIBILITY_UNLISTED, 1, 1),
490 (moderation_pb2.MODERATION_VISIBILITY_SHADOWED, 1, 0),
491 (moderation_pb2.MODERATION_VISIBILITY_HIDDEN, 0, 0),
492 ],
493)
494def test_notifications_follow_the_visibility_of_their_content(db, moderator, visibility, author_sees, other_sees):
495 author, author_token = generate_user()
496 other, other_token = generate_user()
497 sender, sender_token = generate_user()
499 with conversations_session(author_token) as c:
500 group_chat_id = c.CreateGroupChat(
501 conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id, sender.id])
502 ).group_chat_id
503 moderator.approve_group_chat(group_chat_id)
505 with conversations_session(sender_token) as c:
506 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text="Test message"))
508 process_jobs()
510 moderator.set_group_chat_visibility(group_chat_id, visibility)
512 for token, expected in [(author_token, author_sees), (other_token, other_sees)]:
513 with api_session(token) as api:
514 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == expected
515 with notifications_session(token) as notifications:
516 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
517 assert len(res.notifications) == expected
520def test_GetVapidPublicKey(db):
521 _, token = generate_user()
523 with notifications_session(token) as notifications:
524 assert (
525 notifications.GetVapidPublicKey(empty_pb2.Empty()).vapid_public_key
526 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"
527 )
530def test_RegisterPushNotificationSubscription(db):
531 _, token = generate_user()
533 subscription_info = {
534 "endpoint": "https://updates.push.services.mozilla.com/wpush/v2/gAAAAABmW2_iYKVyZRJPhAhktbkXd6Bc8zjIUvtVi5diYL7ZYn8FHka94kIdF46Mp8DwCDWlACnbKOEo97ikaa7JYowGLiGz3qsWL7Vo19LaV4I71mUDUOIKxWIsfp_kM77MlRJQKDUddv-sYyiffOyg63d1lnc_BMIyLXt69T5SEpfnfWTNb6I",
535 "expirationTime": None,
536 "keys": {
537 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
538 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
539 },
540 }
542 with notifications_session(token) as notifications:
543 res = notifications.RegisterPushNotificationSubscription(
544 notifications_pb2.RegisterPushNotificationSubscriptionReq(
545 full_subscription_json=json.dumps(subscription_info),
546 )
547 )
550def test_RegisterPushNotificationSubscription_invalid_endpoint(db):
551 _, token = generate_user()
553 subscription_info = {
554 "endpoint": "https://permanently-removed.invalid/some-id",
555 "expirationTime": None,
556 "keys": {
557 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
558 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
559 },
560 }
562 with notifications_session(token) as notifications:
563 with pytest.raises(grpc.RpcError) as e:
564 notifications.RegisterPushNotificationSubscription(
565 notifications_pb2.RegisterPushNotificationSubscriptionReq(
566 full_subscription_json=json.dumps(subscription_info),
567 )
568 )
569 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
572def test_SendTestPushNotification(db, push_collector: PushCollector):
573 user, token = generate_user()
575 with notifications_session(token) as notifications:
576 notifications.SendTestPushNotification(empty_pb2.Empty())
578 assert push_collector.count_for_user(user.id) == 1
579 push = push_collector.pop_for_user(user.id, last=True)
580 assert push.content.title == "Push notifications test"
581 assert push.content.body == "If you see this, then it's working :)"
584def test_SendBlogPostNotification(db, email_collector: EmailCollector, push_collector: PushCollector):
585 super_user, super_token = generate_user(is_superuser=True)
587 user1, user1_token = generate_user()
588 # enabled email
589 user2, user2_token = generate_user()
590 # disabled push
591 user3, user3_token = generate_user()
593 topic_action = NotificationTopicAction.general__new_blog_post
595 with notifications_session(user2_token) as notifications:
596 notifications.SetNotificationSettings(
597 notifications_pb2.SetNotificationSettingsReq(
598 preferences=[
599 notifications_pb2.SingleNotificationPreference(
600 topic=topic_action.topic,
601 action=topic_action.action,
602 delivery_method="email",
603 enabled=True,
604 )
605 ],
606 )
607 )
609 with notifications_session(user3_token) as notifications:
610 notifications.SetNotificationSettings(
611 notifications_pb2.SetNotificationSettingsReq(
612 preferences=[
613 notifications_pb2.SingleNotificationPreference(
614 topic=topic_action.topic,
615 action=topic_action.action,
616 delivery_method="push",
617 enabled=False,
618 )
619 ],
620 )
621 )
623 with real_editor_session(super_token) as editor_api:
624 editor_api.SendBlogPostNotification(
625 editor_pb2.SendBlogPostNotificationReq(
626 title="Couchers.org v0.9.9 Release Notes",
627 blurb="Read about last major updates before v1!",
628 url="https://couchers.org/blog/2025/05/11/v0.9.9-release",
629 )
630 )
632 email = email_collector.pop_for_recipient(user2.email, last=True)
633 assert email.recipient == user2.email
634 assert "Couchers.org v0.9.9 Release Notes" in email.html
635 assert "Couchers.org v0.9.9 Release Notes" in email.plain
636 assert "Read about last major updates before v1!" in email.html
637 assert "Read about last major updates before v1!" in email.plain
638 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.html
639 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.plain
641 push = push_collector.pop_for_user(user1.id, last=True)
642 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
643 assert push.content.body == "Read about last major updates before v1!"
644 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
646 push = push_collector.pop_for_user(user2.id, last=True)
647 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
648 assert push.content.body == "Read about last major updates before v1!"
649 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
651 assert push_collector.count_for_user(user3.id) == 0
654def test_get_topic_actions_by_delivery_type(db):
655 user, token = generate_user()
657 # these are enabled by default
658 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults
659 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults
661 # these are disabled by default
662 assert NotificationDeliveryType.push not in NotificationTopicAction.event__create_any.defaults
663 assert NotificationDeliveryType.push not in NotificationTopicAction.discussion__create.defaults
665 with notifications_session(token) as notifications:
666 notifications.SetNotificationSettings(
667 notifications_pb2.SetNotificationSettingsReq(
668 preferences=[
669 notifications_pb2.SingleNotificationPreference(
670 topic=NotificationTopicAction.reference__receive_friend.topic,
671 action=NotificationTopicAction.reference__receive_friend.action,
672 delivery_method="push",
673 enabled=False,
674 ),
675 notifications_pb2.SingleNotificationPreference(
676 topic=NotificationTopicAction.event__create_any.topic,
677 action=NotificationTopicAction.event__create_any.action,
678 delivery_method="push",
679 enabled=True,
680 ),
681 ],
682 )
683 )
685 with session_scope() as session:
686 deliver = get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push)
687 assert NotificationTopicAction.reference__receive_friend not in deliver
688 assert NotificationTopicAction.host_request__accept in deliver
689 assert NotificationTopicAction.event__create_any in deliver
690 assert NotificationTopicAction.discussion__create not in deliver
691 assert NotificationTopicAction.account_deletion__start in deliver
694def test_reset_preference(db):
695 user, token = generate_user()
697 topic_action = NotificationTopicAction.event__create_any
698 # neither delivery type is on by default, so enabling both leaves two overrides to tell apart
699 assert NotificationDeliveryType.push not in topic_action.defaults
700 assert NotificationDeliveryType.email not in topic_action.defaults
702 def get_setting() -> notifications_pb2.NotificationItem:
703 with notifications_session(token) as notifications:
704 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
705 items: list[notifications_pb2.NotificationItem] = [
706 item
707 for group in res.groups
708 for topic in group.topics
709 if topic.topic == topic_action.topic
710 for item in topic.items
711 if item.action == topic_action.action
712 ]
713 (item,) = items
714 return item
716 with notifications_session(token) as notifications:
717 notifications.SetNotificationSettings(
718 notifications_pb2.SetNotificationSettingsReq(
719 preferences=[
720 notifications_pb2.SingleNotificationPreference(
721 topic=topic_action.topic,
722 action=topic_action.action,
723 delivery_method=delivery_method,
724 enabled=True,
725 )
726 for delivery_method in ["push", "email"]
727 ]
728 )
729 )
731 assert get_setting().push
732 assert get_setting().email
734 # there is no API for clearing an override back to its default, it's only reachable internally
735 with session_scope() as session:
736 reset_preference(session, user.id, topic_action, NotificationDeliveryType.push)
738 # only the push override should go, leaving the email one in place
739 assert not get_setting().push
740 assert get_setting().email
743def test_event_reminder_email_sent(db, email_collector: EmailCollector):
744 user, token = generate_user()
745 title = "Board Game Night"
747 # Saturday, July 5, 2025 at 4:40:00 AM UTC
748 start_time = timestamp_pb2.Timestamp(seconds=1751690400)
749 timezone = "Etc/GMT-2" # Etc/GMT-2 means GMT+2
750 expected_time_str = "Saturday, July 5, 6:40 AM"
752 with session_scope() as session:
753 user_in_session = session.get_one(User, user.id)
755 notify(
756 session,
757 user_id=user.id,
758 topic_action=NotificationTopicAction.event__reminder,
759 key="",
760 data=notification_data_pb2.EventReminder(
761 event=events_pb2.Event(
762 event_id=1, slug="board-game-night", title=title, start_time=start_time, timezone=timezone
763 ),
764 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)),
765 ),
766 )
768 email = email_collector.pop_for_recipient(user.email, last=True)
769 assert email.recipient == user.email
770 assert title in email.html
771 assert title in email.plain
772 assert expected_time_str in email.html
773 assert expected_time_str in email.plain
776def test_notify_many(db):
777 user1, _ = generate_user()
778 user2, _ = generate_user()
780 with session_scope() as session:
781 notify_many(session, [])
783 with session_scope() as session:
784 notify_many(
785 session,
786 [
787 PendingNotification(
788 user_id=user1.id,
789 topic_action=NotificationTopicAction.badge__add,
790 key="a",
791 data=notification_data_pb2.BadgeAdd(badge_id="volunteer"),
792 ),
793 PendingNotification(
794 user_id=user2.id,
795 topic_action=NotificationTopicAction.badge__remove,
796 key="b",
797 ),
798 ],
799 )
801 with session_scope() as session:
802 notifications = session.execute(select(Notification).order_by(Notification.id)).scalars().all()
803 assert [(n.user_id, n.topic_action, n.key) for n in notifications] == [
804 (user1.id, NotificationTopicAction.badge__add, "a"),
805 (user2.id, NotificationTopicAction.badge__remove, "b"),
806 ]
807 assert notification_data_pb2.BadgeAdd.FromString(notifications[0].data).badge_id == "volunteer"
808 assert notifications[1].data == b""
810 jobs = (
811 session.execute(
812 select(BackgroundJob)
813 .where(BackgroundJob.job_type == handle_notification.__name__)
814 .order_by(BackgroundJob.id)
815 )
816 .scalars()
817 .all()
818 )
819 assert [jobs_pb2.HandleNotificationPayload.FromString(job.payload).notification_id for job in jobs] == [
820 n.id for n in notifications
821 ]
824def test_RegisterMobilePushNotificationSubscription(db):
825 user, token = generate_user()
827 with notifications_session(token) as notifications:
828 notifications.RegisterMobilePushNotificationSubscription(
829 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
830 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
831 device_name="My iPhone",
832 device_type="ios",
833 )
834 )
836 # Check subscription was created
837 with session_scope() as session:
838 sub = session.execute(
839 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
840 ).scalar_one()
841 assert sub.platform == PushNotificationPlatform.expo
842 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]"
843 assert sub.device_name == "My iPhone"
844 assert sub.device_type == DeviceType.ios
845 assert sub.disabled_at == DATETIME_INFINITY
848def test_RegisterMobilePushNotificationSubscription_android(db):
849 user, token = generate_user()
851 with notifications_session(token) as notifications:
852 notifications.RegisterMobilePushNotificationSubscription(
853 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
854 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
855 device_name="My Android",
856 device_type="android",
857 )
858 )
860 with session_scope() as session:
861 sub = session.execute(
862 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
863 ).scalar_one()
864 assert sub.platform == PushNotificationPlatform.expo
865 assert sub.device_type == DeviceType.android
868def test_RegisterMobilePushNotificationSubscription_no_device_type(db):
869 user, token = generate_user()
871 with notifications_session(token) as notifications:
872 notifications.RegisterMobilePushNotificationSubscription(
873 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
874 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]",
875 )
876 )
878 with session_scope() as session:
879 sub = session.execute(
880 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
881 ).scalar_one()
882 assert sub.platform == PushNotificationPlatform.expo
883 assert sub.device_name is None
884 assert sub.device_type is None
887def test_RegisterMobilePushNotificationSubscription_re_enable(db):
888 user, token = generate_user()
890 # Create a disabled subscription directly in the DB
891 with session_scope() as session:
892 sub = PushNotificationSubscription(
893 user_id=user.id,
894 platform=PushNotificationPlatform.expo,
895 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
896 device_name="Old Device",
897 device_type=DeviceType.ios,
898 )
899 sub.disabled_at = now()
900 session.add(sub)
901 session.flush()
902 sub_id = sub.id
904 # Re-register with the same token
905 with notifications_session(token) as notifications:
906 notifications.RegisterMobilePushNotificationSubscription(
907 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
908 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
909 device_name="New Device Name",
910 device_type="android",
911 )
912 )
914 # Check subscription was re-enabled and updated
915 with session_scope() as session:
916 sub = session.execute(
917 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
918 ).scalar_one()
919 assert sub.disabled_at == DATETIME_INFINITY
920 assert sub.device_name == "New Device Name"
921 assert sub.device_type == DeviceType.android
924def test_RegisterMobilePushNotificationSubscription_already_exists(db):
925 user, token = generate_user()
927 # Create an active subscription directly in the DB
928 with session_scope() as session:
929 sub = PushNotificationSubscription(
930 user_id=user.id,
931 platform=PushNotificationPlatform.expo,
932 token="ExponentPushToken[existingtoken]",
933 device_name="Existing Device",
934 device_type=DeviceType.ios,
935 )
936 session.add(sub)
938 # Try to register with the same token - should just return without error
939 with notifications_session(token) as notifications:
940 notifications.RegisterMobilePushNotificationSubscription(
941 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
942 token="ExponentPushToken[existingtoken]",
943 device_name="Different Name",
944 )
945 )
947 # Check subscription was NOT modified (already active)
948 with session_scope() as session:
949 sub = session.execute(
950 select(PushNotificationSubscription).where(
951 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]"
952 )
953 ).scalar_one()
954 assert sub.device_name == "Existing Device" # unchanged
957def test_SendTestMobilePushNotification(db, push_collector: PushCollector):
958 user, token = generate_user()
960 with notifications_session(token) as notifications:
961 notifications.SendTestMobilePushNotification(empty_pb2.Empty())
963 push = push_collector.pop_for_user(user.id, last=True)
964 assert push.content.title == "Mobile notifications test"
965 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉"
968def test_get_expo_push_receipts(db):
969 mock_response = Mock()
970 mock_response.status_code = 200
971 mock_response.json.return_value = {
972 "data": {
973 "ticket-1": {"status": "ok"},
974 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
975 }
976 }
978 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post:
979 result = get_expo_push_receipts(["ticket-1", "ticket-2"])
981 mock_post.assert_called_once()
982 call_args = mock_post.call_args
983 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts"
984 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}
986 assert result == {
987 "ticket-1": {"status": "ok"},
988 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
989 }
992def test_get_expo_push_receipts_empty(db):
993 result = get_expo_push_receipts([])
994 assert result == {}
997def test_check_expo_push_receipts_success(db):
998 """Test batch receipt checking with successful delivery."""
999 user, token = generate_user()
1001 # Create a push subscription and delivery attempt (old enough to be checked)
1002 with session_scope() as session:
1003 sub = PushNotificationSubscription(
1004 user_id=user.id,
1005 platform=PushNotificationPlatform.expo,
1006 token="ExponentPushToken[testtoken123]",
1007 device_name="Test Device",
1008 device_type=DeviceType.ios,
1009 )
1010 session.add(sub)
1011 session.flush()
1013 attempt = PushNotificationDeliveryAttempt(
1014 push_notification_subscription_id=sub.id,
1015 outcome=PushNotificationDeliveryOutcome.success,
1016 status_code=200,
1017 expo_ticket_id="test-ticket-id",
1018 )
1019 session.add(attempt)
1020 session.flush()
1021 # Make the attempt old enough to be checked (>15 min)
1022 attempt.time = now() - timedelta(minutes=20)
1023 attempt_id = attempt.id
1024 sub_id = sub.id
1026 # Mock the receipt API call
1027 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1028 mock_post.return_value.status_code = 200
1029 mock_post.return_value.json.return_value = {"data": {"test-ticket-id": {"status": "ok"}}}
1031 check_expo_push_receipts(empty_pb2.Empty())
1033 # Verify the attempt was updated
1034 with session_scope() as session:
1035 attempt = session.execute(
1036 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1037 ).scalar_one()
1038 assert attempt.receipt_checked_at is not None
1039 assert attempt.receipt_status == "ok"
1040 assert attempt.receipt_error_code is None
1042 # Subscription should still be enabled
1043 sub = session.execute(
1044 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1045 ).scalar_one()
1046 assert sub.disabled_at == DATETIME_INFINITY
1049def test_check_expo_push_receipts_device_not_registered(db, frozen_timewarp):
1050 """Test batch receipt checking with DeviceNotRegistered error disables subscription."""
1051 user, token = generate_user()
1053 # Create a push subscription and delivery attempt
1054 with session_scope() as session:
1055 sub = PushNotificationSubscription(
1056 user_id=user.id,
1057 platform=PushNotificationPlatform.expo,
1058 token="ExponentPushToken[devicegone]",
1059 device_name="Test Device",
1060 device_type=DeviceType.android,
1061 )
1062 session.add(sub)
1063 session.flush()
1065 attempt = PushNotificationDeliveryAttempt(
1066 push_notification_subscription_id=sub.id,
1067 outcome=PushNotificationDeliveryOutcome.success,
1068 status_code=200,
1069 expo_ticket_id="ticket-device-gone",
1070 )
1071 session.add(attempt)
1072 session.flush()
1073 # Make the attempt old enough to be checked
1074 attempt.time = now() - timedelta(minutes=20)
1075 attempt_id = attempt.id
1076 sub_id = sub.id
1078 # Mock the receipt API call with DeviceNotRegistered error
1079 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1080 mock_post.return_value.status_code = 200
1081 mock_post.return_value.json.return_value = {
1082 "data": {
1083 "ticket-device-gone": {
1084 "status": "error",
1085 "details": {"error": "DeviceNotRegistered"},
1086 }
1087 }
1088 }
1090 check_expo_push_receipts(empty_pb2.Empty())
1092 # Verify the attempt was updated and subscription disabled
1093 with session_scope() as session:
1094 attempt = session.execute(
1095 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1096 ).scalar_one()
1097 assert attempt.receipt_checked_at is not None
1098 assert attempt.receipt_status == "error"
1099 assert attempt.receipt_error_code == "DeviceNotRegistered"
1101 # Subscription should be disabled
1102 sub = session.execute(
1103 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1104 ).scalar_one()
1105 assert sub.disabled_at == now()
1108def test_check_expo_push_receipts_not_found(db):
1109 """Test batch receipt checking when ticket not found (expired)."""
1110 user, token = generate_user()
1112 with session_scope() as session:
1113 sub = PushNotificationSubscription(
1114 user_id=user.id,
1115 platform=PushNotificationPlatform.expo,
1116 token="ExponentPushToken[notfound]",
1117 )
1118 session.add(sub)
1119 session.flush()
1121 attempt = PushNotificationDeliveryAttempt(
1122 push_notification_subscription_id=sub.id,
1123 outcome=PushNotificationDeliveryOutcome.success,
1124 status_code=200,
1125 expo_ticket_id="unknown-ticket",
1126 )
1127 session.add(attempt)
1128 session.flush()
1129 # Make the attempt old enough to be checked
1130 attempt.time = now() - timedelta(minutes=15)
1131 attempt_id = attempt.id
1132 sub_id = sub.id
1134 # Mock empty receipt response (ticket not found)
1135 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1136 mock_post.return_value.status_code = 200
1137 mock_post.return_value.json.return_value = {"data": {}}
1139 check_expo_push_receipts(empty_pb2.Empty())
1141 with session_scope() as session:
1142 attempt = session.execute(
1143 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1144 ).scalar_one()
1145 assert attempt.receipt_checked_at is not None
1146 assert attempt.receipt_status == "not_found"
1148 # Subscription should still be enabled
1149 sub = session.execute(
1150 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1151 ).scalar_one()
1152 assert sub.disabled_at == DATETIME_INFINITY
1155def test_check_expo_push_receipts_skips_already_checked(db):
1156 """Test that already-checked receipts are not re-checked."""
1157 user, token = generate_user()
1159 # Create an attempt that was already checked
1160 with session_scope() as session:
1161 sub = PushNotificationSubscription(
1162 user_id=user.id,
1163 platform=PushNotificationPlatform.expo,
1164 token="ExponentPushToken[alreadychecked]",
1165 )
1166 session.add(sub)
1167 session.flush()
1169 attempt = PushNotificationDeliveryAttempt(
1170 push_notification_subscription_id=sub.id,
1171 outcome=PushNotificationDeliveryOutcome.success,
1172 status_code=200,
1173 expo_ticket_id="already-checked-ticket",
1174 receipt_checked_at=now(),
1175 receipt_status="ok",
1176 )
1177 session.add(attempt)
1178 session.flush()
1179 # Make the attempt old enough
1180 attempt.time = now() - timedelta(minutes=15)
1182 # Should not call the API since the only attempt is already checked
1183 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1184 check_expo_push_receipts(empty_pb2.Empty())
1185 mock_post.assert_not_called()
1188def test_SendDevPushNotification_success(db, push_collector: PushCollector):
1189 """Test SendDevPushNotification sends push with all specified parameters."""
1190 user, token = generate_user()
1192 # Enable dev APIs for this test
1193 config.ENABLE_DEV_APIS = True
1195 with notifications_session(token) as notifications:
1196 notifications.SendDevPushNotification(
1197 notifications_pb2.SendDevPushNotificationReq(
1198 title="Test Dev Title",
1199 body="Test dev notification body",
1200 icon="https://example.com/icon.png",
1201 url="https://example.com/action",
1202 key="test-key",
1203 ttl=3600,
1204 )
1205 )
1207 push = push_collector.pop_for_user(user.id, last=True)
1208 assert push.content.title == "Test Dev Title"
1209 assert push.content.body == "Test dev notification body"
1210 assert push.content.action_url == "https://example.com/action"
1211 assert push.content.icon_url == "https://example.com/icon.png"
1212 assert push.topic_action == "adhoc:testing"
1213 assert push.key == "test-key"
1214 assert push.ttl == 3600
1217def test_SendDevPushNotification_minimal(db, push_collector: PushCollector):
1218 """Test SendDevPushNotification with minimal parameters."""
1219 user, token = generate_user()
1221 config.ENABLE_DEV_APIS = True
1223 with notifications_session(token) as notifications:
1224 notifications.SendDevPushNotification(
1225 notifications_pb2.SendDevPushNotificationReq(
1226 title="Minimal Title",
1227 body="Minimal body",
1228 )
1229 )
1231 push = push_collector.pop_for_user(user.id, last=True)
1232 assert push.content.title == "Minimal Title"
1233 assert push.content.body == "Minimal body"
1234 assert push.topic_action == "adhoc:testing"
1237def test_SendDevPushNotification_disabled(db, push_collector: PushCollector):
1238 """Test SendDevPushNotification fails when ENABLE_DEV_APIS is disabled."""
1239 user, token = generate_user()
1241 # Ensure dev APIs are disabled (default in tests)
1242 config.ENABLE_DEV_APIS = False
1244 with notifications_session(token) as notifications:
1245 with pytest.raises(grpc.RpcError) as e:
1246 notifications.SendDevPushNotification(
1247 notifications_pb2.SendDevPushNotificationReq(
1248 title="Should Fail",
1249 body="This should not be sent",
1250 )
1251 )
1252 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1253 assert "Development APIs are not enabled" in not_none(e.value.details())
1255 assert push_collector.count_for_user(user.id) == 0
1258def test_SendDevPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1259 """Test SendDevPushNotification fails when push notifications are disabled."""
1260 user, token = generate_user()
1262 config.ENABLE_DEV_APIS = True
1263 config.PUSH_NOTIFICATIONS_ENABLED = False
1265 with notifications_session(token) as notifications:
1266 with pytest.raises(grpc.RpcError) as e:
1267 notifications.SendDevPushNotification(
1268 notifications_pb2.SendDevPushNotificationReq(
1269 title="Should Fail",
1270 body="This should not be sent",
1271 )
1272 )
1273 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1274 assert "Push notifications are currently disabled" in not_none(e.value.details())
1276 assert push_collector.count_for_user(user.id) == 0
1279def test_check_expo_push_receipts_skips_too_recent(db):
1280 """Test that too-recent receipts (<15 min) are not checked."""
1281 user, token = generate_user()
1283 # Create a recent attempt (not old enough to check)
1284 with session_scope() as session:
1285 sub = PushNotificationSubscription(
1286 user_id=user.id,
1287 platform=PushNotificationPlatform.expo,
1288 token="ExponentPushToken[recent]",
1289 )
1290 session.add(sub)
1291 session.flush()
1293 attempt = PushNotificationDeliveryAttempt(
1294 push_notification_subscription_id=sub.id,
1295 outcome=PushNotificationDeliveryOutcome.success,
1296 status_code=200,
1297 expo_ticket_id="recent-ticket",
1298 )
1299 session.add(attempt)
1300 session.flush()
1301 # Make the attempt only 5 minutes old (too recent)
1302 attempt.time = now() - timedelta(minutes=5)
1304 # Should not call the API since the attempt is too recent
1305 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1306 check_expo_push_receipts(empty_pb2.Empty())
1307 mock_post.assert_not_called()
1310def test_check_expo_push_receipts_batch(db):
1311 """Test that multiple receipts are checked in a single batch."""
1312 user, token = generate_user()
1314 # Create multiple delivery attempts
1315 attempt_ids = []
1316 with session_scope() as session:
1317 sub = PushNotificationSubscription(
1318 user_id=user.id,
1319 platform=PushNotificationPlatform.expo,
1320 token="ExponentPushToken[batch]",
1321 )
1322 session.add(sub)
1323 session.flush()
1325 for i in range(3):
1326 attempt = PushNotificationDeliveryAttempt(
1327 push_notification_subscription_id=sub.id,
1328 outcome=PushNotificationDeliveryOutcome.success,
1329 status_code=200,
1330 expo_ticket_id=f"batch-ticket-{i}",
1331 )
1332 session.add(attempt)
1333 session.flush()
1334 attempt.time = now() - timedelta(minutes=20)
1335 attempt_ids.append(attempt.id)
1337 # Mock the batch receipt API call
1338 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1339 mock_post.return_value.status_code = 200
1340 mock_post.return_value.json.return_value = {
1341 "data": {
1342 "batch-ticket-0": {"status": "ok"},
1343 "batch-ticket-1": {"status": "ok"},
1344 "batch-ticket-2": {"status": "ok"},
1345 }
1346 }
1348 check_expo_push_receipts(empty_pb2.Empty())
1350 # Should only call the API once for all tickets
1351 assert mock_post.call_count == 1
1353 # Verify all attempts were updated
1354 with session_scope() as session:
1355 for attempt_id in attempt_ids:
1356 attempt = session.execute(
1357 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1358 ).scalar_one()
1359 assert attempt.receipt_checked_at is not None
1360 assert attempt.receipt_status == "ok"
1363def test_DebugRedeliverPushNotification_success(db, push_collector: PushCollector):
1364 """Test DebugRedeliverPushNotification redelivers an existing notification."""
1365 user, token = generate_user()
1367 config.ENABLE_DEV_APIS = True
1369 # Create a notification for the user
1370 with session_scope() as session:
1371 notify(
1372 session,
1373 user_id=user.id,
1374 topic_action=NotificationTopicAction.badge__add,
1375 key="test-badge",
1376 data=notification_data_pb2.BadgeAdd(
1377 badge_id="volunteer",
1378 badge_name="Active Volunteer",
1379 badge_description="This user is an active volunteer for Couchers.org",
1380 ),
1381 )
1383 process_job()
1385 # Pop the initial push notification
1386 push_collector.pop_for_user(user.id, last=True)
1388 # Get the notification_id
1389 with session_scope() as session:
1390 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1391 notification_id = notification.id
1393 # Redeliver the notification
1394 with notifications_session(token) as notifications:
1395 notifications.DebugRedeliverPushNotification(
1396 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1397 )
1399 # Verify a new push was sent
1400 push = push_collector.pop_for_user(user.id, last=True)
1401 assert "Active Volunteer" in push.content.title
1402 assert push.topic_action == "badge:add"
1403 assert push.key == "test-badge"
1406def test_DebugRedeliverPushNotification_not_found(db, push_collector: PushCollector):
1407 """Test DebugRedeliverPushNotification fails when notification doesn't exist."""
1408 user, token = generate_user()
1410 config.ENABLE_DEV_APIS = True
1412 with notifications_session(token) as notifications:
1413 with pytest.raises(grpc.RpcError) as e:
1414 notifications.DebugRedeliverPushNotification(
1415 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=999999)
1416 )
1417 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1418 assert "notification not found" in not_none(e.value.details()).lower()
1420 assert push_collector.count_for_user(user.id) == 0
1423def test_DebugRedeliverPushNotification_wrong_user(db, push_collector: PushCollector):
1424 """Test DebugRedeliverPushNotification fails when notification belongs to another user."""
1425 user1, token1 = generate_user()
1426 user2, token2 = generate_user()
1428 config.ENABLE_DEV_APIS = True
1430 # Create a notification for user1
1431 with session_scope() as session:
1432 notify(
1433 session,
1434 user_id=user1.id,
1435 topic_action=NotificationTopicAction.badge__add,
1436 key="test-badge",
1437 data=notification_data_pb2.BadgeAdd(
1438 badge_id="volunteer",
1439 badge_name="Active Volunteer",
1440 badge_description="This user is an active volunteer for Couchers.org",
1441 ),
1442 )
1444 process_job()
1446 # Get the notification_id
1447 with session_scope() as session:
1448 notification = session.execute(select(Notification).where(Notification.user_id == user1.id)).scalar_one()
1449 notification_id = notification.id
1451 # user2 tries to redeliver user1's notification
1452 with notifications_session(token2) as notifications:
1453 with pytest.raises(grpc.RpcError) as e:
1454 notifications.DebugRedeliverPushNotification(
1455 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1456 )
1457 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1458 assert "notification not found" in not_none(e.value.details()).lower()
1460 assert push_collector.count_for_user(user2.id) == 0
1463def test_DebugRedeliverPushNotification_disabled(db, push_collector: PushCollector):
1464 """Test DebugRedeliverPushNotification fails when ENABLE_DEV_APIS is disabled."""
1465 user, token = generate_user()
1467 config.ENABLE_DEV_APIS = False
1469 with notifications_session(token) as notifications:
1470 with pytest.raises(grpc.RpcError) as e:
1471 notifications.DebugRedeliverPushNotification(
1472 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1473 )
1474 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1475 assert "Development APIs are not enabled" in not_none(e.value.details())
1477 assert push_collector.count_for_user(user.id) == 0
1480def test_DebugRedeliverPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1481 """Test DebugRedeliverPushNotification fails when push notifications are disabled."""
1482 user, token = generate_user()
1484 config.ENABLE_DEV_APIS = True
1485 config.PUSH_NOTIFICATIONS_ENABLED = False
1487 with notifications_session(token) as notifications:
1488 with pytest.raises(grpc.RpcError) as e:
1489 notifications.DebugRedeliverPushNotification(
1490 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1491 )
1492 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1493 assert "Push notifications are currently disabled" in not_none(e.value.details())
1495 assert push_collector.count_for_user(user.id) == 0
1498def test_handle_notification_email_delivery(db, email_collector: EmailCollector):
1499 """Test that email notifications are delivered when email preference is enabled."""
1500 user, token = generate_user()
1502 topic_action = NotificationTopicAction.badge__add
1504 # Enable email notifications for this topic
1505 with notifications_session(token) as notifications:
1506 notifications.SetNotificationSettings(
1507 notifications_pb2.SetNotificationSettingsReq(
1508 preferences=[
1509 notifications_pb2.SingleNotificationPreference(
1510 topic=topic_action.topic,
1511 action=topic_action.action,
1512 delivery_method="email",
1513 enabled=True,
1514 )
1515 ],
1516 )
1517 )
1519 with session_scope() as session:
1520 notify(
1521 session,
1522 user_id=user.id,
1523 topic_action=topic_action,
1524 key="test-badge",
1525 data=notification_data_pb2.BadgeAdd(
1526 badge_id="volunteer",
1527 badge_name="Active Volunteer",
1528 badge_description="This user is an active volunteer",
1529 ),
1530 )
1532 email = email_collector.pop_for_recipient(user.email, last=True)
1533 assert email.recipient == user.email
1535 with session_scope() as session:
1536 delivery = session.execute(
1537 select(NotificationDelivery)
1538 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1539 .where(Notification.user_id == user.id)
1540 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.email)
1541 ).scalar_one()
1542 assert delivery.delivered is not None
1545def test_handle_notification_push_delivery(db, push_collector: PushCollector):
1546 """Test that push notifications are delivered immediately when push preference is enabled."""
1547 user, token = generate_user()
1549 topic_action = NotificationTopicAction.badge__add
1551 with session_scope() as session:
1552 notify(
1553 session,
1554 user_id=user.id,
1555 topic_action=topic_action,
1556 key="test-badge",
1557 data=notification_data_pb2.BadgeAdd(
1558 badge_id="volunteer",
1559 badge_name="Active Volunteer",
1560 badge_description="This user is an active volunteer",
1561 ),
1562 )
1564 process_job()
1566 push = push_collector.pop_for_user(user.id, last=True)
1567 assert "Active Volunteer" in push.content.title
1569 with session_scope() as session:
1570 delivery = session.execute(
1571 select(NotificationDelivery)
1572 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1573 .where(Notification.user_id == user.id)
1574 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1575 ).scalar_one()
1576 assert delivery.delivered is not None
1579def test_handle_notification_digest_delivery(db):
1580 """Test that digest notifications are queued without a delivered timestamp."""
1581 user, token = generate_user()
1583 topic_action = NotificationTopicAction.badge__add
1585 # Enable only digest notifications for this topic
1586 with notifications_session(token) as notifications:
1587 notifications.SetNotificationSettings(
1588 notifications_pb2.SetNotificationSettingsReq(
1589 preferences=[
1590 notifications_pb2.SingleNotificationPreference(
1591 topic=topic_action.topic,
1592 action=topic_action.action,
1593 delivery_method="push",
1594 enabled=False,
1595 ),
1596 notifications_pb2.SingleNotificationPreference(
1597 topic=topic_action.topic,
1598 action=topic_action.action,
1599 delivery_method="digest",
1600 enabled=True,
1601 ),
1602 ],
1603 )
1604 )
1606 with session_scope() as session:
1607 notify(
1608 session,
1609 user_id=user.id,
1610 topic_action=topic_action,
1611 key="test-badge",
1612 data=notification_data_pb2.BadgeAdd(
1613 badge_id="volunteer",
1614 badge_name="Active Volunteer",
1615 badge_description="This user is an active volunteer",
1616 ),
1617 )
1619 process_job()
1621 # Verify digest NotificationDelivery was created WITHOUT delivered timestamp
1622 with session_scope() as session:
1623 delivery = session.execute(
1624 select(NotificationDelivery)
1625 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1626 .where(Notification.user_id == user.id)
1627 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.digest)
1628 ).scalar_one()
1629 assert delivery.delivered is None
1632def test_handle_notification_banned_user_no_email(db, email_collector: EmailCollector):
1633 """Test that banned users don't receive email notifications."""
1634 user, token = generate_user()
1636 topic_action = NotificationTopicAction.badge__add
1638 # Enable email notifications
1639 with notifications_session(token) as notifications:
1640 notifications.SetNotificationSettings(
1641 notifications_pb2.SetNotificationSettingsReq(
1642 preferences=[
1643 notifications_pb2.SingleNotificationPreference(
1644 topic=topic_action.topic,
1645 action=topic_action.action,
1646 delivery_method="email",
1647 enabled=True,
1648 )
1649 ],
1650 )
1651 )
1653 # Ban the user
1654 with session_scope() as session:
1655 session.execute(update(User).where(User.id == user.id).values(banned_at=now()))
1657 with session_scope() as session:
1658 notify(
1659 session,
1660 user_id=user.id,
1661 topic_action=topic_action,
1662 key="test-badge",
1663 data=notification_data_pb2.BadgeAdd(
1664 badge_id="volunteer",
1665 badge_name="Active Volunteer",
1666 badge_description="This user is an active volunteer",
1667 ),
1668 )
1670 # Email should not be sent to the banned user
1671 assert email_collector.count_for_recipient(user.email) == 0
1674def test_handle_notification_deleted_user_no_regular_email(db, email_collector: EmailCollector):
1675 """Test that deleted users don't receive non-account-deletion email notifications."""
1676 user, token = generate_user()
1678 topic_action = NotificationTopicAction.badge__add
1680 # Enable email notifications
1681 with notifications_session(token) as notifications:
1682 notifications.SetNotificationSettings(
1683 notifications_pb2.SetNotificationSettingsReq(
1684 preferences=[
1685 notifications_pb2.SingleNotificationPreference(
1686 topic=topic_action.topic,
1687 action=topic_action.action,
1688 delivery_method="email",
1689 enabled=True,
1690 )
1691 ],
1692 )
1693 )
1695 # Delete the user
1696 with session_scope() as session:
1697 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1699 with session_scope() as session:
1700 notify(
1701 session,
1702 user_id=user.id,
1703 topic_action=topic_action,
1704 key="test-badge",
1705 data=notification_data_pb2.BadgeAdd(
1706 badge_id="volunteer",
1707 badge_name="Active Volunteer",
1708 badge_description="This user is an active volunteer",
1709 ),
1710 )
1712 # Email should not be sent to deleted user for non-account-deletion notification
1713 assert email_collector.count_for_recipient(user.email) == 0
1716def test_handle_notification_deleted_user_receives_account_deletion_email(db, email_collector: EmailCollector):
1717 """Test that deleted users CAN receive account deletion notifications."""
1718 user, token = generate_user()
1720 topic_action = NotificationTopicAction.account_deletion__complete
1722 # Delete the user
1723 with session_scope() as session:
1724 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1726 with session_scope() as session:
1727 notify(
1728 session,
1729 user_id=user.id,
1730 topic_action=topic_action,
1731 key="",
1732 data=notification_data_pb2.AccountDeletionComplete(
1733 undelete_token="test-token",
1734 undelete_days=7,
1735 ),
1736 )
1738 # Email SHOULD be sent to deleted user for account deletion notification
1739 email = email_collector.pop_for_recipient(user.email, last=True)
1740 assert email.recipient == user.email
1743def test_handle_notification_do_not_email_respected(db, email_collector: EmailCollector):
1744 """Test that users with do_not_email set don't receive non-critical emails."""
1745 user, token = generate_user()
1747 topic_action = NotificationTopicAction.badge__add
1749 # Enable email notifications
1750 with notifications_session(token) as notifications:
1751 notifications.SetNotificationSettings(
1752 notifications_pb2.SetNotificationSettingsReq(
1753 preferences=[
1754 notifications_pb2.SingleNotificationPreference(
1755 topic=topic_action.topic,
1756 action=topic_action.action,
1757 delivery_method="email",
1758 enabled=True,
1759 )
1760 ],
1761 )
1762 )
1764 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1765 with session_scope() as session:
1766 session.execute(
1767 update(User)
1768 .where(User.id == user.id)
1769 .values(
1770 hosting_status=HostingStatus.cant_host,
1771 meetup_status=MeetupStatus.does_not_want_to_meetup,
1772 do_not_email=True,
1773 )
1774 )
1776 with session_scope() as session:
1777 notify(
1778 session,
1779 user_id=user.id,
1780 topic_action=topic_action,
1781 key="test-badge",
1782 data=notification_data_pb2.BadgeAdd(
1783 badge_id="volunteer",
1784 badge_name="Active Volunteer",
1785 badge_description="This user is an active volunteer",
1786 ),
1787 )
1789 # Email should not be sent when do_not_email is True
1790 assert email_collector.count_for_recipient(user.email) == 0
1793def test_handle_notification_critical_bypasses_do_not_email(db, email_collector: EmailCollector):
1794 """Test that critical notifications bypass do_not_email setting."""
1795 user, token = generate_user()
1797 topic_action = NotificationTopicAction.password__change
1799 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1800 with session_scope() as session:
1801 session.execute(
1802 update(User)
1803 .where(User.id == user.id)
1804 .values(
1805 hosting_status=HostingStatus.cant_host,
1806 meetup_status=MeetupStatus.does_not_want_to_meetup,
1807 do_not_email=True,
1808 )
1809 )
1811 with session_scope() as session:
1812 notify(
1813 session,
1814 user_id=user.id,
1815 topic_action=topic_action,
1816 key="",
1817 data=None,
1818 )
1820 # Critical email SHOULD be sent even with do_not_email=True
1821 email = email_collector.pop_for_recipient(user.email, last=True)
1822 assert email.recipient == user.email
1825def test_handle_notification_duplicate_delivery_skipped(db, push_collector: PushCollector):
1826 """Test that duplicate deliveries are skipped when NotificationDelivery already exists."""
1827 user, token = generate_user()
1829 topic_action = NotificationTopicAction.badge__add
1831 # Create notification manually
1832 with session_scope() as session:
1833 notification = Notification(
1834 user_id=user.id,
1835 topic_action=topic_action,
1836 key="test-badge",
1837 data=notification_data_pb2.BadgeAdd(
1838 badge_id="volunteer",
1839 badge_name="Active Volunteer",
1840 badge_description="This user is an active volunteer",
1841 ).SerializeToString(),
1842 )
1843 session.add(notification)
1844 session.flush()
1845 notification_id = notification.id
1847 # Manually create a push delivery (simulating it was already delivered)
1848 session.add(
1849 NotificationDelivery(
1850 notification_id=notification_id,
1851 delivery_type=NotificationDeliveryType.push,
1852 delivered=now(),
1853 )
1854 )
1856 # Try to handle the notification again
1857 handle_notification(jobs_pb2.HandleNotificationPayload(notification_id=notification_id))
1859 # No new push should be sent since delivery already exists
1860 assert push_collector.count_for_user(user.id) == 0
1862 # Verify only one delivery exists
1863 with session_scope() as session:
1864 delivery_count = len(
1865 session.execute(
1866 select(NotificationDelivery)
1867 .where(NotificationDelivery.notification_id == notification_id)
1868 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1869 )
1870 .scalars()
1871 .all()
1872 )
1873 assert delivery_count == 1
1876def test_handle_notification_deferred_when_content_not_visible(db, moderator):
1877 """Test that notifications linked to non-visible moderated content are deferred."""
1878 user1, token1 = generate_user(complete_profile=True)
1879 user2, token2 = generate_user(complete_profile=True)
1881 # Create a friend request (which creates a moderation state)
1882 # This also queues a notification via SendFriendRequest
1883 with api_session(token2) as api:
1884 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1886 # Process the queued job (handle_notification)
1887 process_job()
1889 # The notification should exist but have no deliveries because content is shadowed
1890 with session_scope() as session:
1891 notification = session.execute(
1892 select(Notification)
1893 .where(Notification.user_id == user1.id)
1894 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1895 ).scalar_one()
1897 deliveries = (
1898 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1899 .scalars()
1900 .all()
1901 )
1902 # No deliveries because content is not yet visible (shadowed)
1903 assert len(deliveries) == 0
1906def test_handle_notification_delivered_when_content_visible(db, moderator):
1907 """Test that notifications linked to visible moderated content are delivered."""
1908 user1, token1 = generate_user(complete_profile=True)
1909 user2, token2 = generate_user(complete_profile=True)
1911 # Create a friend request
1912 with api_session(token2) as api:
1913 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1914 res = api.ListFriendRequests(empty_pb2.Empty())
1915 fr_id = res.sent[0].friend_request_id
1917 # Process initial job (which is deferred because content is shadowed)
1918 process_job()
1920 # Approve the friend request so it becomes visible (this queues the notification job again)
1921 moderator.approve_friend_request(fr_id)
1923 # Process the notification job that was re-queued after approval
1924 process_jobs()
1926 # Notification should have been delivered
1927 with session_scope() as session:
1928 notification = session.execute(
1929 select(Notification)
1930 .where(Notification.user_id == user1.id)
1931 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1932 ).scalar_one()
1934 deliveries = (
1935 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1936 .scalars()
1937 .all()
1938 )
1939 # At least one delivery should exist
1940 assert len(deliveries) > 0
1943def test_notification_serializes_shadowed_actor(db, moderator):
1944 recipient, _ = generate_user(complete_profile=True)
1945 sender, sender_token = generate_user(complete_profile=True)
1947 with session_scope() as session:
1948 session.execute(update(User).where(User.id == sender.id).values(shadowed_at=now()))
1950 with api_session(sender_token) as api:
1951 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=recipient.id))
1953 process_job()
1955 with session_scope() as session:
1956 notification = session.execute(
1957 select(Notification)
1958 .where(Notification.user_id == recipient.id)
1959 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1960 ).scalar_one()
1961 data = notification_data_pb2.FriendRequestCreate.FromString(notification.data)
1962 assert data.other_user.user_id == sender.id
1963 assert not data.other_user.is_ghost
1966def test_handle_notification_multiple_delivery_types(
1967 db, email_collector: EmailCollector, push_collector: PushCollector
1968):
1969 """Test that multiple delivery types are processed for a single notification."""
1970 user, token = generate_user()
1972 topic_action = NotificationTopicAction.badge__add
1974 # Enable both email and push notifications
1975 with notifications_session(token) as notifications:
1976 notifications.SetNotificationSettings(
1977 notifications_pb2.SetNotificationSettingsReq(
1978 preferences=[
1979 notifications_pb2.SingleNotificationPreference(
1980 topic=topic_action.topic,
1981 action=topic_action.action,
1982 delivery_method="email",
1983 enabled=True,
1984 ),
1985 notifications_pb2.SingleNotificationPreference(
1986 topic=topic_action.topic,
1987 action=topic_action.action,
1988 delivery_method="push",
1989 enabled=True,
1990 ),
1991 notifications_pb2.SingleNotificationPreference(
1992 topic=topic_action.topic,
1993 action=topic_action.action,
1994 delivery_method="digest",
1995 enabled=True,
1996 ),
1997 ],
1998 )
1999 )
2001 with session_scope() as session:
2002 notify(
2003 session,
2004 user_id=user.id,
2005 topic_action=topic_action,
2006 key="test-badge",
2007 data=notification_data_pb2.BadgeAdd(
2008 badge_id="volunteer",
2009 badge_name="Active Volunteer",
2010 badge_description="This user is an active volunteer",
2011 ),
2012 )
2014 # Email should be sent
2015 email_collector.pop_for_recipient(user.email, last=True)
2017 # Push should be sent
2018 push = push_collector.pop_for_user(user.id, last=True)
2019 assert "Active Volunteer" in push.content.title
2021 # All three delivery types should have deliveries
2022 with session_scope() as session:
2023 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
2025 deliveries = (
2026 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
2027 .scalars()
2028 .all()
2029 )
2031 delivery_types = {d.delivery_type for d in deliveries}
2032 assert NotificationDeliveryType.email in delivery_types
2033 assert NotificationDeliveryType.push in delivery_types
2034 assert NotificationDeliveryType.digest in delivery_types
2036 # Email and push should have delivered timestamps
2037 for delivery in deliveries:
2038 if delivery.delivery_type in [NotificationDeliveryType.email, NotificationDeliveryType.push]:
2039 assert delivery.delivered is not None
2040 elif delivery.delivery_type == NotificationDeliveryType.digest: 2040 ↛ 2037line 2040 didn't jump to line 2037 because the condition on line 2040 was always true
2041 assert delivery.delivered is None
2044# a real uncompressed P-256 point, so the aes128gcm encryption in send_web_push actually runs
2045_P256DH_KEY = decode_key("BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0")
2048def _make_web_push_sub(user_id: int) -> int:
2049 with session_scope() as session:
2050 sub = PushNotificationSubscription(
2051 user_id=user_id,
2052 platform=PushNotificationPlatform.web_push,
2053 endpoint="https://updates.push.services.mozilla.com/wpush/v2/sometoken",
2054 auth_key=b"0123456789abcdef",
2055 p256dh_key=b"0123456789abcdef0123456789abcdef",
2056 full_subscription_info="{}",
2057 user_agent="Mozilla/5.0",
2058 )
2059 session.add(sub)
2060 session.flush()
2061 return sub.id
2064def test_web_push_bad_request_is_permanent_message_failure(db):
2065 """A rejected request won't be accepted on a retry, so don't retry it."""
2066 user, _ = generate_user()
2067 sub_id = _make_web_push_sub(user.id)
2069 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2070 mock_send.return_value = Mock(status_code=400, text="bad request", headers={})
2071 send_raw_push_notification_v2(
2072 jobs_pb2.SendRawPushNotificationPayloadV2(
2073 push_notification_subscription_id=sub_id,
2074 title="title",
2075 body="body",
2076 )
2077 )
2079 with session_scope() as session:
2080 attempt = session.execute(
2081 select(PushNotificationDeliveryAttempt).where(
2082 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2083 )
2084 ).scalar_one()
2085 assert attempt.outcome == PushNotificationDeliveryOutcome.permanent_message_failure
2086 assert attempt.status_code == 400
2088 # the subscription itself is still fine
2089 sub = session.execute(
2090 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
2091 ).scalar_one()
2092 assert sub.disabled_at == DATETIME_INFINITY
2095@pytest.mark.parametrize("status_code", [401, 403, 413, 429, 500, 502, 503])
2096def test_web_push_transient_error_is_retried(db, status_code):
2097 """Anything we don't recognise as permanent stays retryable."""
2098 user, _ = generate_user()
2099 sub_id = _make_web_push_sub(user.id)
2101 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2102 mock_send.return_value = Mock(status_code=status_code, text="try again later", headers={})
2103 with pytest.raises(PushNotificationError):
2104 send_raw_push_notification_v2(
2105 jobs_pb2.SendRawPushNotificationPayloadV2(
2106 push_notification_subscription_id=sub_id,
2107 title="title",
2108 body="body",
2109 )
2110 )
2112 with session_scope() as session:
2113 attempt = session.execute(
2114 select(PushNotificationDeliveryAttempt).where(
2115 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2116 )
2117 ).scalar_one()
2118 assert attempt.outcome == PushNotificationDeliveryOutcome.transient_failure
2119 assert attempt.status_code == status_code
2122@pytest.mark.parametrize("status_code", [404, 410])
2123def test_web_push_gone_disables_subscription(db, frozen_timewarp, status_code):
2124 user, _ = generate_user()
2125 sub_id = _make_web_push_sub(user.id)
2127 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2128 mock_send.return_value = Mock(status_code=status_code, text="gone", headers={})
2129 send_raw_push_notification_v2(
2130 jobs_pb2.SendRawPushNotificationPayloadV2(
2131 push_notification_subscription_id=sub_id,
2132 title="title",
2133 body="body",
2134 )
2135 )
2137 with session_scope() as session:
2138 attempt = session.execute(
2139 select(PushNotificationDeliveryAttempt).where(
2140 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2141 )
2142 ).scalar_one()
2143 assert attempt.outcome == PushNotificationDeliveryOutcome.permanent_subscription_failure
2145 sub = session.execute(
2146 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
2147 ).scalar_one()
2148 assert sub.disabled_at == now()
2151@pytest.mark.parametrize(("ttl", "expected"), [(0, "no-cache"), (3600, "cache")])
2152def test_web_push_sends_wns_cache_policy(ttl, expected):
2153 """WNS (Windows Notification Service, Edge's push backend) rejects a mismatched cache policy and ttl."""
2154 with patch("couchers.notifications.web_push_api.requests.post") as mock_post:
2155 send_web_push(
2156 b"data",
2157 "https://wns2-par02p.notify.windows.com/w/?token=abc",
2158 b"0123456789abcdef",
2159 _P256DH_KEY,
2160 "mailto:testing@couchers.org.invalid",
2161 config.PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY,
2162 ttl=ttl,
2163 )
2165 headers = mock_post.call_args.kwargs["headers"]
2166 assert headers["ttl"] == str(ttl)
2167 assert headers["x-wns-cache-policy"] == expected
2170def test_web_push_bad_request_reports_wns_error_to_sentry(db):
2171 """WNS explains a rejection in headers with an empty body, and the drop is silent otherwise."""
2172 user, _ = generate_user()
2173 sub_id = _make_web_push_sub(user.id)
2175 with (
2176 patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send,
2177 patch("couchers.notifications.send_raw_push_notification.sentry_sdk") as mock_sentry,
2178 ):
2179 mock_send.return_value = Mock(
2180 status_code=400,
2181 text="",
2182 headers={
2183 "X-WNS-Error-Description": "Ttl value conflicts with X-WNS-Cache-Policy.",
2184 "X-WNS-Status": "dropped",
2185 "Content-Length": "0",
2186 },
2187 )
2188 send_raw_push_notification_v2(
2189 jobs_pb2.SendRawPushNotificationPayloadV2(
2190 push_notification_subscription_id=sub_id,
2191 title="title",
2192 body="body",
2193 )
2194 )
2196 reported = mock_sentry.capture_exception.call_args[0][0]
2197 assert reported.response_headers["X-WNS-Error-Description"] == "Ttl value conflicts with X-WNS-Cache-Policy."
2198 # unrelated headers aren't worth reporting
2199 assert "Content-Length" not in reported.response_headers
2200 # the message carries them too, so they show up in the Sentry issue itself
2201 assert "Ttl value conflicts" in str(reported)
2203 scope = mock_sentry.new_scope.return_value.__enter__.return_value
2204 context = scope.set_context.call_args[0][1]
2205 assert context["status_code"] == 400
2206 assert context["response_headers"]["X-WNS-Status"] == "dropped"
2209def test_web_push_success_records_body_verbatim(db):
2210 """response holds the body exactly as received, never a wrapper around it."""
2211 user, _ = generate_user()
2212 sub_id = _make_web_push_sub(user.id)
2214 with patch("couchers.notifications.send_raw_push_notification.send_web_push") as mock_send:
2215 mock_send.return_value = Mock(status_code=201, text="ok", headers={"X-WNS-Status": "received"})
2216 send_raw_push_notification_v2(
2217 jobs_pb2.SendRawPushNotificationPayloadV2(
2218 push_notification_subscription_id=sub_id,
2219 title="title",
2220 body="body",
2221 )
2222 )
2224 with session_scope() as session:
2225 attempt = session.execute(
2226 select(PushNotificationDeliveryAttempt).where(
2227 PushNotificationDeliveryAttempt.push_notification_subscription_id == sub_id
2228 )
2229 ).scalar_one()
2230 assert attempt.outcome == PushNotificationDeliveryOutcome.success
2231 assert attempt.response == "ok"