Coverage for app/backend/src/tests/test_notifications.py: 99%
780 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 11:50 +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
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_event_reminder_email_sent(db, email_collector: EmailCollector):
661 user, token = generate_user()
662 title = "Board Game Night"
664 # Saturday, July 5, 2025 at 4:40:00 AM UTC
665 start_time = timestamp_pb2.Timestamp(seconds=1751690400)
666 timezone = "Etc/GMT-2" # Etc/GMT-2 means GMT+2
667 expected_time_str = "Saturday, July 5, 6:40 AM"
669 with session_scope() as session:
670 user_in_session = session.get_one(User, user.id)
672 notify(
673 session,
674 user_id=user.id,
675 topic_action=NotificationTopicAction.event__reminder,
676 key="",
677 data=notification_data_pb2.EventReminder(
678 event=events_pb2.Event(
679 event_id=1, slug="board-game-night", title=title, start_time=start_time, timezone=timezone
680 ),
681 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)),
682 ),
683 )
685 email = email_collector.pop_for_recipient(user.email, last=True)
686 assert email.recipient == user.email
687 assert title in email.html
688 assert title in email.plain
689 assert expected_time_str in email.html
690 assert expected_time_str in email.plain
693def test_RegisterMobilePushNotificationSubscription(db):
694 user, token = generate_user()
696 with notifications_session(token) as notifications:
697 notifications.RegisterMobilePushNotificationSubscription(
698 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
699 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
700 device_name="My iPhone",
701 device_type="ios",
702 )
703 )
705 # Check subscription was created
706 with session_scope() as session:
707 sub = session.execute(
708 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
709 ).scalar_one()
710 assert sub.platform == PushNotificationPlatform.expo
711 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]"
712 assert sub.device_name == "My iPhone"
713 assert sub.device_type == DeviceType.ios
714 assert sub.disabled_at == DATETIME_INFINITY
717def test_RegisterMobilePushNotificationSubscription_android(db):
718 user, token = generate_user()
720 with notifications_session(token) as notifications:
721 notifications.RegisterMobilePushNotificationSubscription(
722 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
723 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
724 device_name="My Android",
725 device_type="android",
726 )
727 )
729 with session_scope() as session:
730 sub = session.execute(
731 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
732 ).scalar_one()
733 assert sub.platform == PushNotificationPlatform.expo
734 assert sub.device_type == DeviceType.android
737def test_RegisterMobilePushNotificationSubscription_no_device_type(db):
738 user, token = generate_user()
740 with notifications_session(token) as notifications:
741 notifications.RegisterMobilePushNotificationSubscription(
742 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
743 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]",
744 )
745 )
747 with session_scope() as session:
748 sub = session.execute(
749 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
750 ).scalar_one()
751 assert sub.platform == PushNotificationPlatform.expo
752 assert sub.device_name is None
753 assert sub.device_type is None
756def test_RegisterMobilePushNotificationSubscription_re_enable(db):
757 user, token = generate_user()
759 # Create a disabled subscription directly in the DB
760 with session_scope() as session:
761 sub = PushNotificationSubscription(
762 user_id=user.id,
763 platform=PushNotificationPlatform.expo,
764 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
765 device_name="Old Device",
766 device_type=DeviceType.ios,
767 )
768 sub.disabled_at = now()
769 session.add(sub)
770 session.flush()
771 sub_id = sub.id
773 # Re-register with the same token
774 with notifications_session(token) as notifications:
775 notifications.RegisterMobilePushNotificationSubscription(
776 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
777 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
778 device_name="New Device Name",
779 device_type="android",
780 )
781 )
783 # Check subscription was re-enabled and updated
784 with session_scope() as session:
785 sub = session.execute(
786 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
787 ).scalar_one()
788 assert sub.disabled_at == DATETIME_INFINITY
789 assert sub.device_name == "New Device Name"
790 assert sub.device_type == DeviceType.android
793def test_RegisterMobilePushNotificationSubscription_already_exists(db):
794 user, token = generate_user()
796 # Create an active subscription directly in the DB
797 with session_scope() as session:
798 sub = PushNotificationSubscription(
799 user_id=user.id,
800 platform=PushNotificationPlatform.expo,
801 token="ExponentPushToken[existingtoken]",
802 device_name="Existing Device",
803 device_type=DeviceType.ios,
804 )
805 session.add(sub)
807 # Try to register with the same token - should just return without error
808 with notifications_session(token) as notifications:
809 notifications.RegisterMobilePushNotificationSubscription(
810 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
811 token="ExponentPushToken[existingtoken]",
812 device_name="Different Name",
813 )
814 )
816 # Check subscription was NOT modified (already active)
817 with session_scope() as session:
818 sub = session.execute(
819 select(PushNotificationSubscription).where(
820 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]"
821 )
822 ).scalar_one()
823 assert sub.device_name == "Existing Device" # unchanged
826def test_SendTestMobilePushNotification(db, push_collector: PushCollector):
827 user, token = generate_user()
829 with notifications_session(token) as notifications:
830 notifications.SendTestMobilePushNotification(empty_pb2.Empty())
832 push = push_collector.pop_for_user(user.id, last=True)
833 assert push.content.title == "Mobile notifications test"
834 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉"
837def test_get_expo_push_receipts(db):
838 mock_response = Mock()
839 mock_response.status_code = 200
840 mock_response.json.return_value = {
841 "data": {
842 "ticket-1": {"status": "ok"},
843 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
844 }
845 }
847 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post:
848 result = get_expo_push_receipts(["ticket-1", "ticket-2"])
850 mock_post.assert_called_once()
851 call_args = mock_post.call_args
852 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts"
853 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}
855 assert result == {
856 "ticket-1": {"status": "ok"},
857 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
858 }
861def test_get_expo_push_receipts_empty(db):
862 result = get_expo_push_receipts([])
863 assert result == {}
866def test_check_expo_push_receipts_success(db):
867 """Test batch receipt checking with successful delivery."""
868 user, token = generate_user()
870 # Create a push subscription and delivery attempt (old enough to be checked)
871 with session_scope() as session:
872 sub = PushNotificationSubscription(
873 user_id=user.id,
874 platform=PushNotificationPlatform.expo,
875 token="ExponentPushToken[testtoken123]",
876 device_name="Test Device",
877 device_type=DeviceType.ios,
878 )
879 session.add(sub)
880 session.flush()
882 attempt = PushNotificationDeliveryAttempt(
883 push_notification_subscription_id=sub.id,
884 outcome=PushNotificationDeliveryOutcome.success,
885 status_code=200,
886 expo_ticket_id="test-ticket-id",
887 )
888 session.add(attempt)
889 session.flush()
890 # Make the attempt old enough to be checked (>15 min)
891 attempt.time = now() - timedelta(minutes=20)
892 attempt_id = attempt.id
893 sub_id = sub.id
895 # Mock the receipt API call
896 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
897 mock_post.return_value.status_code = 200
898 mock_post.return_value.json.return_value = {"data": {"test-ticket-id": {"status": "ok"}}}
900 check_expo_push_receipts(empty_pb2.Empty())
902 # Verify the attempt was updated
903 with session_scope() as session:
904 attempt = session.execute(
905 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
906 ).scalar_one()
907 assert attempt.receipt_checked_at is not None
908 assert attempt.receipt_status == "ok"
909 assert attempt.receipt_error_code is None
911 # Subscription should still be enabled
912 sub = session.execute(
913 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
914 ).scalar_one()
915 assert sub.disabled_at == DATETIME_INFINITY
918def test_check_expo_push_receipts_device_not_registered(db):
919 """Test batch receipt checking with DeviceNotRegistered error disables subscription."""
920 user, token = generate_user()
922 # Create a push subscription and delivery attempt
923 with session_scope() as session:
924 sub = PushNotificationSubscription(
925 user_id=user.id,
926 platform=PushNotificationPlatform.expo,
927 token="ExponentPushToken[devicegone]",
928 device_name="Test Device",
929 device_type=DeviceType.android,
930 )
931 session.add(sub)
932 session.flush()
934 attempt = PushNotificationDeliveryAttempt(
935 push_notification_subscription_id=sub.id,
936 outcome=PushNotificationDeliveryOutcome.success,
937 status_code=200,
938 expo_ticket_id="ticket-device-gone",
939 )
940 session.add(attempt)
941 session.flush()
942 # Make the attempt old enough to be checked
943 attempt.time = now() - timedelta(minutes=15)
944 attempt_id = attempt.id
945 sub_id = sub.id
947 # Mock the receipt API call with DeviceNotRegistered error
948 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
949 mock_post.return_value.status_code = 200
950 mock_post.return_value.json.return_value = {
951 "data": {
952 "ticket-device-gone": {
953 "status": "error",
954 "details": {"error": "DeviceNotRegistered"},
955 }
956 }
957 }
959 check_expo_push_receipts(empty_pb2.Empty())
961 # Verify the attempt was updated and subscription disabled
962 with session_scope() as session:
963 attempt = session.execute(
964 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
965 ).scalar_one()
966 assert attempt.receipt_checked_at is not None
967 assert attempt.receipt_status == "error"
968 assert attempt.receipt_error_code == "DeviceNotRegistered"
970 # Subscription should be disabled
971 sub = session.execute(
972 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
973 ).scalar_one()
974 assert sub.disabled_at <= now()
977def test_check_expo_push_receipts_not_found(db):
978 """Test batch receipt checking when ticket not found (expired)."""
979 user, token = generate_user()
981 with session_scope() as session:
982 sub = PushNotificationSubscription(
983 user_id=user.id,
984 platform=PushNotificationPlatform.expo,
985 token="ExponentPushToken[notfound]",
986 )
987 session.add(sub)
988 session.flush()
990 attempt = PushNotificationDeliveryAttempt(
991 push_notification_subscription_id=sub.id,
992 outcome=PushNotificationDeliveryOutcome.success,
993 status_code=200,
994 expo_ticket_id="unknown-ticket",
995 )
996 session.add(attempt)
997 session.flush()
998 # Make the attempt old enough to be checked
999 attempt.time = now() - timedelta(minutes=15)
1000 attempt_id = attempt.id
1001 sub_id = sub.id
1003 # Mock empty receipt response (ticket not found)
1004 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1005 mock_post.return_value.status_code = 200
1006 mock_post.return_value.json.return_value = {"data": {}}
1008 check_expo_push_receipts(empty_pb2.Empty())
1010 with session_scope() as session:
1011 attempt = session.execute(
1012 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1013 ).scalar_one()
1014 assert attempt.receipt_checked_at is not None
1015 assert attempt.receipt_status == "not_found"
1017 # Subscription should still be enabled
1018 sub = session.execute(
1019 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
1020 ).scalar_one()
1021 assert sub.disabled_at == DATETIME_INFINITY
1024def test_check_expo_push_receipts_skips_already_checked(db):
1025 """Test that already-checked receipts are not re-checked."""
1026 user, token = generate_user()
1028 # Create an attempt that was already checked
1029 with session_scope() as session:
1030 sub = PushNotificationSubscription(
1031 user_id=user.id,
1032 platform=PushNotificationPlatform.expo,
1033 token="ExponentPushToken[alreadychecked]",
1034 )
1035 session.add(sub)
1036 session.flush()
1038 attempt = PushNotificationDeliveryAttempt(
1039 push_notification_subscription_id=sub.id,
1040 outcome=PushNotificationDeliveryOutcome.success,
1041 status_code=200,
1042 expo_ticket_id="already-checked-ticket",
1043 receipt_checked_at=now(),
1044 receipt_status="ok",
1045 )
1046 session.add(attempt)
1047 session.flush()
1048 # Make the attempt old enough
1049 attempt.time = now() - timedelta(minutes=15)
1051 # Should not call the API since the only attempt is already checked
1052 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1053 check_expo_push_receipts(empty_pb2.Empty())
1054 mock_post.assert_not_called()
1057def test_SendDevPushNotification_success(db, push_collector: PushCollector):
1058 """Test SendDevPushNotification sends push with all specified parameters."""
1059 user, token = generate_user()
1061 # Enable dev APIs for this test
1062 config.ENABLE_DEV_APIS = True
1064 with notifications_session(token) as notifications:
1065 notifications.SendDevPushNotification(
1066 notifications_pb2.SendDevPushNotificationReq(
1067 title="Test Dev Title",
1068 body="Test dev notification body",
1069 icon="https://example.com/icon.png",
1070 url="https://example.com/action",
1071 key="test-key",
1072 ttl=3600,
1073 )
1074 )
1076 push = push_collector.pop_for_user(user.id, last=True)
1077 assert push.content.title == "Test Dev Title"
1078 assert push.content.body == "Test dev notification body"
1079 assert push.content.action_url == "https://example.com/action"
1080 assert push.content.icon_url == "https://example.com/icon.png"
1081 assert push.topic_action == "adhoc:testing"
1082 assert push.key == "test-key"
1083 assert push.ttl == 3600
1086def test_SendDevPushNotification_minimal(db, push_collector: PushCollector):
1087 """Test SendDevPushNotification with minimal parameters."""
1088 user, token = generate_user()
1090 config.ENABLE_DEV_APIS = True
1092 with notifications_session(token) as notifications:
1093 notifications.SendDevPushNotification(
1094 notifications_pb2.SendDevPushNotificationReq(
1095 title="Minimal Title",
1096 body="Minimal body",
1097 )
1098 )
1100 push = push_collector.pop_for_user(user.id, last=True)
1101 assert push.content.title == "Minimal Title"
1102 assert push.content.body == "Minimal body"
1103 assert push.topic_action == "adhoc:testing"
1106def test_SendDevPushNotification_disabled(db, push_collector: PushCollector):
1107 """Test SendDevPushNotification fails when ENABLE_DEV_APIS is disabled."""
1108 user, token = generate_user()
1110 # Ensure dev APIs are disabled (default in tests)
1111 config.ENABLE_DEV_APIS = False
1113 with notifications_session(token) as notifications:
1114 with pytest.raises(grpc.RpcError) as e:
1115 notifications.SendDevPushNotification(
1116 notifications_pb2.SendDevPushNotificationReq(
1117 title="Should Fail",
1118 body="This should not be sent",
1119 )
1120 )
1121 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1122 assert "Development APIs are not enabled" in not_none(e.value.details())
1124 assert push_collector.count_for_user(user.id) == 0
1127def test_SendDevPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1128 """Test SendDevPushNotification fails when push notifications are disabled."""
1129 user, token = generate_user()
1131 config.ENABLE_DEV_APIS = True
1132 config.PUSH_NOTIFICATIONS_ENABLED = False
1134 with notifications_session(token) as notifications:
1135 with pytest.raises(grpc.RpcError) as e:
1136 notifications.SendDevPushNotification(
1137 notifications_pb2.SendDevPushNotificationReq(
1138 title="Should Fail",
1139 body="This should not be sent",
1140 )
1141 )
1142 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1143 assert "Push notifications are currently disabled" in not_none(e.value.details())
1145 assert push_collector.count_for_user(user.id) == 0
1148def test_check_expo_push_receipts_skips_too_recent(db):
1149 """Test that too-recent receipts (<15 min) are not checked."""
1150 user, token = generate_user()
1152 # Create a recent attempt (not old enough to check)
1153 with session_scope() as session:
1154 sub = PushNotificationSubscription(
1155 user_id=user.id,
1156 platform=PushNotificationPlatform.expo,
1157 token="ExponentPushToken[recent]",
1158 )
1159 session.add(sub)
1160 session.flush()
1162 attempt = PushNotificationDeliveryAttempt(
1163 push_notification_subscription_id=sub.id,
1164 outcome=PushNotificationDeliveryOutcome.success,
1165 status_code=200,
1166 expo_ticket_id="recent-ticket",
1167 )
1168 session.add(attempt)
1169 session.flush()
1170 # Make the attempt only 5 minutes old (too recent)
1171 attempt.time = now() - timedelta(minutes=5)
1173 # Should not call the API since the attempt is too recent
1174 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1175 check_expo_push_receipts(empty_pb2.Empty())
1176 mock_post.assert_not_called()
1179def test_check_expo_push_receipts_batch(db):
1180 """Test that multiple receipts are checked in a single batch."""
1181 user, token = generate_user()
1183 # Create multiple delivery attempts
1184 attempt_ids = []
1185 with session_scope() as session:
1186 sub = PushNotificationSubscription(
1187 user_id=user.id,
1188 platform=PushNotificationPlatform.expo,
1189 token="ExponentPushToken[batch]",
1190 )
1191 session.add(sub)
1192 session.flush()
1194 for i in range(3):
1195 attempt = PushNotificationDeliveryAttempt(
1196 push_notification_subscription_id=sub.id,
1197 outcome=PushNotificationDeliveryOutcome.success,
1198 status_code=200,
1199 expo_ticket_id=f"batch-ticket-{i}",
1200 )
1201 session.add(attempt)
1202 session.flush()
1203 attempt.time = now() - timedelta(minutes=20)
1204 attempt_ids.append(attempt.id)
1206 # Mock the batch receipt API call
1207 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1208 mock_post.return_value.status_code = 200
1209 mock_post.return_value.json.return_value = {
1210 "data": {
1211 "batch-ticket-0": {"status": "ok"},
1212 "batch-ticket-1": {"status": "ok"},
1213 "batch-ticket-2": {"status": "ok"},
1214 }
1215 }
1217 check_expo_push_receipts(empty_pb2.Empty())
1219 # Should only call the API once for all tickets
1220 assert mock_post.call_count == 1
1222 # Verify all attempts were updated
1223 with session_scope() as session:
1224 for attempt_id in attempt_ids:
1225 attempt = session.execute(
1226 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1227 ).scalar_one()
1228 assert attempt.receipt_checked_at is not None
1229 assert attempt.receipt_status == "ok"
1232def test_DebugRedeliverPushNotification_success(db, push_collector: PushCollector):
1233 """Test DebugRedeliverPushNotification redelivers an existing notification."""
1234 user, token = generate_user()
1236 config.ENABLE_DEV_APIS = True
1238 # Create a notification for the user
1239 with session_scope() as session:
1240 notify(
1241 session,
1242 user_id=user.id,
1243 topic_action=NotificationTopicAction.badge__add,
1244 key="test-badge",
1245 data=notification_data_pb2.BadgeAdd(
1246 badge_id="volunteer",
1247 badge_name="Active Volunteer",
1248 badge_description="This user is an active volunteer for Couchers.org",
1249 ),
1250 )
1252 process_job()
1254 # Pop the initial push notification
1255 push_collector.pop_for_user(user.id, last=True)
1257 # Get the notification_id
1258 with session_scope() as session:
1259 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1260 notification_id = notification.id
1262 # Redeliver the notification
1263 with notifications_session(token) as notifications:
1264 notifications.DebugRedeliverPushNotification(
1265 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1266 )
1268 # Verify a new push was sent
1269 push = push_collector.pop_for_user(user.id, last=True)
1270 assert "Active Volunteer" in push.content.title
1271 assert push.topic_action == "badge:add"
1272 assert push.key == "test-badge"
1275def test_DebugRedeliverPushNotification_not_found(db, push_collector: PushCollector):
1276 """Test DebugRedeliverPushNotification fails when notification doesn't exist."""
1277 user, token = generate_user()
1279 config.ENABLE_DEV_APIS = True
1281 with notifications_session(token) as notifications:
1282 with pytest.raises(grpc.RpcError) as e:
1283 notifications.DebugRedeliverPushNotification(
1284 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=999999)
1285 )
1286 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1287 assert "notification not found" in not_none(e.value.details()).lower()
1289 assert push_collector.count_for_user(user.id) == 0
1292def test_DebugRedeliverPushNotification_wrong_user(db, push_collector: PushCollector):
1293 """Test DebugRedeliverPushNotification fails when notification belongs to another user."""
1294 user1, token1 = generate_user()
1295 user2, token2 = generate_user()
1297 config.ENABLE_DEV_APIS = True
1299 # Create a notification for user1
1300 with session_scope() as session:
1301 notify(
1302 session,
1303 user_id=user1.id,
1304 topic_action=NotificationTopicAction.badge__add,
1305 key="test-badge",
1306 data=notification_data_pb2.BadgeAdd(
1307 badge_id="volunteer",
1308 badge_name="Active Volunteer",
1309 badge_description="This user is an active volunteer for Couchers.org",
1310 ),
1311 )
1313 process_job()
1315 # Get the notification_id
1316 with session_scope() as session:
1317 notification = session.execute(select(Notification).where(Notification.user_id == user1.id)).scalar_one()
1318 notification_id = notification.id
1320 # user2 tries to redeliver user1's notification
1321 with notifications_session(token2) as notifications:
1322 with pytest.raises(grpc.RpcError) as e:
1323 notifications.DebugRedeliverPushNotification(
1324 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1325 )
1326 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1327 assert "notification not found" in not_none(e.value.details()).lower()
1329 assert push_collector.count_for_user(user2.id) == 0
1332def test_DebugRedeliverPushNotification_disabled(db, push_collector: PushCollector):
1333 """Test DebugRedeliverPushNotification fails when ENABLE_DEV_APIS is disabled."""
1334 user, token = generate_user()
1336 config.ENABLE_DEV_APIS = False
1338 with notifications_session(token) as notifications:
1339 with pytest.raises(grpc.RpcError) as e:
1340 notifications.DebugRedeliverPushNotification(
1341 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1342 )
1343 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1344 assert "Development APIs are not enabled" in not_none(e.value.details())
1346 assert push_collector.count_for_user(user.id) == 0
1349def test_DebugRedeliverPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1350 """Test DebugRedeliverPushNotification fails when push notifications are disabled."""
1351 user, token = generate_user()
1353 config.ENABLE_DEV_APIS = True
1354 config.PUSH_NOTIFICATIONS_ENABLED = False
1356 with notifications_session(token) as notifications:
1357 with pytest.raises(grpc.RpcError) as e:
1358 notifications.DebugRedeliverPushNotification(
1359 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1360 )
1361 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1362 assert "Push notifications are currently disabled" in not_none(e.value.details())
1364 assert push_collector.count_for_user(user.id) == 0
1367def test_handle_notification_email_delivery(db, email_collector: EmailCollector):
1368 """Test that email notifications are delivered when email preference is enabled."""
1369 user, token = generate_user()
1371 topic_action = NotificationTopicAction.badge__add
1373 # Enable email notifications for this topic
1374 with notifications_session(token) as notifications:
1375 notifications.SetNotificationSettings(
1376 notifications_pb2.SetNotificationSettingsReq(
1377 preferences=[
1378 notifications_pb2.SingleNotificationPreference(
1379 topic=topic_action.topic,
1380 action=topic_action.action,
1381 delivery_method="email",
1382 enabled=True,
1383 )
1384 ],
1385 )
1386 )
1388 with session_scope() as session:
1389 notify(
1390 session,
1391 user_id=user.id,
1392 topic_action=topic_action,
1393 key="test-badge",
1394 data=notification_data_pb2.BadgeAdd(
1395 badge_id="volunteer",
1396 badge_name="Active Volunteer",
1397 badge_description="This user is an active volunteer",
1398 ),
1399 )
1401 email = email_collector.pop_for_recipient(user.email, last=True)
1402 assert email.recipient == user.email
1404 with session_scope() as session:
1405 delivery = session.execute(
1406 select(NotificationDelivery)
1407 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1408 .where(Notification.user_id == user.id)
1409 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.email)
1410 ).scalar_one()
1411 assert delivery.delivered is not None
1414def test_handle_notification_push_delivery(db, push_collector: PushCollector):
1415 """Test that push notifications are delivered immediately when push preference is enabled."""
1416 user, token = generate_user()
1418 topic_action = NotificationTopicAction.badge__add
1420 with session_scope() as session:
1421 notify(
1422 session,
1423 user_id=user.id,
1424 topic_action=topic_action,
1425 key="test-badge",
1426 data=notification_data_pb2.BadgeAdd(
1427 badge_id="volunteer",
1428 badge_name="Active Volunteer",
1429 badge_description="This user is an active volunteer",
1430 ),
1431 )
1433 process_job()
1435 push = push_collector.pop_for_user(user.id, last=True)
1436 assert "Active Volunteer" in push.content.title
1438 with session_scope() as session:
1439 delivery = session.execute(
1440 select(NotificationDelivery)
1441 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1442 .where(Notification.user_id == user.id)
1443 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1444 ).scalar_one()
1445 assert delivery.delivered is not None
1448def test_handle_notification_digest_delivery(db):
1449 """Test that digest notifications are queued without a delivered timestamp."""
1450 user, token = generate_user()
1452 topic_action = NotificationTopicAction.badge__add
1454 # Enable only digest notifications for this topic
1455 with notifications_session(token) as notifications:
1456 notifications.SetNotificationSettings(
1457 notifications_pb2.SetNotificationSettingsReq(
1458 preferences=[
1459 notifications_pb2.SingleNotificationPreference(
1460 topic=topic_action.topic,
1461 action=topic_action.action,
1462 delivery_method="push",
1463 enabled=False,
1464 ),
1465 notifications_pb2.SingleNotificationPreference(
1466 topic=topic_action.topic,
1467 action=topic_action.action,
1468 delivery_method="digest",
1469 enabled=True,
1470 ),
1471 ],
1472 )
1473 )
1475 with session_scope() as session:
1476 notify(
1477 session,
1478 user_id=user.id,
1479 topic_action=topic_action,
1480 key="test-badge",
1481 data=notification_data_pb2.BadgeAdd(
1482 badge_id="volunteer",
1483 badge_name="Active Volunteer",
1484 badge_description="This user is an active volunteer",
1485 ),
1486 )
1488 process_job()
1490 # Verify digest NotificationDelivery was created WITHOUT delivered timestamp
1491 with session_scope() as session:
1492 delivery = session.execute(
1493 select(NotificationDelivery)
1494 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1495 .where(Notification.user_id == user.id)
1496 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.digest)
1497 ).scalar_one()
1498 assert delivery.delivered is None
1501def test_handle_notification_banned_user_no_email(db, email_collector: EmailCollector):
1502 """Test that banned users don't receive email notifications."""
1503 user, token = generate_user()
1505 topic_action = NotificationTopicAction.badge__add
1507 # Enable email notifications
1508 with notifications_session(token) as notifications:
1509 notifications.SetNotificationSettings(
1510 notifications_pb2.SetNotificationSettingsReq(
1511 preferences=[
1512 notifications_pb2.SingleNotificationPreference(
1513 topic=topic_action.topic,
1514 action=topic_action.action,
1515 delivery_method="email",
1516 enabled=True,
1517 )
1518 ],
1519 )
1520 )
1522 # Ban the user
1523 with session_scope() as session:
1524 session.execute(update(User).where(User.id == user.id).values(banned_at=now()))
1526 with session_scope() as session:
1527 notify(
1528 session,
1529 user_id=user.id,
1530 topic_action=topic_action,
1531 key="test-badge",
1532 data=notification_data_pb2.BadgeAdd(
1533 badge_id="volunteer",
1534 badge_name="Active Volunteer",
1535 badge_description="This user is an active volunteer",
1536 ),
1537 )
1539 # Email should not be sent to the banned user
1540 assert email_collector.count_for_recipient(user.email) == 0
1543def test_handle_notification_deleted_user_no_regular_email(db, email_collector: EmailCollector):
1544 """Test that deleted users don't receive non-account-deletion email notifications."""
1545 user, token = generate_user()
1547 topic_action = NotificationTopicAction.badge__add
1549 # Enable email notifications
1550 with notifications_session(token) as notifications:
1551 notifications.SetNotificationSettings(
1552 notifications_pb2.SetNotificationSettingsReq(
1553 preferences=[
1554 notifications_pb2.SingleNotificationPreference(
1555 topic=topic_action.topic,
1556 action=topic_action.action,
1557 delivery_method="email",
1558 enabled=True,
1559 )
1560 ],
1561 )
1562 )
1564 # Delete the user
1565 with session_scope() as session:
1566 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1568 with session_scope() as session:
1569 notify(
1570 session,
1571 user_id=user.id,
1572 topic_action=topic_action,
1573 key="test-badge",
1574 data=notification_data_pb2.BadgeAdd(
1575 badge_id="volunteer",
1576 badge_name="Active Volunteer",
1577 badge_description="This user is an active volunteer",
1578 ),
1579 )
1581 # Email should not be sent to deleted user for non-account-deletion notification
1582 assert email_collector.count_for_recipient(user.email) == 0
1585def test_handle_notification_deleted_user_receives_account_deletion_email(db, email_collector: EmailCollector):
1586 """Test that deleted users CAN receive account deletion notifications."""
1587 user, token = generate_user()
1589 topic_action = NotificationTopicAction.account_deletion__complete
1591 # Delete the user
1592 with session_scope() as session:
1593 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1595 with session_scope() as session:
1596 notify(
1597 session,
1598 user_id=user.id,
1599 topic_action=topic_action,
1600 key="",
1601 data=notification_data_pb2.AccountDeletionComplete(
1602 undelete_token="test-token",
1603 undelete_days=7,
1604 ),
1605 )
1607 # Email SHOULD be sent to deleted user for account deletion notification
1608 email = email_collector.pop_for_recipient(user.email, last=True)
1609 assert email.recipient == user.email
1612def test_handle_notification_do_not_email_respected(db, email_collector: EmailCollector):
1613 """Test that users with do_not_email set don't receive non-critical emails."""
1614 user, token = generate_user()
1616 topic_action = NotificationTopicAction.badge__add
1618 # Enable email notifications
1619 with notifications_session(token) as notifications:
1620 notifications.SetNotificationSettings(
1621 notifications_pb2.SetNotificationSettingsReq(
1622 preferences=[
1623 notifications_pb2.SingleNotificationPreference(
1624 topic=topic_action.topic,
1625 action=topic_action.action,
1626 delivery_method="email",
1627 enabled=True,
1628 )
1629 ],
1630 )
1631 )
1633 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1634 with session_scope() as session:
1635 session.execute(
1636 update(User)
1637 .where(User.id == user.id)
1638 .values(
1639 hosting_status=HostingStatus.cant_host,
1640 meetup_status=MeetupStatus.does_not_want_to_meetup,
1641 do_not_email=True,
1642 )
1643 )
1645 with session_scope() as session:
1646 notify(
1647 session,
1648 user_id=user.id,
1649 topic_action=topic_action,
1650 key="test-badge",
1651 data=notification_data_pb2.BadgeAdd(
1652 badge_id="volunteer",
1653 badge_name="Active Volunteer",
1654 badge_description="This user is an active volunteer",
1655 ),
1656 )
1658 # Email should not be sent when do_not_email is True
1659 assert email_collector.count_for_recipient(user.email) == 0
1662def test_handle_notification_critical_bypasses_do_not_email(db, email_collector: EmailCollector):
1663 """Test that critical notifications bypass do_not_email setting."""
1664 user, token = generate_user()
1666 topic_action = NotificationTopicAction.password__change
1668 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1669 with session_scope() as session:
1670 session.execute(
1671 update(User)
1672 .where(User.id == user.id)
1673 .values(
1674 hosting_status=HostingStatus.cant_host,
1675 meetup_status=MeetupStatus.does_not_want_to_meetup,
1676 do_not_email=True,
1677 )
1678 )
1680 with session_scope() as session:
1681 notify(
1682 session,
1683 user_id=user.id,
1684 topic_action=topic_action,
1685 key="",
1686 data=None,
1687 )
1689 # Critical email SHOULD be sent even with do_not_email=True
1690 email = email_collector.pop_for_recipient(user.email, last=True)
1691 assert email.recipient == user.email
1694def test_handle_notification_duplicate_delivery_skipped(db, push_collector: PushCollector):
1695 """Test that duplicate deliveries are skipped when NotificationDelivery already exists."""
1696 user, token = generate_user()
1698 topic_action = NotificationTopicAction.badge__add
1700 # Create notification manually
1701 with session_scope() as session:
1702 notification = Notification(
1703 user_id=user.id,
1704 topic_action=topic_action,
1705 key="test-badge",
1706 data=notification_data_pb2.BadgeAdd(
1707 badge_id="volunteer",
1708 badge_name="Active Volunteer",
1709 badge_description="This user is an active volunteer",
1710 ).SerializeToString(),
1711 )
1712 session.add(notification)
1713 session.flush()
1714 notification_id = notification.id
1716 # Manually create a push delivery (simulating it was already delivered)
1717 session.add(
1718 NotificationDelivery(
1719 notification_id=notification_id,
1720 delivery_type=NotificationDeliveryType.push,
1721 delivered=now(),
1722 )
1723 )
1725 # Try to handle the notification again
1726 handle_notification(jobs_pb2.HandleNotificationPayload(notification_id=notification_id))
1728 # No new push should be sent since delivery already exists
1729 assert push_collector.count_for_user(user.id) == 0
1731 # Verify only one delivery exists
1732 with session_scope() as session:
1733 delivery_count = len(
1734 session.execute(
1735 select(NotificationDelivery)
1736 .where(NotificationDelivery.notification_id == notification_id)
1737 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1738 )
1739 .scalars()
1740 .all()
1741 )
1742 assert delivery_count == 1
1745def test_handle_notification_deferred_when_content_not_visible(db, moderator):
1746 """Test that notifications linked to non-visible moderated content are deferred."""
1747 user1, token1 = generate_user(complete_profile=True)
1748 user2, token2 = generate_user(complete_profile=True)
1750 # Create a friend request (which creates a moderation state)
1751 # This also queues a notification via SendFriendRequest
1752 with api_session(token2) as api:
1753 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1755 # Process the queued job (handle_notification)
1756 process_job()
1758 # The notification should exist but have no deliveries because content is shadowed
1759 with session_scope() as session:
1760 notification = session.execute(
1761 select(Notification)
1762 .where(Notification.user_id == user1.id)
1763 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1764 ).scalar_one()
1766 deliveries = (
1767 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1768 .scalars()
1769 .all()
1770 )
1771 # No deliveries because content is not yet visible (shadowed)
1772 assert len(deliveries) == 0
1775def test_handle_notification_delivered_when_content_visible(db, moderator):
1776 """Test that notifications linked to visible moderated content are delivered."""
1777 user1, token1 = generate_user(complete_profile=True)
1778 user2, token2 = generate_user(complete_profile=True)
1780 # Create a friend request
1781 with api_session(token2) as api:
1782 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1783 res = api.ListFriendRequests(empty_pb2.Empty())
1784 fr_id = res.sent[0].friend_request_id
1786 # Process initial job (which is deferred because content is shadowed)
1787 process_job()
1789 # Approve the friend request so it becomes visible (this queues the notification job again)
1790 moderator.approve_friend_request(fr_id)
1792 # Process the notification job that was re-queued after approval
1793 process_jobs()
1795 # Notification should have been delivered
1796 with session_scope() as session:
1797 notification = session.execute(
1798 select(Notification)
1799 .where(Notification.user_id == user1.id)
1800 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1801 ).scalar_one()
1803 deliveries = (
1804 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1805 .scalars()
1806 .all()
1807 )
1808 # At least one delivery should exist
1809 assert len(deliveries) > 0
1812def test_notification_serializes_shadowed_actor(db, moderator):
1813 recipient, _ = generate_user(complete_profile=True)
1814 sender, sender_token = generate_user(complete_profile=True)
1816 with session_scope() as session:
1817 session.execute(update(User).where(User.id == sender.id).values(shadowed_at=now()))
1819 with api_session(sender_token) as api:
1820 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=recipient.id))
1822 process_job()
1824 with session_scope() as session:
1825 notification = session.execute(
1826 select(Notification)
1827 .where(Notification.user_id == recipient.id)
1828 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1829 ).scalar_one()
1830 data = notification_data_pb2.FriendRequestCreate.FromString(notification.data)
1831 assert data.other_user.user_id == sender.id
1832 assert not data.other_user.is_ghost
1835def test_handle_notification_multiple_delivery_types(
1836 db, email_collector: EmailCollector, push_collector: PushCollector
1837):
1838 """Test that multiple delivery types are processed for a single notification."""
1839 user, token = generate_user()
1841 topic_action = NotificationTopicAction.badge__add
1843 # Enable both email and push notifications
1844 with notifications_session(token) as notifications:
1845 notifications.SetNotificationSettings(
1846 notifications_pb2.SetNotificationSettingsReq(
1847 preferences=[
1848 notifications_pb2.SingleNotificationPreference(
1849 topic=topic_action.topic,
1850 action=topic_action.action,
1851 delivery_method="email",
1852 enabled=True,
1853 ),
1854 notifications_pb2.SingleNotificationPreference(
1855 topic=topic_action.topic,
1856 action=topic_action.action,
1857 delivery_method="push",
1858 enabled=True,
1859 ),
1860 notifications_pb2.SingleNotificationPreference(
1861 topic=topic_action.topic,
1862 action=topic_action.action,
1863 delivery_method="digest",
1864 enabled=True,
1865 ),
1866 ],
1867 )
1868 )
1870 with session_scope() as session:
1871 notify(
1872 session,
1873 user_id=user.id,
1874 topic_action=topic_action,
1875 key="test-badge",
1876 data=notification_data_pb2.BadgeAdd(
1877 badge_id="volunteer",
1878 badge_name="Active Volunteer",
1879 badge_description="This user is an active volunteer",
1880 ),
1881 )
1883 # Email should be sent
1884 email_collector.pop_for_recipient(user.email, last=True)
1886 # Push should be sent
1887 push = push_collector.pop_for_user(user.id, last=True)
1888 assert "Active Volunteer" in push.content.title
1890 # All three delivery types should have deliveries
1891 with session_scope() as session:
1892 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1894 deliveries = (
1895 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1896 .scalars()
1897 .all()
1898 )
1900 delivery_types = {d.delivery_type for d in deliveries}
1901 assert NotificationDeliveryType.email in delivery_types
1902 assert NotificationDeliveryType.push in delivery_types
1903 assert NotificationDeliveryType.digest in delivery_types
1905 # Email and push should have delivered timestamps
1906 for delivery in deliveries:
1907 if delivery.delivery_type in [NotificationDeliveryType.email, NotificationDeliveryType.push]:
1908 assert delivery.delivered is not None
1909 elif delivery.delivery_type == NotificationDeliveryType.digest: 1909 ↛ 1906line 1909 didn't jump to line 1906 because the condition on line 1909 was always true
1910 assert delivery.delivered is None