Coverage for app/backend/src/tests/test_notifications.py: 99%
799 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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.settings import get_topic_actions_by_delivery_type, reset_preference
38from couchers.proto import (
39 api_pb2,
40 auth_pb2,
41 conversations_pb2,
42 editor_pb2,
43 events_pb2,
44 notification_data_pb2,
45 notifications_pb2,
46)
47from couchers.proto.internal import jobs_pb2, unsubscribe_pb2
48from couchers.servicers.api import user_model_to_pb
49from couchers.utils import not_none, now
50from tests.fixtures.db import generate_user
51from tests.fixtures.misc import EmailCollector, PushCollector, process_jobs
52from tests.fixtures.sessions import (
53 api_session,
54 auth_api_session,
55 conversations_session,
56 notifications_session,
57 real_editor_session,
58)
61@pytest.fixture(autouse=True)
62def _(testconfig):
63 pass
66@pytest.mark.parametrize("enabled", [True, False])
67def test_SetNotificationSettings_preferences_respected_editable(db, enabled):
68 user, token = generate_user()
70 # enable a notification type and check it gets delivered
71 topic_action = NotificationTopicAction.badge__add
73 with notifications_session(token) as notifications:
74 notifications.SetNotificationSettings(
75 notifications_pb2.SetNotificationSettingsReq(
76 preferences=[
77 notifications_pb2.SingleNotificationPreference(
78 topic=topic_action.topic,
79 action=topic_action.action,
80 delivery_method="push",
81 enabled=enabled,
82 )
83 ],
84 )
85 )
87 with session_scope() as session:
88 notify(
89 session,
90 user_id=user.id,
91 topic_action=topic_action,
92 key="",
93 data=notification_data_pb2.BadgeAdd(
94 badge_id="volunteer",
95 badge_name="Active Volunteer",
96 badge_description="This user is an active volunteer for Couchers.org",
97 ),
98 )
100 process_job()
102 with session_scope() as session:
103 deliv = session.execute(
104 select(NotificationDelivery)
105 .join(Notification, Notification.id == NotificationDelivery.notification_id)
106 .where(Notification.user_id == user.id)
107 .where(Notification.topic_action == topic_action)
108 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
109 ).scalar_one_or_none()
111 if enabled:
112 assert deliv is not None
113 else:
114 assert deliv is None
117def test_SetNotificationSettings_preferences_not_editable(db):
118 user, token = generate_user()
120 # enable a notification type and check it gets delivered
121 topic_action = NotificationTopicAction.password_reset__start
123 with notifications_session(token) as notifications:
124 with pytest.raises(grpc.RpcError) as e:
125 notifications.SetNotificationSettings(
126 notifications_pb2.SetNotificationSettingsReq(
127 preferences=[
128 notifications_pb2.SingleNotificationPreference(
129 topic=topic_action.topic,
130 action=topic_action.action,
131 delivery_method="push",
132 enabled=False,
133 )
134 ],
135 )
136 )
137 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
138 assert e.value.details() == "That notification preference is not user editable."
141def test_unsubscribe(db, email_collector: EmailCollector):
142 # this is the ugliest test i've written
144 user, token = generate_user()
146 topic_action = NotificationTopicAction.badge__add
148 # first enable email notifs
149 with notifications_session(token) as notifications:
150 notifications.SetNotificationSettings(
151 notifications_pb2.SetNotificationSettingsReq(
152 preferences=[
153 notifications_pb2.SingleNotificationPreference(
154 topic=topic_action.topic,
155 action=topic_action.action,
156 delivery_method=method,
157 enabled=enabled,
158 )
159 for method, enabled in [("email", True), ("digest", False), ("push", False)]
160 ],
161 )
162 )
164 with session_scope() as session:
165 notify(
166 session,
167 user_id=user.id,
168 topic_action=topic_action,
169 key="",
170 data=notification_data_pb2.BadgeAdd(
171 badge_id="volunteer",
172 badge_name="Active Volunteer",
173 badge_description="This user is an active volunteer for Couchers.org",
174 ),
175 )
177 email = email_collector.pop_for_recipient(user.email, last=True)
179 # very ugly
180 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
181 for link in re.findall(r'<a href="(.*?)"', email.html): 181 ↛ 202line 181 didn't jump to line 202 because the loop on line 181 didn't complete
182 if "payload" not in link:
183 continue
184 print(link)
185 url_parts = urlparse(html.unescape(link))
186 params = parse_qs(url_parts.query)
187 print(params["payload"][0])
188 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
189 if payload.HasField("topic_action"): 189 ↛ 181line 189 didn't jump to line 181 because the condition on line 189 was always true
190 with auth_api_session() as (auth_api, metadata_interceptor):
191 assert (
192 auth_api.Unsubscribe(
193 auth_pb2.UnsubscribeReq(
194 payload=b64decode(params["payload"][0]),
195 sig=b64decode(params["sig"][0]),
196 )
197 ).response
198 == "You've been unsubscribed from email notifications of that type."
199 )
200 break
201 else:
202 raise Exception("Didn't find link")
204 with notifications_session(token) as notifications:
205 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
207 for group in res.groups:
208 for topic in group.topics:
209 for item in topic.items:
210 if topic == topic_action.topic and item == topic_action.action: 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true
211 assert not item.email
213 with session_scope() as session:
214 notify(
215 session,
216 user_id=user.id,
217 topic_action=topic_action,
218 key="",
219 data=notification_data_pb2.BadgeAdd(
220 badge_id="volunteer",
221 badge_name="Active Volunteer",
222 badge_description="This user is an active volunteer for Couchers.org",
223 ),
224 )
226 assert email_collector.count_for_recipient(user.email) == 0
229def test_unsubscribe_do_not_email(db, email_collector: EmailCollector, moderator):
230 user, token = generate_user()
232 _, token2 = generate_user(complete_profile=True)
233 with api_session(token2) as api:
234 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
235 res = api.ListFriendRequests(empty_pb2.Empty())
236 fr_id = res.sent[0].friend_request_id
238 # Moderator approves the friend request, which triggers the notification email
239 moderator.approve_friend_request(fr_id)
241 email = email_collector.pop_for_recipient(user.email, last=True)
242 assert email.recipient == user.email
243 # very ugly
244 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
245 for link in re.findall(r'<a href="(.*?)"', email.html): 245 ↛ 266line 245 didn't jump to line 266 because the loop on line 245 didn't complete
246 if "payload" not in link:
247 continue
248 print(link)
249 url_parts = urlparse(html.unescape(link))
250 params = parse_qs(url_parts.query)
251 print(params["payload"][0])
252 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
253 if payload.HasField("do_not_email"):
254 with auth_api_session() as (auth_api, metadata_interceptor):
255 assert (
256 auth_api.Unsubscribe(
257 auth_pb2.UnsubscribeReq(
258 payload=b64decode(params["payload"][0]),
259 sig=b64decode(params["sig"][0]),
260 )
261 ).response
262 == "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."
263 )
264 break
265 else:
266 raise Exception("Didn't find link")
268 _, token3 = generate_user(complete_profile=True)
269 with api_session(token3) as api:
270 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
271 res = api.ListFriendRequests(empty_pb2.Empty())
272 fr_id3 = res.sent[0].friend_request_id
274 # Approving this friend request should NOT send an email since user has do_not_email set
275 moderator.approve_friend_request(fr_id3)
277 assert email_collector.count_for_recipient(user.email) == 0
279 with session_scope() as session:
280 user_ = session.execute(select(User).where(User.id == user.id)).scalar_one()
281 assert user_.do_not_email
284def test_get_do_not_email(db):
285 _, token = generate_user()
287 with session_scope() as session:
288 user = session.execute(select(User)).scalar_one()
289 user.do_not_email = False
291 with notifications_session(token) as notifications:
292 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
293 assert not res.do_not_email_enabled
295 with session_scope() as session:
296 user = session.execute(select(User)).scalar_one()
297 user.do_not_email = True
298 user.hosting_status = HostingStatus.cant_host
299 user.meetup_status = MeetupStatus.does_not_want_to_meetup
301 with notifications_session(token) as notifications:
302 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
303 assert res.do_not_email_enabled
306def test_set_do_not_email(db):
307 _, token = generate_user()
309 with session_scope() as session:
310 user = session.execute(select(User)).scalar_one()
311 user.do_not_email = False
312 user.hosting_status = HostingStatus.can_host
313 user.meetup_status = MeetupStatus.wants_to_meetup
315 with notifications_session(token) as notifications:
316 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
318 with session_scope() as session:
319 user = session.execute(select(User)).scalar_one()
320 assert not user.do_not_email
322 with notifications_session(token) as notifications:
323 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
325 with session_scope() as session:
326 user = session.execute(select(User)).scalar_one()
327 assert user.do_not_email
328 assert user.hosting_status == HostingStatus.cant_host
329 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup
331 with notifications_session(token) as notifications:
332 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
334 with session_scope() as session:
335 user = session.execute(select(User)).scalar_one()
336 assert not user.do_not_email
339def test_list_notifications(db, push_collector: PushCollector, moderator):
340 user1, token1 = generate_user()
341 user2, token2 = generate_user()
343 with api_session(token2) as api:
344 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
345 res = api.ListFriendRequests(empty_pb2.Empty())
346 fr_id = res.sent[0].friend_request_id
348 # Moderator approves the friend request so the notification is sent
349 moderator.approve_friend_request(fr_id)
351 with notifications_session(token1) as notifications:
352 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
353 assert len(res.notifications) == 1
355 n = res.notifications[0]
357 assert n.topic == "friend_request"
358 assert n.action == "create"
359 assert n.key == str(user2.id)
360 assert n.title == f"Friend request from {user2.name}"
361 assert n.body == f"{user2.name} wants to be your friend."
362 assert n.icon.startswith("http://localhost:5001/img/thumbnail/")
363 assert n.url == f"http://localhost:3000/connections/friends/?from={user2.id}"
365 with conversations_session(token2) as c:
366 res = c.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[user1.id]))
367 group_chat_id = res.group_chat_id
368 moderator.approve_group_chat(group_chat_id)
369 for i in range(17):
370 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text=f"Test message {i}"))
372 process_jobs()
374 all_notifs = []
375 with notifications_session(token1) as notifications:
376 page_token = None
377 for _ in range(100): 377 ↛ 390line 377 didn't jump to line 390
378 res = notifications.ListNotifications(
379 notifications_pb2.ListNotificationsReq(
380 page_size=5,
381 page_token=page_token,
382 )
383 )
384 assert len(res.notifications) == 5 or not res.next_page_token
385 all_notifs += res.notifications
386 page_token = res.next_page_token
387 if not page_token:
388 break
390 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"{user2.name} wants to be your friend."]
391 assert bodys == [n.body for n in all_notifs]
394def test_notifications_seen(db, push_collector: PushCollector, moderator):
395 user1, token1 = generate_user()
396 user2, token2 = generate_user()
397 user3, token3 = generate_user()
398 user4, token4 = generate_user()
400 with api_session(token2) as api:
401 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
402 res = api.ListFriendRequests(empty_pb2.Empty())
403 fr_id2 = res.sent[0].friend_request_id
405 with api_session(token3) as api:
406 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
407 res = api.ListFriendRequests(empty_pb2.Empty())
408 fr_id3 = res.sent[0].friend_request_id
410 # Moderator approves the friend requests so notifications are sent
411 moderator.approve_friend_request(fr_id2)
412 moderator.approve_friend_request(fr_id3)
414 with notifications_session(token1) as notifications, api_session(token1) as api:
415 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
416 assert len(res.notifications) == 2
417 assert [n.is_seen for n in res.notifications] == [False, False]
418 notification_ids = [n.notification_id for n in res.notifications]
419 # should be listed desc time
420 assert notification_ids[0] > notification_ids[1]
422 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
424 with api_session(token4) as api:
425 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
426 res = api.ListFriendRequests(empty_pb2.Empty())
427 fr_id4 = res.sent[0].friend_request_id
429 # Moderator approves the friend request so notification is sent
430 moderator.approve_friend_request(fr_id4)
432 with notifications_session(token1) as notifications, api_session(token1) as api:
433 # mark everything before just the last one as seen (pretend we didn't load the last one yet in the api)
434 notifications.MarkAllNotificationsSeen(
435 notifications_pb2.MarkAllNotificationsSeenReq(latest_notification_id=notification_ids[0])
436 )
438 # last one is still unseen
439 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
441 # mark the first one unseen
442 notifications.MarkNotificationSeen(
443 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids[1], set_seen=False)
444 )
445 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
447 # mark the last one seen
448 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
449 assert len(res.notifications) == 3
450 assert [n.is_seen for n in res.notifications] == [False, True, False]
451 notification_ids2 = [n.notification_id for n in res.notifications]
453 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
455 notifications.MarkNotificationSeen(
456 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids2[0], set_seen=True)
457 )
459 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
460 assert len(res.notifications) == 3
461 assert [n.is_seen for n in res.notifications] == [True, True, False]
463 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
466def test_unseen_notification_count_excludes_ums_hidden(db, moderator):
467 user1, token1 = generate_user()
468 user2, token2 = generate_user()
470 with api_session(token2) as api:
471 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
472 res = api.ListFriendRequests(empty_pb2.Empty())
473 fr_id = res.sent[0].friend_request_id
475 # Before moderation the friend request is shadowed, so the resulting notification
476 # is not visible to the recipient and must not contribute to their unseen count.
477 with api_session(token1) as api:
478 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 0
480 moderator.approve_friend_request(fr_id)
482 with api_session(token1) as api:
483 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
486def test_GetVapidPublicKey(db):
487 _, token = generate_user()
489 with notifications_session(token) as notifications:
490 assert (
491 notifications.GetVapidPublicKey(empty_pb2.Empty()).vapid_public_key
492 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"
493 )
496def test_RegisterPushNotificationSubscription(db):
497 _, token = generate_user()
499 subscription_info = {
500 "endpoint": "https://updates.push.services.mozilla.com/wpush/v2/gAAAAABmW2_iYKVyZRJPhAhktbkXd6Bc8zjIUvtVi5diYL7ZYn8FHka94kIdF46Mp8DwCDWlACnbKOEo97ikaa7JYowGLiGz3qsWL7Vo19LaV4I71mUDUOIKxWIsfp_kM77MlRJQKDUddv-sYyiffOyg63d1lnc_BMIyLXt69T5SEpfnfWTNb6I",
501 "expirationTime": None,
502 "keys": {
503 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
504 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
505 },
506 }
508 with notifications_session(token) as notifications:
509 res = notifications.RegisterPushNotificationSubscription(
510 notifications_pb2.RegisterPushNotificationSubscriptionReq(
511 full_subscription_json=json.dumps(subscription_info),
512 )
513 )
516def test_RegisterPushNotificationSubscription_invalid_endpoint(db):
517 _, token = generate_user()
519 subscription_info = {
520 "endpoint": "https://permanently-removed.invalid/some-id",
521 "expirationTime": None,
522 "keys": {
523 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
524 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
525 },
526 }
528 with notifications_session(token) as notifications:
529 with pytest.raises(grpc.RpcError) as e:
530 notifications.RegisterPushNotificationSubscription(
531 notifications_pb2.RegisterPushNotificationSubscriptionReq(
532 full_subscription_json=json.dumps(subscription_info),
533 )
534 )
535 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
538def test_SendTestPushNotification(db, push_collector: PushCollector):
539 user, token = generate_user()
541 with notifications_session(token) as notifications:
542 notifications.SendTestPushNotification(empty_pb2.Empty())
544 assert push_collector.count_for_user(user.id) == 1
545 push = push_collector.pop_for_user(user.id, last=True)
546 assert push.content.title == "Push notifications test"
547 assert push.content.body == "If you see this, then it's working :)"
550def test_SendBlogPostNotification(db, email_collector: EmailCollector, push_collector: PushCollector):
551 super_user, super_token = generate_user(is_superuser=True)
553 user1, user1_token = generate_user()
554 # enabled email
555 user2, user2_token = generate_user()
556 # disabled push
557 user3, user3_token = generate_user()
559 topic_action = NotificationTopicAction.general__new_blog_post
561 with notifications_session(user2_token) as notifications:
562 notifications.SetNotificationSettings(
563 notifications_pb2.SetNotificationSettingsReq(
564 preferences=[
565 notifications_pb2.SingleNotificationPreference(
566 topic=topic_action.topic,
567 action=topic_action.action,
568 delivery_method="email",
569 enabled=True,
570 )
571 ],
572 )
573 )
575 with notifications_session(user3_token) as notifications:
576 notifications.SetNotificationSettings(
577 notifications_pb2.SetNotificationSettingsReq(
578 preferences=[
579 notifications_pb2.SingleNotificationPreference(
580 topic=topic_action.topic,
581 action=topic_action.action,
582 delivery_method="push",
583 enabled=False,
584 )
585 ],
586 )
587 )
589 with real_editor_session(super_token) as editor_api:
590 editor_api.SendBlogPostNotification(
591 editor_pb2.SendBlogPostNotificationReq(
592 title="Couchers.org v0.9.9 Release Notes",
593 blurb="Read about last major updates before v1!",
594 url="https://couchers.org/blog/2025/05/11/v0.9.9-release",
595 )
596 )
598 email = email_collector.pop_for_recipient(user2.email, last=True)
599 assert email.recipient == user2.email
600 assert "Couchers.org v0.9.9 Release Notes" in email.html
601 assert "Couchers.org v0.9.9 Release Notes" in email.plain
602 assert "Read about last major updates before v1!" in email.html
603 assert "Read about last major updates before v1!" in email.plain
604 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.html
605 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email.plain
607 push = push_collector.pop_for_user(user1.id, last=True)
608 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
609 assert push.content.body == "Read about last major updates before v1!"
610 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
612 push = push_collector.pop_for_user(user2.id, last=True)
613 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
614 assert push.content.body == "Read about last major updates before v1!"
615 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
617 assert push_collector.count_for_user(user3.id) == 0
620def test_get_topic_actions_by_delivery_type(db):
621 user, token = generate_user()
623 # these are enabled by default
624 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults
625 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults
627 # these are disabled by default
628 assert NotificationDeliveryType.push not in NotificationTopicAction.event__create_any.defaults
629 assert NotificationDeliveryType.push not in NotificationTopicAction.discussion__create.defaults
631 with notifications_session(token) as notifications:
632 notifications.SetNotificationSettings(
633 notifications_pb2.SetNotificationSettingsReq(
634 preferences=[
635 notifications_pb2.SingleNotificationPreference(
636 topic=NotificationTopicAction.reference__receive_friend.topic,
637 action=NotificationTopicAction.reference__receive_friend.action,
638 delivery_method="push",
639 enabled=False,
640 ),
641 notifications_pb2.SingleNotificationPreference(
642 topic=NotificationTopicAction.event__create_any.topic,
643 action=NotificationTopicAction.event__create_any.action,
644 delivery_method="push",
645 enabled=True,
646 ),
647 ],
648 )
649 )
651 with session_scope() as session:
652 deliver = get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push)
653 assert NotificationTopicAction.reference__receive_friend not in deliver
654 assert NotificationTopicAction.host_request__accept in deliver
655 assert NotificationTopicAction.event__create_any in deliver
656 assert NotificationTopicAction.discussion__create not in deliver
657 assert NotificationTopicAction.account_deletion__start in deliver
660def test_reset_preference(db):
661 user, token = generate_user()
663 topic_action = NotificationTopicAction.event__create_any
664 # neither delivery type is on by default, so enabling both leaves two overrides to tell apart
665 assert NotificationDeliveryType.push not in topic_action.defaults
666 assert NotificationDeliveryType.email not in topic_action.defaults
668 def get_setting() -> notifications_pb2.NotificationItem:
669 with notifications_session(token) as notifications:
670 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
671 items: list[notifications_pb2.NotificationItem] = [
672 item
673 for group in res.groups
674 for topic in group.topics
675 if topic.topic == topic_action.topic
676 for item in topic.items
677 if item.action == topic_action.action
678 ]
679 (item,) = items
680 return item
682 with notifications_session(token) as notifications:
683 notifications.SetNotificationSettings(
684 notifications_pb2.SetNotificationSettingsReq(
685 preferences=[
686 notifications_pb2.SingleNotificationPreference(
687 topic=topic_action.topic,
688 action=topic_action.action,
689 delivery_method=delivery_method,
690 enabled=True,
691 )
692 for delivery_method in ["push", "email"]
693 ]
694 )
695 )
697 assert get_setting().push
698 assert get_setting().email
700 # there is no API for clearing an override back to its default, it's only reachable internally
701 with session_scope() as session:
702 reset_preference(session, user.id, topic_action, NotificationDeliveryType.push)
704 # only the push override should go, leaving the email one in place
705 assert not get_setting().push
706 assert get_setting().email
709def test_event_reminder_email_sent(db, email_collector: EmailCollector):
710 user, token = generate_user()
711 title = "Board Game Night"
713 # Saturday, July 5, 2025 at 4:40:00 AM UTC
714 start_time = timestamp_pb2.Timestamp(seconds=1751690400)
715 timezone = "Etc/GMT-2" # Etc/GMT-2 means GMT+2
716 expected_time_str = "Saturday, July 5, 6:40 AM"
718 with session_scope() as session:
719 user_in_session = session.get_one(User, user.id)
721 notify(
722 session,
723 user_id=user.id,
724 topic_action=NotificationTopicAction.event__reminder,
725 key="",
726 data=notification_data_pb2.EventReminder(
727 event=events_pb2.Event(
728 event_id=1, slug="board-game-night", title=title, start_time=start_time, timezone=timezone
729 ),
730 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)),
731 ),
732 )
734 email = email_collector.pop_for_recipient(user.email, last=True)
735 assert email.recipient == user.email
736 assert title in email.html
737 assert title in email.plain
738 assert expected_time_str in email.html
739 assert expected_time_str in email.plain
742def test_RegisterMobilePushNotificationSubscription(db):
743 user, token = generate_user()
745 with notifications_session(token) as notifications:
746 notifications.RegisterMobilePushNotificationSubscription(
747 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
748 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
749 device_name="My iPhone",
750 device_type="ios",
751 )
752 )
754 # Check subscription was created
755 with session_scope() as session:
756 sub = session.execute(
757 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
758 ).scalar_one()
759 assert sub.platform == PushNotificationPlatform.expo
760 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]"
761 assert sub.device_name == "My iPhone"
762 assert sub.device_type == DeviceType.ios
763 assert sub.disabled_at == DATETIME_INFINITY
766def test_RegisterMobilePushNotificationSubscription_android(db):
767 user, token = generate_user()
769 with notifications_session(token) as notifications:
770 notifications.RegisterMobilePushNotificationSubscription(
771 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
772 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
773 device_name="My Android",
774 device_type="android",
775 )
776 )
778 with session_scope() as session:
779 sub = session.execute(
780 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
781 ).scalar_one()
782 assert sub.platform == PushNotificationPlatform.expo
783 assert sub.device_type == DeviceType.android
786def test_RegisterMobilePushNotificationSubscription_no_device_type(db):
787 user, token = generate_user()
789 with notifications_session(token) as notifications:
790 notifications.RegisterMobilePushNotificationSubscription(
791 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
792 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]",
793 )
794 )
796 with session_scope() as session:
797 sub = session.execute(
798 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
799 ).scalar_one()
800 assert sub.platform == PushNotificationPlatform.expo
801 assert sub.device_name is None
802 assert sub.device_type is None
805def test_RegisterMobilePushNotificationSubscription_re_enable(db):
806 user, token = generate_user()
808 # Create a disabled subscription directly in the DB
809 with session_scope() as session:
810 sub = PushNotificationSubscription(
811 user_id=user.id,
812 platform=PushNotificationPlatform.expo,
813 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
814 device_name="Old Device",
815 device_type=DeviceType.ios,
816 )
817 sub.disabled_at = now()
818 session.add(sub)
819 session.flush()
820 sub_id = sub.id
822 # Re-register with the same token
823 with notifications_session(token) as notifications:
824 notifications.RegisterMobilePushNotificationSubscription(
825 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
826 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
827 device_name="New Device Name",
828 device_type="android",
829 )
830 )
832 # Check subscription was re-enabled and updated
833 with session_scope() as session:
834 sub = session.execute(
835 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
836 ).scalar_one()
837 assert sub.disabled_at == DATETIME_INFINITY
838 assert sub.device_name == "New Device Name"
839 assert sub.device_type == DeviceType.android
842def test_RegisterMobilePushNotificationSubscription_already_exists(db):
843 user, token = generate_user()
845 # Create an active subscription directly in the DB
846 with session_scope() as session:
847 sub = PushNotificationSubscription(
848 user_id=user.id,
849 platform=PushNotificationPlatform.expo,
850 token="ExponentPushToken[existingtoken]",
851 device_name="Existing Device",
852 device_type=DeviceType.ios,
853 )
854 session.add(sub)
856 # Try to register with the same token - should just return without error
857 with notifications_session(token) as notifications:
858 notifications.RegisterMobilePushNotificationSubscription(
859 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
860 token="ExponentPushToken[existingtoken]",
861 device_name="Different Name",
862 )
863 )
865 # Check subscription was NOT modified (already active)
866 with session_scope() as session:
867 sub = session.execute(
868 select(PushNotificationSubscription).where(
869 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]"
870 )
871 ).scalar_one()
872 assert sub.device_name == "Existing Device" # unchanged
875def test_SendTestMobilePushNotification(db, push_collector: PushCollector):
876 user, token = generate_user()
878 with notifications_session(token) as notifications:
879 notifications.SendTestMobilePushNotification(empty_pb2.Empty())
881 push = push_collector.pop_for_user(user.id, last=True)
882 assert push.content.title == "Mobile notifications test"
883 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉"
886def test_get_expo_push_receipts(db):
887 mock_response = Mock()
888 mock_response.status_code = 200
889 mock_response.json.return_value = {
890 "data": {
891 "ticket-1": {"status": "ok"},
892 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
893 }
894 }
896 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post:
897 result = get_expo_push_receipts(["ticket-1", "ticket-2"])
899 mock_post.assert_called_once()
900 call_args = mock_post.call_args
901 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts"
902 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}
904 assert result == {
905 "ticket-1": {"status": "ok"},
906 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
907 }
910def test_get_expo_push_receipts_empty(db):
911 result = get_expo_push_receipts([])
912 assert result == {}
915def test_check_expo_push_receipts_success(db):
916 """Test batch receipt checking with successful delivery."""
917 user, token = generate_user()
919 # Create a push subscription and delivery attempt (old enough to be checked)
920 with session_scope() as session:
921 sub = PushNotificationSubscription(
922 user_id=user.id,
923 platform=PushNotificationPlatform.expo,
924 token="ExponentPushToken[testtoken123]",
925 device_name="Test Device",
926 device_type=DeviceType.ios,
927 )
928 session.add(sub)
929 session.flush()
931 attempt = PushNotificationDeliveryAttempt(
932 push_notification_subscription_id=sub.id,
933 outcome=PushNotificationDeliveryOutcome.success,
934 status_code=200,
935 expo_ticket_id="test-ticket-id",
936 )
937 session.add(attempt)
938 session.flush()
939 # Make the attempt old enough to be checked (>15 min)
940 attempt.time = now() - timedelta(minutes=20)
941 attempt_id = attempt.id
942 sub_id = sub.id
944 # Mock the receipt API call
945 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
946 mock_post.return_value.status_code = 200
947 mock_post.return_value.json.return_value = {"data": {"test-ticket-id": {"status": "ok"}}}
949 check_expo_push_receipts(empty_pb2.Empty())
951 # Verify the attempt was updated
952 with session_scope() as session:
953 attempt = session.execute(
954 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
955 ).scalar_one()
956 assert attempt.receipt_checked_at is not None
957 assert attempt.receipt_status == "ok"
958 assert attempt.receipt_error_code is None
960 # Subscription should still be enabled
961 sub = session.execute(
962 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
963 ).scalar_one()
964 assert sub.disabled_at == DATETIME_INFINITY
967def test_check_expo_push_receipts_device_not_registered(db):
968 """Test batch receipt checking with DeviceNotRegistered error disables subscription."""
969 user, token = generate_user()
971 # Create a push subscription and delivery attempt
972 with session_scope() as session:
973 sub = PushNotificationSubscription(
974 user_id=user.id,
975 platform=PushNotificationPlatform.expo,
976 token="ExponentPushToken[devicegone]",
977 device_name="Test Device",
978 device_type=DeviceType.android,
979 )
980 session.add(sub)
981 session.flush()
983 attempt = PushNotificationDeliveryAttempt(
984 push_notification_subscription_id=sub.id,
985 outcome=PushNotificationDeliveryOutcome.success,
986 status_code=200,
987 expo_ticket_id="ticket-device-gone",
988 )
989 session.add(attempt)
990 session.flush()
991 # Make the attempt old enough to be checked
992 attempt.time = now() - timedelta(minutes=15)
993 attempt_id = attempt.id
994 sub_id = sub.id
996 # Mock the receipt API call with DeviceNotRegistered error
997 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
998 mock_post.return_value.status_code = 200
999 mock_post.return_value.json.return_value = {
1000 "data": {
1001 "ticket-device-gone": {
1002 "status": "error",
1003 "details": {"error": "DeviceNotRegistered"},
1004 }
1005 }
1006 }
1008 check_expo_push_receipts(empty_pb2.Empty())
1010 # Verify the attempt was updated and subscription disabled
1011 with session_scope() as session:
1012 attempt = session.execute(
1013 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1014 ).scalar_one()
1015 assert attempt.receipt_checked_at is not None
1016 assert attempt.receipt_status == "error"
1017 assert attempt.receipt_error_code == "DeviceNotRegistered"
1019 # Subscription should be disabled
1020 sub = session.execute(
1021 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1022 ).scalar_one()
1023 assert sub.disabled_at <= now()
1026def test_check_expo_push_receipts_not_found(db):
1027 """Test batch receipt checking when ticket not found (expired)."""
1028 user, token = generate_user()
1030 with session_scope() as session:
1031 sub = PushNotificationSubscription(
1032 user_id=user.id,
1033 platform=PushNotificationPlatform.expo,
1034 token="ExponentPushToken[notfound]",
1035 )
1036 session.add(sub)
1037 session.flush()
1039 attempt = PushNotificationDeliveryAttempt(
1040 push_notification_subscription_id=sub.id,
1041 outcome=PushNotificationDeliveryOutcome.success,
1042 status_code=200,
1043 expo_ticket_id="unknown-ticket",
1044 )
1045 session.add(attempt)
1046 session.flush()
1047 # Make the attempt old enough to be checked
1048 attempt.time = now() - timedelta(minutes=15)
1049 attempt_id = attempt.id
1050 sub_id = sub.id
1052 # Mock empty receipt response (ticket not found)
1053 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1054 mock_post.return_value.status_code = 200
1055 mock_post.return_value.json.return_value = {"data": {}}
1057 check_expo_push_receipts(empty_pb2.Empty())
1059 with session_scope() as session:
1060 attempt = session.execute(
1061 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1062 ).scalar_one()
1063 assert attempt.receipt_checked_at is not None
1064 assert attempt.receipt_status == "not_found"
1066 # Subscription should still be enabled
1067 sub = session.execute(
1068 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1069 ).scalar_one()
1070 assert sub.disabled_at == DATETIME_INFINITY
1073def test_check_expo_push_receipts_skips_already_checked(db):
1074 """Test that already-checked receipts are not re-checked."""
1075 user, token = generate_user()
1077 # Create an attempt that was already checked
1078 with session_scope() as session:
1079 sub = PushNotificationSubscription(
1080 user_id=user.id,
1081 platform=PushNotificationPlatform.expo,
1082 token="ExponentPushToken[alreadychecked]",
1083 )
1084 session.add(sub)
1085 session.flush()
1087 attempt = PushNotificationDeliveryAttempt(
1088 push_notification_subscription_id=sub.id,
1089 outcome=PushNotificationDeliveryOutcome.success,
1090 status_code=200,
1091 expo_ticket_id="already-checked-ticket",
1092 receipt_checked_at=now(),
1093 receipt_status="ok",
1094 )
1095 session.add(attempt)
1096 session.flush()
1097 # Make the attempt old enough
1098 attempt.time = now() - timedelta(minutes=15)
1100 # Should not call the API since the only attempt is already checked
1101 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1102 check_expo_push_receipts(empty_pb2.Empty())
1103 mock_post.assert_not_called()
1106def test_SendDevPushNotification_success(db, push_collector: PushCollector):
1107 """Test SendDevPushNotification sends push with all specified parameters."""
1108 user, token = generate_user()
1110 # Enable dev APIs for this test
1111 config.ENABLE_DEV_APIS = True
1113 with notifications_session(token) as notifications:
1114 notifications.SendDevPushNotification(
1115 notifications_pb2.SendDevPushNotificationReq(
1116 title="Test Dev Title",
1117 body="Test dev notification body",
1118 icon="https://example.com/icon.png",
1119 url="https://example.com/action",
1120 key="test-key",
1121 ttl=3600,
1122 )
1123 )
1125 push = push_collector.pop_for_user(user.id, last=True)
1126 assert push.content.title == "Test Dev Title"
1127 assert push.content.body == "Test dev notification body"
1128 assert push.content.action_url == "https://example.com/action"
1129 assert push.content.icon_url == "https://example.com/icon.png"
1130 assert push.topic_action == "adhoc:testing"
1131 assert push.key == "test-key"
1132 assert push.ttl == 3600
1135def test_SendDevPushNotification_minimal(db, push_collector: PushCollector):
1136 """Test SendDevPushNotification with minimal parameters."""
1137 user, token = generate_user()
1139 config.ENABLE_DEV_APIS = True
1141 with notifications_session(token) as notifications:
1142 notifications.SendDevPushNotification(
1143 notifications_pb2.SendDevPushNotificationReq(
1144 title="Minimal Title",
1145 body="Minimal body",
1146 )
1147 )
1149 push = push_collector.pop_for_user(user.id, last=True)
1150 assert push.content.title == "Minimal Title"
1151 assert push.content.body == "Minimal body"
1152 assert push.topic_action == "adhoc:testing"
1155def test_SendDevPushNotification_disabled(db, push_collector: PushCollector):
1156 """Test SendDevPushNotification fails when ENABLE_DEV_APIS is disabled."""
1157 user, token = generate_user()
1159 # Ensure dev APIs are disabled (default in tests)
1160 config.ENABLE_DEV_APIS = False
1162 with notifications_session(token) as notifications:
1163 with pytest.raises(grpc.RpcError) as e:
1164 notifications.SendDevPushNotification(
1165 notifications_pb2.SendDevPushNotificationReq(
1166 title="Should Fail",
1167 body="This should not be sent",
1168 )
1169 )
1170 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1171 assert "Development APIs are not enabled" in not_none(e.value.details())
1173 assert push_collector.count_for_user(user.id) == 0
1176def test_SendDevPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1177 """Test SendDevPushNotification fails when push notifications are disabled."""
1178 user, token = generate_user()
1180 config.ENABLE_DEV_APIS = True
1181 config.PUSH_NOTIFICATIONS_ENABLED = False
1183 with notifications_session(token) as notifications:
1184 with pytest.raises(grpc.RpcError) as e:
1185 notifications.SendDevPushNotification(
1186 notifications_pb2.SendDevPushNotificationReq(
1187 title="Should Fail",
1188 body="This should not be sent",
1189 )
1190 )
1191 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1192 assert "Push notifications are currently disabled" in not_none(e.value.details())
1194 assert push_collector.count_for_user(user.id) == 0
1197def test_check_expo_push_receipts_skips_too_recent(db):
1198 """Test that too-recent receipts (<15 min) are not checked."""
1199 user, token = generate_user()
1201 # Create a recent attempt (not old enough to check)
1202 with session_scope() as session:
1203 sub = PushNotificationSubscription(
1204 user_id=user.id,
1205 platform=PushNotificationPlatform.expo,
1206 token="ExponentPushToken[recent]",
1207 )
1208 session.add(sub)
1209 session.flush()
1211 attempt = PushNotificationDeliveryAttempt(
1212 push_notification_subscription_id=sub.id,
1213 outcome=PushNotificationDeliveryOutcome.success,
1214 status_code=200,
1215 expo_ticket_id="recent-ticket",
1216 )
1217 session.add(attempt)
1218 session.flush()
1219 # Make the attempt only 5 minutes old (too recent)
1220 attempt.time = now() - timedelta(minutes=5)
1222 # Should not call the API since the attempt is too recent
1223 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1224 check_expo_push_receipts(empty_pb2.Empty())
1225 mock_post.assert_not_called()
1228def test_check_expo_push_receipts_batch(db):
1229 """Test that multiple receipts are checked in a single batch."""
1230 user, token = generate_user()
1232 # Create multiple delivery attempts
1233 attempt_ids = []
1234 with session_scope() as session:
1235 sub = PushNotificationSubscription(
1236 user_id=user.id,
1237 platform=PushNotificationPlatform.expo,
1238 token="ExponentPushToken[batch]",
1239 )
1240 session.add(sub)
1241 session.flush()
1243 for i in range(3):
1244 attempt = PushNotificationDeliveryAttempt(
1245 push_notification_subscription_id=sub.id,
1246 outcome=PushNotificationDeliveryOutcome.success,
1247 status_code=200,
1248 expo_ticket_id=f"batch-ticket-{i}",
1249 )
1250 session.add(attempt)
1251 session.flush()
1252 attempt.time = now() - timedelta(minutes=20)
1253 attempt_ids.append(attempt.id)
1255 # Mock the batch receipt API call
1256 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1257 mock_post.return_value.status_code = 200
1258 mock_post.return_value.json.return_value = {
1259 "data": {
1260 "batch-ticket-0": {"status": "ok"},
1261 "batch-ticket-1": {"status": "ok"},
1262 "batch-ticket-2": {"status": "ok"},
1263 }
1264 }
1266 check_expo_push_receipts(empty_pb2.Empty())
1268 # Should only call the API once for all tickets
1269 assert mock_post.call_count == 1
1271 # Verify all attempts were updated
1272 with session_scope() as session:
1273 for attempt_id in attempt_ids:
1274 attempt = session.execute(
1275 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1276 ).scalar_one()
1277 assert attempt.receipt_checked_at is not None
1278 assert attempt.receipt_status == "ok"
1281def test_DebugRedeliverPushNotification_success(db, push_collector: PushCollector):
1282 """Test DebugRedeliverPushNotification redelivers an existing notification."""
1283 user, token = generate_user()
1285 config.ENABLE_DEV_APIS = True
1287 # Create a notification for the user
1288 with session_scope() as session:
1289 notify(
1290 session,
1291 user_id=user.id,
1292 topic_action=NotificationTopicAction.badge__add,
1293 key="test-badge",
1294 data=notification_data_pb2.BadgeAdd(
1295 badge_id="volunteer",
1296 badge_name="Active Volunteer",
1297 badge_description="This user is an active volunteer for Couchers.org",
1298 ),
1299 )
1301 process_job()
1303 # Pop the initial push notification
1304 push_collector.pop_for_user(user.id, last=True)
1306 # Get the notification_id
1307 with session_scope() as session:
1308 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1309 notification_id = notification.id
1311 # Redeliver the notification
1312 with notifications_session(token) as notifications:
1313 notifications.DebugRedeliverPushNotification(
1314 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1315 )
1317 # Verify a new push was sent
1318 push = push_collector.pop_for_user(user.id, last=True)
1319 assert "Active Volunteer" in push.content.title
1320 assert push.topic_action == "badge:add"
1321 assert push.key == "test-badge"
1324def test_DebugRedeliverPushNotification_not_found(db, push_collector: PushCollector):
1325 """Test DebugRedeliverPushNotification fails when notification doesn't exist."""
1326 user, token = generate_user()
1328 config.ENABLE_DEV_APIS = True
1330 with notifications_session(token) as notifications:
1331 with pytest.raises(grpc.RpcError) as e:
1332 notifications.DebugRedeliverPushNotification(
1333 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=999999)
1334 )
1335 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1336 assert "notification not found" in not_none(e.value.details()).lower()
1338 assert push_collector.count_for_user(user.id) == 0
1341def test_DebugRedeliverPushNotification_wrong_user(db, push_collector: PushCollector):
1342 """Test DebugRedeliverPushNotification fails when notification belongs to another user."""
1343 user1, token1 = generate_user()
1344 user2, token2 = generate_user()
1346 config.ENABLE_DEV_APIS = True
1348 # Create a notification for user1
1349 with session_scope() as session:
1350 notify(
1351 session,
1352 user_id=user1.id,
1353 topic_action=NotificationTopicAction.badge__add,
1354 key="test-badge",
1355 data=notification_data_pb2.BadgeAdd(
1356 badge_id="volunteer",
1357 badge_name="Active Volunteer",
1358 badge_description="This user is an active volunteer for Couchers.org",
1359 ),
1360 )
1362 process_job()
1364 # Get the notification_id
1365 with session_scope() as session:
1366 notification = session.execute(select(Notification).where(Notification.user_id == user1.id)).scalar_one()
1367 notification_id = notification.id
1369 # user2 tries to redeliver user1's notification
1370 with notifications_session(token2) as notifications:
1371 with pytest.raises(grpc.RpcError) as e:
1372 notifications.DebugRedeliverPushNotification(
1373 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1374 )
1375 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1376 assert "notification not found" in not_none(e.value.details()).lower()
1378 assert push_collector.count_for_user(user2.id) == 0
1381def test_DebugRedeliverPushNotification_disabled(db, push_collector: PushCollector):
1382 """Test DebugRedeliverPushNotification fails when ENABLE_DEV_APIS is disabled."""
1383 user, token = generate_user()
1385 config.ENABLE_DEV_APIS = False
1387 with notifications_session(token) as notifications:
1388 with pytest.raises(grpc.RpcError) as e:
1389 notifications.DebugRedeliverPushNotification(
1390 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1391 )
1392 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1393 assert "Development APIs are not enabled" in not_none(e.value.details())
1395 assert push_collector.count_for_user(user.id) == 0
1398def test_DebugRedeliverPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1399 """Test DebugRedeliverPushNotification fails when push notifications are disabled."""
1400 user, token = generate_user()
1402 config.ENABLE_DEV_APIS = True
1403 config.PUSH_NOTIFICATIONS_ENABLED = False
1405 with notifications_session(token) as notifications:
1406 with pytest.raises(grpc.RpcError) as e:
1407 notifications.DebugRedeliverPushNotification(
1408 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1409 )
1410 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1411 assert "Push notifications are currently disabled" in not_none(e.value.details())
1413 assert push_collector.count_for_user(user.id) == 0
1416def test_handle_notification_email_delivery(db, email_collector: EmailCollector):
1417 """Test that email notifications are delivered when email preference is enabled."""
1418 user, token = generate_user()
1420 topic_action = NotificationTopicAction.badge__add
1422 # Enable email notifications for this topic
1423 with notifications_session(token) as notifications:
1424 notifications.SetNotificationSettings(
1425 notifications_pb2.SetNotificationSettingsReq(
1426 preferences=[
1427 notifications_pb2.SingleNotificationPreference(
1428 topic=topic_action.topic,
1429 action=topic_action.action,
1430 delivery_method="email",
1431 enabled=True,
1432 )
1433 ],
1434 )
1435 )
1437 with session_scope() as session:
1438 notify(
1439 session,
1440 user_id=user.id,
1441 topic_action=topic_action,
1442 key="test-badge",
1443 data=notification_data_pb2.BadgeAdd(
1444 badge_id="volunteer",
1445 badge_name="Active Volunteer",
1446 badge_description="This user is an active volunteer",
1447 ),
1448 )
1450 email = email_collector.pop_for_recipient(user.email, last=True)
1451 assert email.recipient == user.email
1453 with session_scope() as session:
1454 delivery = session.execute(
1455 select(NotificationDelivery)
1456 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1457 .where(Notification.user_id == user.id)
1458 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.email)
1459 ).scalar_one()
1460 assert delivery.delivered is not None
1463def test_handle_notification_push_delivery(db, push_collector: PushCollector):
1464 """Test that push notifications are delivered immediately when push preference is enabled."""
1465 user, token = generate_user()
1467 topic_action = NotificationTopicAction.badge__add
1469 with session_scope() as session:
1470 notify(
1471 session,
1472 user_id=user.id,
1473 topic_action=topic_action,
1474 key="test-badge",
1475 data=notification_data_pb2.BadgeAdd(
1476 badge_id="volunteer",
1477 badge_name="Active Volunteer",
1478 badge_description="This user is an active volunteer",
1479 ),
1480 )
1482 process_job()
1484 push = push_collector.pop_for_user(user.id, last=True)
1485 assert "Active Volunteer" in push.content.title
1487 with session_scope() as session:
1488 delivery = session.execute(
1489 select(NotificationDelivery)
1490 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1491 .where(Notification.user_id == user.id)
1492 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1493 ).scalar_one()
1494 assert delivery.delivered is not None
1497def test_handle_notification_digest_delivery(db):
1498 """Test that digest notifications are queued without a delivered timestamp."""
1499 user, token = generate_user()
1501 topic_action = NotificationTopicAction.badge__add
1503 # Enable only digest notifications for this topic
1504 with notifications_session(token) as notifications:
1505 notifications.SetNotificationSettings(
1506 notifications_pb2.SetNotificationSettingsReq(
1507 preferences=[
1508 notifications_pb2.SingleNotificationPreference(
1509 topic=topic_action.topic,
1510 action=topic_action.action,
1511 delivery_method="push",
1512 enabled=False,
1513 ),
1514 notifications_pb2.SingleNotificationPreference(
1515 topic=topic_action.topic,
1516 action=topic_action.action,
1517 delivery_method="digest",
1518 enabled=True,
1519 ),
1520 ],
1521 )
1522 )
1524 with session_scope() as session:
1525 notify(
1526 session,
1527 user_id=user.id,
1528 topic_action=topic_action,
1529 key="test-badge",
1530 data=notification_data_pb2.BadgeAdd(
1531 badge_id="volunteer",
1532 badge_name="Active Volunteer",
1533 badge_description="This user is an active volunteer",
1534 ),
1535 )
1537 process_job()
1539 # Verify digest NotificationDelivery was created WITHOUT delivered timestamp
1540 with session_scope() as session:
1541 delivery = session.execute(
1542 select(NotificationDelivery)
1543 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1544 .where(Notification.user_id == user.id)
1545 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.digest)
1546 ).scalar_one()
1547 assert delivery.delivered is None
1550def test_handle_notification_banned_user_no_email(db, email_collector: EmailCollector):
1551 """Test that banned users don't receive email notifications."""
1552 user, token = generate_user()
1554 topic_action = NotificationTopicAction.badge__add
1556 # Enable email notifications
1557 with notifications_session(token) as notifications:
1558 notifications.SetNotificationSettings(
1559 notifications_pb2.SetNotificationSettingsReq(
1560 preferences=[
1561 notifications_pb2.SingleNotificationPreference(
1562 topic=topic_action.topic,
1563 action=topic_action.action,
1564 delivery_method="email",
1565 enabled=True,
1566 )
1567 ],
1568 )
1569 )
1571 # Ban the user
1572 with session_scope() as session:
1573 session.execute(update(User).where(User.id == user.id).values(banned_at=now()))
1575 with session_scope() as session:
1576 notify(
1577 session,
1578 user_id=user.id,
1579 topic_action=topic_action,
1580 key="test-badge",
1581 data=notification_data_pb2.BadgeAdd(
1582 badge_id="volunteer",
1583 badge_name="Active Volunteer",
1584 badge_description="This user is an active volunteer",
1585 ),
1586 )
1588 # Email should not be sent to the banned user
1589 assert email_collector.count_for_recipient(user.email) == 0
1592def test_handle_notification_deleted_user_no_regular_email(db, email_collector: EmailCollector):
1593 """Test that deleted users don't receive non-account-deletion email notifications."""
1594 user, token = generate_user()
1596 topic_action = NotificationTopicAction.badge__add
1598 # Enable email notifications
1599 with notifications_session(token) as notifications:
1600 notifications.SetNotificationSettings(
1601 notifications_pb2.SetNotificationSettingsReq(
1602 preferences=[
1603 notifications_pb2.SingleNotificationPreference(
1604 topic=topic_action.topic,
1605 action=topic_action.action,
1606 delivery_method="email",
1607 enabled=True,
1608 )
1609 ],
1610 )
1611 )
1613 # Delete the user
1614 with session_scope() as session:
1615 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1617 with session_scope() as session:
1618 notify(
1619 session,
1620 user_id=user.id,
1621 topic_action=topic_action,
1622 key="test-badge",
1623 data=notification_data_pb2.BadgeAdd(
1624 badge_id="volunteer",
1625 badge_name="Active Volunteer",
1626 badge_description="This user is an active volunteer",
1627 ),
1628 )
1630 # Email should not be sent to deleted user for non-account-deletion notification
1631 assert email_collector.count_for_recipient(user.email) == 0
1634def test_handle_notification_deleted_user_receives_account_deletion_email(db, email_collector: EmailCollector):
1635 """Test that deleted users CAN receive account deletion notifications."""
1636 user, token = generate_user()
1638 topic_action = NotificationTopicAction.account_deletion__complete
1640 # Delete the user
1641 with session_scope() as session:
1642 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1644 with session_scope() as session:
1645 notify(
1646 session,
1647 user_id=user.id,
1648 topic_action=topic_action,
1649 key="",
1650 data=notification_data_pb2.AccountDeletionComplete(
1651 undelete_token="test-token",
1652 undelete_days=7,
1653 ),
1654 )
1656 # Email SHOULD be sent to deleted user for account deletion notification
1657 email = email_collector.pop_for_recipient(user.email, last=True)
1658 assert email.recipient == user.email
1661def test_handle_notification_do_not_email_respected(db, email_collector: EmailCollector):
1662 """Test that users with do_not_email set don't receive non-critical emails."""
1663 user, token = generate_user()
1665 topic_action = NotificationTopicAction.badge__add
1667 # Enable email notifications
1668 with notifications_session(token) as notifications:
1669 notifications.SetNotificationSettings(
1670 notifications_pb2.SetNotificationSettingsReq(
1671 preferences=[
1672 notifications_pb2.SingleNotificationPreference(
1673 topic=topic_action.topic,
1674 action=topic_action.action,
1675 delivery_method="email",
1676 enabled=True,
1677 )
1678 ],
1679 )
1680 )
1682 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1683 with session_scope() as session:
1684 session.execute(
1685 update(User)
1686 .where(User.id == user.id)
1687 .values(
1688 hosting_status=HostingStatus.cant_host,
1689 meetup_status=MeetupStatus.does_not_want_to_meetup,
1690 do_not_email=True,
1691 )
1692 )
1694 with session_scope() as session:
1695 notify(
1696 session,
1697 user_id=user.id,
1698 topic_action=topic_action,
1699 key="test-badge",
1700 data=notification_data_pb2.BadgeAdd(
1701 badge_id="volunteer",
1702 badge_name="Active Volunteer",
1703 badge_description="This user is an active volunteer",
1704 ),
1705 )
1707 # Email should not be sent when do_not_email is True
1708 assert email_collector.count_for_recipient(user.email) == 0
1711def test_handle_notification_critical_bypasses_do_not_email(db, email_collector: EmailCollector):
1712 """Test that critical notifications bypass do_not_email setting."""
1713 user, token = generate_user()
1715 topic_action = NotificationTopicAction.password__change
1717 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1718 with session_scope() as session:
1719 session.execute(
1720 update(User)
1721 .where(User.id == user.id)
1722 .values(
1723 hosting_status=HostingStatus.cant_host,
1724 meetup_status=MeetupStatus.does_not_want_to_meetup,
1725 do_not_email=True,
1726 )
1727 )
1729 with session_scope() as session:
1730 notify(
1731 session,
1732 user_id=user.id,
1733 topic_action=topic_action,
1734 key="",
1735 data=None,
1736 )
1738 # Critical email SHOULD be sent even with do_not_email=True
1739 email = email_collector.pop_for_recipient(user.email, last=True)
1740 assert email.recipient == user.email
1743def test_handle_notification_duplicate_delivery_skipped(db, push_collector: PushCollector):
1744 """Test that duplicate deliveries are skipped when NotificationDelivery already exists."""
1745 user, token = generate_user()
1747 topic_action = NotificationTopicAction.badge__add
1749 # Create notification manually
1750 with session_scope() as session:
1751 notification = Notification(
1752 user_id=user.id,
1753 topic_action=topic_action,
1754 key="test-badge",
1755 data=notification_data_pb2.BadgeAdd(
1756 badge_id="volunteer",
1757 badge_name="Active Volunteer",
1758 badge_description="This user is an active volunteer",
1759 ).SerializeToString(),
1760 )
1761 session.add(notification)
1762 session.flush()
1763 notification_id = notification.id
1765 # Manually create a push delivery (simulating it was already delivered)
1766 session.add(
1767 NotificationDelivery(
1768 notification_id=notification_id,
1769 delivery_type=NotificationDeliveryType.push,
1770 delivered=now(),
1771 )
1772 )
1774 # Try to handle the notification again
1775 handle_notification(jobs_pb2.HandleNotificationPayload(notification_id=notification_id))
1777 # No new push should be sent since delivery already exists
1778 assert push_collector.count_for_user(user.id) == 0
1780 # Verify only one delivery exists
1781 with session_scope() as session:
1782 delivery_count = len(
1783 session.execute(
1784 select(NotificationDelivery)
1785 .where(NotificationDelivery.notification_id == notification_id)
1786 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1787 )
1788 .scalars()
1789 .all()
1790 )
1791 assert delivery_count == 1
1794def test_handle_notification_deferred_when_content_not_visible(db, moderator):
1795 """Test that notifications linked to non-visible moderated content are deferred."""
1796 user1, token1 = generate_user(complete_profile=True)
1797 user2, token2 = generate_user(complete_profile=True)
1799 # Create a friend request (which creates a moderation state)
1800 # This also queues a notification via SendFriendRequest
1801 with api_session(token2) as api:
1802 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1804 # Process the queued job (handle_notification)
1805 process_job()
1807 # The notification should exist but have no deliveries because content is shadowed
1808 with session_scope() as session:
1809 notification = session.execute(
1810 select(Notification)
1811 .where(Notification.user_id == user1.id)
1812 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1813 ).scalar_one()
1815 deliveries = (
1816 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1817 .scalars()
1818 .all()
1819 )
1820 # No deliveries because content is not yet visible (shadowed)
1821 assert len(deliveries) == 0
1824def test_handle_notification_delivered_when_content_visible(db, moderator):
1825 """Test that notifications linked to visible moderated content are delivered."""
1826 user1, token1 = generate_user(complete_profile=True)
1827 user2, token2 = generate_user(complete_profile=True)
1829 # Create a friend request
1830 with api_session(token2) as api:
1831 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1832 res = api.ListFriendRequests(empty_pb2.Empty())
1833 fr_id = res.sent[0].friend_request_id
1835 # Process initial job (which is deferred because content is shadowed)
1836 process_job()
1838 # Approve the friend request so it becomes visible (this queues the notification job again)
1839 moderator.approve_friend_request(fr_id)
1841 # Process the notification job that was re-queued after approval
1842 process_jobs()
1844 # Notification should have been delivered
1845 with session_scope() as session:
1846 notification = session.execute(
1847 select(Notification)
1848 .where(Notification.user_id == user1.id)
1849 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1850 ).scalar_one()
1852 deliveries = (
1853 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1854 .scalars()
1855 .all()
1856 )
1857 # At least one delivery should exist
1858 assert len(deliveries) > 0
1861def test_notification_serializes_shadowed_actor(db, moderator):
1862 recipient, _ = generate_user(complete_profile=True)
1863 sender, sender_token = generate_user(complete_profile=True)
1865 with session_scope() as session:
1866 session.execute(update(User).where(User.id == sender.id).values(shadowed_at=now()))
1868 with api_session(sender_token) as api:
1869 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=recipient.id))
1871 process_job()
1873 with session_scope() as session:
1874 notification = session.execute(
1875 select(Notification)
1876 .where(Notification.user_id == recipient.id)
1877 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1878 ).scalar_one()
1879 data = notification_data_pb2.FriendRequestCreate.FromString(notification.data)
1880 assert data.other_user.user_id == sender.id
1881 assert not data.other_user.is_ghost
1884def test_handle_notification_multiple_delivery_types(
1885 db, email_collector: EmailCollector, push_collector: PushCollector
1886):
1887 """Test that multiple delivery types are processed for a single notification."""
1888 user, token = generate_user()
1890 topic_action = NotificationTopicAction.badge__add
1892 # Enable both email and push notifications
1893 with notifications_session(token) as notifications:
1894 notifications.SetNotificationSettings(
1895 notifications_pb2.SetNotificationSettingsReq(
1896 preferences=[
1897 notifications_pb2.SingleNotificationPreference(
1898 topic=topic_action.topic,
1899 action=topic_action.action,
1900 delivery_method="email",
1901 enabled=True,
1902 ),
1903 notifications_pb2.SingleNotificationPreference(
1904 topic=topic_action.topic,
1905 action=topic_action.action,
1906 delivery_method="push",
1907 enabled=True,
1908 ),
1909 notifications_pb2.SingleNotificationPreference(
1910 topic=topic_action.topic,
1911 action=topic_action.action,
1912 delivery_method="digest",
1913 enabled=True,
1914 ),
1915 ],
1916 )
1917 )
1919 with session_scope() as session:
1920 notify(
1921 session,
1922 user_id=user.id,
1923 topic_action=topic_action,
1924 key="test-badge",
1925 data=notification_data_pb2.BadgeAdd(
1926 badge_id="volunteer",
1927 badge_name="Active Volunteer",
1928 badge_description="This user is an active volunteer",
1929 ),
1930 )
1932 # Email should be sent
1933 email_collector.pop_for_recipient(user.email, last=True)
1935 # Push should be sent
1936 push = push_collector.pop_for_user(user.id, last=True)
1937 assert "Active Volunteer" in push.content.title
1939 # All three delivery types should have deliveries
1940 with session_scope() as session:
1941 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1943 deliveries = (
1944 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1945 .scalars()
1946 .all()
1947 )
1949 delivery_types = {d.delivery_type for d in deliveries}
1950 assert NotificationDeliveryType.email in delivery_types
1951 assert NotificationDeliveryType.push in delivery_types
1952 assert NotificationDeliveryType.digest in delivery_types
1954 # Email and push should have delivered timestamps
1955 for delivery in deliveries:
1956 if delivery.delivery_type in [NotificationDeliveryType.email, NotificationDeliveryType.push]:
1957 assert delivery.delivered is not None
1958 elif delivery.delivery_type == NotificationDeliveryType.digest: 1958 ↛ 1955line 1958 didn't jump to line 1955 because the condition on line 1958 was always true
1959 assert delivery.delivered is None