Coverage for app / backend / src / tests / test_notifications.py: 99%
763 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 14:14 +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.i18n import LocalizationContext
19from couchers.jobs.handlers import check_expo_push_receipts
20from couchers.jobs.worker import process_job
21from couchers.models import (
22 DeviceType,
23 HostingStatus,
24 MeetupStatus,
25 Notification,
26 NotificationDelivery,
27 NotificationDeliveryType,
28 NotificationTopicAction,
29 PushNotificationDeliveryAttempt,
30 PushNotificationDeliveryOutcome,
31 PushNotificationPlatform,
32 PushNotificationSubscription,
33 User,
34)
35from couchers.notifications.background import handle_notification
36from couchers.notifications.expo_api import get_expo_push_receipts
37from couchers.notifications.notify import notify
38from couchers.notifications.settings import get_topic_actions_by_delivery_type
39from couchers.proto import (
40 api_pb2,
41 auth_pb2,
42 conversations_pb2,
43 editor_pb2,
44 events_pb2,
45 notification_data_pb2,
46 notifications_pb2,
47)
48from couchers.proto.internal import jobs_pb2, unsubscribe_pb2
49from couchers.servicers.api import user_model_to_pb
50from couchers.utils import not_none, now
51from tests.fixtures.db import generate_user
52from tests.fixtures.misc import PushCollector, email_fields, mock_notification_email, process_jobs
53from tests.fixtures.sessions import (
54 api_session,
55 auth_api_session,
56 conversations_session,
57 notifications_session,
58 real_editor_session,
59)
62@pytest.fixture(autouse=True)
63def _(testconfig):
64 pass
67@pytest.mark.parametrize("enabled", [True, False])
68def test_SetNotificationSettings_preferences_respected_editable(db, enabled):
69 user, token = generate_user()
71 # enable a notification type and check it gets delivered
72 topic_action = NotificationTopicAction.badge__add
74 with notifications_session(token) as notifications:
75 notifications.SetNotificationSettings(
76 notifications_pb2.SetNotificationSettingsReq(
77 preferences=[
78 notifications_pb2.SingleNotificationPreference(
79 topic=topic_action.topic,
80 action=topic_action.action,
81 delivery_method="push",
82 enabled=enabled,
83 )
84 ],
85 )
86 )
88 with session_scope() as session:
89 notify(
90 session,
91 user_id=user.id,
92 topic_action=topic_action,
93 key="",
94 data=notification_data_pb2.BadgeAdd(
95 badge_id="volunteer",
96 badge_name="Active Volunteer",
97 badge_description="This user is an active volunteer for Couchers.org",
98 ),
99 )
101 process_job()
103 with session_scope() as session:
104 deliv = session.execute(
105 select(NotificationDelivery)
106 .join(Notification, Notification.id == NotificationDelivery.notification_id)
107 .where(Notification.user_id == user.id)
108 .where(Notification.topic_action == topic_action)
109 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
110 ).scalar_one_or_none()
112 if enabled:
113 assert deliv is not None
114 else:
115 assert deliv is None
118def test_SetNotificationSettings_preferences_not_editable(db):
119 user, token = generate_user()
121 # enable a notification type and check it gets delivered
122 topic_action = NotificationTopicAction.password_reset__start
124 with notifications_session(token) as notifications:
125 with pytest.raises(grpc.RpcError) as e:
126 notifications.SetNotificationSettings(
127 notifications_pb2.SetNotificationSettingsReq(
128 preferences=[
129 notifications_pb2.SingleNotificationPreference(
130 topic=topic_action.topic,
131 action=topic_action.action,
132 delivery_method="push",
133 enabled=False,
134 )
135 ],
136 )
137 )
138 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
139 assert e.value.details() == "That notification preference is not user editable."
142def test_unsubscribe(db):
143 # this is the ugliest test i've written
145 user, token = generate_user()
147 topic_action = NotificationTopicAction.badge__add
149 # first enable email notifs
150 with notifications_session(token) as notifications:
151 notifications.SetNotificationSettings(
152 notifications_pb2.SetNotificationSettingsReq(
153 preferences=[
154 notifications_pb2.SingleNotificationPreference(
155 topic=topic_action.topic,
156 action=topic_action.action,
157 delivery_method=method,
158 enabled=enabled,
159 )
160 for method, enabled in [("email", True), ("digest", False), ("push", False)]
161 ],
162 )
163 )
165 with mock_notification_email() as mock:
166 with session_scope() as session:
167 notify(
168 session,
169 user_id=user.id,
170 topic_action=topic_action,
171 key="",
172 data=notification_data_pb2.BadgeAdd(
173 badge_id="volunteer",
174 badge_name="Active Volunteer",
175 badge_description="This user is an active volunteer for Couchers.org",
176 ),
177 )
179 assert mock.call_count == 1
180 assert email_fields(mock).recipient == user.email
181 # very ugly
182 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
183 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html): 183 ↛ 204line 183 didn't jump to line 204 because the loop on line 183 didn't complete
184 if "payload" not in link:
185 continue
186 print(link)
187 url_parts = urlparse(html.unescape(link))
188 params = parse_qs(url_parts.query)
189 print(params["payload"][0])
190 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
191 if payload.HasField("topic_action"): 191 ↛ 183line 191 didn't jump to line 183 because the condition on line 191 was always true
192 with auth_api_session() as (auth_api, metadata_interceptor):
193 assert (
194 auth_api.Unsubscribe(
195 auth_pb2.UnsubscribeReq(
196 payload=b64decode(params["payload"][0]),
197 sig=b64decode(params["sig"][0]),
198 )
199 ).response
200 == "You've been unsubscribed from email notifications of that type."
201 )
202 break
203 else:
204 raise Exception("Didn't find link")
206 with notifications_session(token) as notifications:
207 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
209 for group in res.groups:
210 for topic in group.topics:
211 for item in topic.items:
212 if topic == topic_action.topic and item == topic_action.action: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 assert not item.email
215 with mock_notification_email() as mock:
216 with session_scope() as session:
217 notify(
218 session,
219 user_id=user.id,
220 topic_action=topic_action,
221 key="",
222 data=notification_data_pb2.BadgeAdd(
223 badge_id="volunteer",
224 badge_name="Active Volunteer",
225 badge_description="This user is an active volunteer for Couchers.org",
226 ),
227 )
229 assert mock.call_count == 0
232def test_unsubscribe_do_not_email(db, moderator):
233 user, token = generate_user()
235 _, token2 = generate_user(complete_profile=True)
236 with api_session(token2) as api:
237 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
238 res = api.ListFriendRequests(empty_pb2.Empty())
239 fr_id = res.sent[0].friend_request_id
241 # Moderator approves the friend request, which triggers the notification email
242 with mock_notification_email() as mock:
243 moderator.approve_friend_request(fr_id)
245 assert mock.call_count == 1
246 assert email_fields(mock).recipient == user.email
247 # very ugly
248 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=
249 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html): 249 ↛ 270line 249 didn't jump to line 270 because the loop on line 249 didn't complete
250 if "payload" not in link:
251 continue
252 print(link)
253 url_parts = urlparse(html.unescape(link))
254 params = parse_qs(url_parts.query)
255 print(params["payload"][0])
256 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0]))
257 if payload.HasField("do_not_email"):
258 with auth_api_session() as (auth_api, metadata_interceptor):
259 assert (
260 auth_api.Unsubscribe(
261 auth_pb2.UnsubscribeReq(
262 payload=b64decode(params["payload"][0]),
263 sig=b64decode(params["sig"][0]),
264 )
265 ).response
266 == "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."
267 )
268 break
269 else:
270 raise Exception("Didn't find link")
272 _, token3 = generate_user(complete_profile=True)
273 with api_session(token3) as api:
274 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
275 res = api.ListFriendRequests(empty_pb2.Empty())
276 fr_id3 = res.sent[0].friend_request_id
278 # Approving this friend request should NOT send an email since user has do_not_email set
279 with mock_notification_email() as mock:
280 moderator.approve_friend_request(fr_id3)
282 assert mock.call_count == 0
284 with session_scope() as session:
285 user_ = session.execute(select(User).where(User.id == user.id)).scalar_one()
286 assert user_.do_not_email
289def test_get_do_not_email(db):
290 _, token = generate_user()
292 with session_scope() as session:
293 user = session.execute(select(User)).scalar_one()
294 user.do_not_email = False
296 with notifications_session(token) as notifications:
297 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
298 assert not res.do_not_email_enabled
300 with session_scope() as session:
301 user = session.execute(select(User)).scalar_one()
302 user.do_not_email = True
303 user.hosting_status = HostingStatus.cant_host
304 user.meetup_status = MeetupStatus.does_not_want_to_meetup
306 with notifications_session(token) as notifications:
307 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq())
308 assert res.do_not_email_enabled
311def test_set_do_not_email(db):
312 _, token = generate_user()
314 with session_scope() as session:
315 user = session.execute(select(User)).scalar_one()
316 user.do_not_email = False
317 user.hosting_status = HostingStatus.can_host
318 user.meetup_status = MeetupStatus.wants_to_meetup
320 with notifications_session(token) as notifications:
321 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
323 with session_scope() as session:
324 user = session.execute(select(User)).scalar_one()
325 assert not user.do_not_email
327 with notifications_session(token) as notifications:
328 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
330 with session_scope() as session:
331 user = session.execute(select(User)).scalar_one()
332 assert user.do_not_email
333 assert user.hosting_status == HostingStatus.cant_host
334 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup
336 with notifications_session(token) as notifications:
337 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False))
339 with session_scope() as session:
340 user = session.execute(select(User)).scalar_one()
341 assert not user.do_not_email
344def test_list_notifications(db, push_collector: PushCollector, moderator):
345 user1, token1 = generate_user()
346 user2, token2 = generate_user()
348 with api_session(token2) as api:
349 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
350 res = api.ListFriendRequests(empty_pb2.Empty())
351 fr_id = res.sent[0].friend_request_id
353 # Moderator approves the friend request so the notification is sent
354 moderator.approve_friend_request(fr_id)
356 with notifications_session(token1) as notifications:
357 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
358 assert len(res.notifications) == 1
360 n = res.notifications[0]
362 assert n.topic == "friend_request"
363 assert n.action == "create"
364 assert n.key == str(user2.id)
365 assert n.title == f"Friend request from {user2.name}"
366 assert n.body == f"{user2.name} wants to be your friend."
367 assert n.icon.startswith("http://localhost:5001/img/thumbnail/")
368 assert n.url == "http://localhost:3000/connections/friends/"
370 with conversations_session(token2) as c:
371 res = c.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[user1.id]))
372 group_chat_id = res.group_chat_id
373 moderator.approve_group_chat(group_chat_id)
374 for i in range(17):
375 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text=f"Test message {i}"))
377 process_jobs()
379 all_notifs = []
380 with notifications_session(token1) as notifications:
381 page_token = None
382 for _ in range(100): 382 ↛ 395line 382 didn't jump to line 395
383 res = notifications.ListNotifications(
384 notifications_pb2.ListNotificationsReq(
385 page_size=5,
386 page_token=page_token,
387 )
388 )
389 assert len(res.notifications) == 5 or not res.next_page_token
390 all_notifs += res.notifications
391 page_token = res.next_page_token
392 if not page_token:
393 break
395 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"{user2.name} wants to be your friend."]
396 assert bodys == [n.body for n in all_notifs]
399def test_notifications_seen(db, push_collector: PushCollector, moderator):
400 user1, token1 = generate_user()
401 user2, token2 = generate_user()
402 user3, token3 = generate_user()
403 user4, token4 = generate_user()
405 with api_session(token2) as api:
406 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
407 res = api.ListFriendRequests(empty_pb2.Empty())
408 fr_id2 = res.sent[0].friend_request_id
410 with api_session(token3) as api:
411 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
412 res = api.ListFriendRequests(empty_pb2.Empty())
413 fr_id3 = res.sent[0].friend_request_id
415 # Moderator approves the friend requests so notifications are sent
416 moderator.approve_friend_request(fr_id2)
417 moderator.approve_friend_request(fr_id3)
419 with notifications_session(token1) as notifications, api_session(token1) as api:
420 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
421 assert len(res.notifications) == 2
422 assert [n.is_seen for n in res.notifications] == [False, False]
423 notification_ids = [n.notification_id for n in res.notifications]
424 # should be listed desc time
425 assert notification_ids[0] > notification_ids[1]
427 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
429 with api_session(token4) as api:
430 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
431 res = api.ListFriendRequests(empty_pb2.Empty())
432 fr_id4 = res.sent[0].friend_request_id
434 # Moderator approves the friend request so notification is sent
435 moderator.approve_friend_request(fr_id4)
437 with notifications_session(token1) as notifications, api_session(token1) as api:
438 # mark everything before just the last one as seen (pretend we didn't load the last one yet in the api)
439 notifications.MarkAllNotificationsSeen(
440 notifications_pb2.MarkAllNotificationsSeenReq(latest_notification_id=notification_ids[0])
441 )
443 # last one is still unseen
444 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
446 # mark the first one unseen
447 notifications.MarkNotificationSeen(
448 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids[1], set_seen=False)
449 )
450 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
452 # mark the last one seen
453 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
454 assert len(res.notifications) == 3
455 assert [n.is_seen for n in res.notifications] == [False, True, False]
456 notification_ids2 = [n.notification_id for n in res.notifications]
458 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2
460 notifications.MarkNotificationSeen(
461 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids2[0], set_seen=True)
462 )
464 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq())
465 assert len(res.notifications) == 3
466 assert [n.is_seen for n in res.notifications] == [True, True, False]
468 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1
471def test_GetVapidPublicKey(db):
472 _, token = generate_user()
474 with notifications_session(token) as notifications:
475 assert (
476 notifications.GetVapidPublicKey(empty_pb2.Empty()).vapid_public_key
477 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"
478 )
481def test_RegisterPushNotificationSubscription(db):
482 _, token = generate_user()
484 subscription_info = {
485 "endpoint": "https://updates.push.services.mozilla.com/wpush/v2/gAAAAABmW2_iYKVyZRJPhAhktbkXd6Bc8zjIUvtVi5diYL7ZYn8FHka94kIdF46Mp8DwCDWlACnbKOEo97ikaa7JYowGLiGz3qsWL7Vo19LaV4I71mUDUOIKxWIsfp_kM77MlRJQKDUddv-sYyiffOyg63d1lnc_BMIyLXt69T5SEpfnfWTNb6I",
486 "expirationTime": None,
487 "keys": {
488 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",
489 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",
490 },
491 }
493 with notifications_session(token) as notifications:
494 res = notifications.RegisterPushNotificationSubscription(
495 notifications_pb2.RegisterPushNotificationSubscriptionReq(
496 full_subscription_json=json.dumps(subscription_info),
497 )
498 )
501def test_SendTestPushNotification(db, push_collector: PushCollector):
502 user, token = generate_user()
504 with notifications_session(token) as notifications:
505 notifications.SendTestPushNotification(empty_pb2.Empty())
507 assert push_collector.count_for_user(user.id) == 1
508 push = push_collector.pop_for_user(user.id, last=True)
509 assert push.content.title == "Push notifications test"
510 assert push.content.body == "If you see this, then it's working :)"
513def test_SendBlogPostNotification(db, push_collector: PushCollector):
514 super_user, super_token = generate_user(is_superuser=True)
516 user1, user1_token = generate_user()
517 # enabled email
518 user2, user2_token = generate_user()
519 # disabled push
520 user3, user3_token = generate_user()
522 topic_action = NotificationTopicAction.general__new_blog_post
524 with notifications_session(user2_token) as notifications:
525 notifications.SetNotificationSettings(
526 notifications_pb2.SetNotificationSettingsReq(
527 preferences=[
528 notifications_pb2.SingleNotificationPreference(
529 topic=topic_action.topic,
530 action=topic_action.action,
531 delivery_method="email",
532 enabled=True,
533 )
534 ],
535 )
536 )
538 with notifications_session(user3_token) as notifications:
539 notifications.SetNotificationSettings(
540 notifications_pb2.SetNotificationSettingsReq(
541 preferences=[
542 notifications_pb2.SingleNotificationPreference(
543 topic=topic_action.topic,
544 action=topic_action.action,
545 delivery_method="push",
546 enabled=False,
547 )
548 ],
549 )
550 )
552 with mock_notification_email() as mock:
553 with real_editor_session(super_token) as editor_api:
554 editor_api.SendBlogPostNotification(
555 editor_pb2.SendBlogPostNotificationReq(
556 title="Couchers.org v0.9.9 Release Notes",
557 blurb="Read about last major updates before v1!",
558 url="https://couchers.org/blog/2025/05/11/v0.9.9-release",
559 )
560 )
562 process_jobs()
564 assert mock.call_count == 1
565 assert email_fields(mock).recipient == user2.email
566 assert "Couchers.org v0.9.9 Release Notes" in email_fields(mock).html
567 assert "Couchers.org v0.9.9 Release Notes" in email_fields(mock).plain
568 assert "Read about last major updates before v1!" in email_fields(mock).html
569 assert "Read about last major updates before v1!" in email_fields(mock).plain
570 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email_fields(mock).html
571 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email_fields(mock).plain
573 push = push_collector.pop_for_user(user1.id, last=True)
574 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
575 assert push.content.body == "Read about last major updates before v1!"
576 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
578 push = push_collector.pop_for_user(user2.id, last=True)
579 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes"
580 assert push.content.body == "Read about last major updates before v1!"
581 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release"
583 assert push_collector.count_for_user(user3.id) == 0
586def test_get_topic_actions_by_delivery_type(db):
587 user, token = generate_user()
589 # these are enabled by default
590 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults
591 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults
593 # these are disabled by default
594 assert NotificationDeliveryType.push not in NotificationTopicAction.event__create_any.defaults
595 assert NotificationDeliveryType.push not in NotificationTopicAction.discussion__create.defaults
597 with notifications_session(token) as notifications:
598 notifications.SetNotificationSettings(
599 notifications_pb2.SetNotificationSettingsReq(
600 preferences=[
601 notifications_pb2.SingleNotificationPreference(
602 topic=NotificationTopicAction.reference__receive_friend.topic,
603 action=NotificationTopicAction.reference__receive_friend.action,
604 delivery_method="push",
605 enabled=False,
606 ),
607 notifications_pb2.SingleNotificationPreference(
608 topic=NotificationTopicAction.event__create_any.topic,
609 action=NotificationTopicAction.event__create_any.action,
610 delivery_method="push",
611 enabled=True,
612 ),
613 ],
614 )
615 )
617 with session_scope() as session:
618 deliver = get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push)
619 assert NotificationTopicAction.reference__receive_friend not in deliver
620 assert NotificationTopicAction.host_request__accept in deliver
621 assert NotificationTopicAction.event__create_any in deliver
622 assert NotificationTopicAction.discussion__create not in deliver
623 assert NotificationTopicAction.account_deletion__start in deliver
626def test_event_reminder_email_sent(db):
627 user, token = generate_user()
628 title = "Board Game Night"
629 start_event_time = timestamp_pb2.Timestamp(seconds=1751690400)
631 expected_time_str = LocalizationContext.from_user(user).localize_datetime(start_event_time)
633 with mock_notification_email() as mock:
634 with session_scope() as session:
635 user_in_session = session.get_one(User, user.id)
637 notify(
638 session,
639 user_id=user.id,
640 topic_action=NotificationTopicAction.event__reminder,
641 key="",
642 data=notification_data_pb2.EventReminder(
643 event=events_pb2.Event(
644 event_id=1,
645 slug="board-game-night",
646 title=title,
647 start_time=start_event_time,
648 ),
649 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)),
650 ),
651 )
653 assert mock.call_count == 1
654 assert email_fields(mock).recipient == user.email
655 assert title in email_fields(mock).html
656 assert title in email_fields(mock).plain
657 assert expected_time_str in email_fields(mock).html
658 assert expected_time_str in email_fields(mock).plain
661def test_RegisterMobilePushNotificationSubscription(db):
662 user, token = generate_user()
664 with notifications_session(token) as notifications:
665 notifications.RegisterMobilePushNotificationSubscription(
666 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
667 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]",
668 device_name="My iPhone",
669 device_type="ios",
670 )
671 )
673 # Check subscription was created
674 with session_scope() as session:
675 sub = session.execute(
676 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
677 ).scalar_one()
678 assert sub.platform == PushNotificationPlatform.expo
679 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]"
680 assert sub.device_name == "My iPhone"
681 assert sub.device_type == DeviceType.ios
682 assert sub.disabled_at == DATETIME_INFINITY
685def test_RegisterMobilePushNotificationSubscription_android(db):
686 user, token = generate_user()
688 with notifications_session(token) as notifications:
689 notifications.RegisterMobilePushNotificationSubscription(
690 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
691 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]",
692 device_name="My Android",
693 device_type="android",
694 )
695 )
697 with session_scope() as session:
698 sub = session.execute(
699 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
700 ).scalar_one()
701 assert sub.platform == PushNotificationPlatform.expo
702 assert sub.device_type == DeviceType.android
705def test_RegisterMobilePushNotificationSubscription_no_device_type(db):
706 user, token = generate_user()
708 with notifications_session(token) as notifications:
709 notifications.RegisterMobilePushNotificationSubscription(
710 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
711 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]",
712 )
713 )
715 with session_scope() as session:
716 sub = session.execute(
717 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id)
718 ).scalar_one()
719 assert sub.platform == PushNotificationPlatform.expo
720 assert sub.device_name is None
721 assert sub.device_type is None
724def test_RegisterMobilePushNotificationSubscription_re_enable(db):
725 user, token = generate_user()
727 # Create a disabled subscription directly in the DB
728 with session_scope() as session:
729 sub = PushNotificationSubscription(
730 user_id=user.id,
731 platform=PushNotificationPlatform.expo,
732 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
733 device_name="Old Device",
734 device_type=DeviceType.ios,
735 )
736 sub.disabled_at = now()
737 session.add(sub)
738 session.flush()
739 sub_id = sub.id
741 # Re-register with the same token
742 with notifications_session(token) as notifications:
743 notifications.RegisterMobilePushNotificationSubscription(
744 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
745 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]",
746 device_name="New Device Name",
747 device_type="android",
748 )
749 )
751 # Check subscription was re-enabled and updated
752 with session_scope() as session:
753 sub = session.execute(
754 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
755 ).scalar_one()
756 assert sub.disabled_at == DATETIME_INFINITY
757 assert sub.device_name == "New Device Name"
758 assert sub.device_type == DeviceType.android
761def test_RegisterMobilePushNotificationSubscription_already_exists(db):
762 user, token = generate_user()
764 # Create an active subscription directly in the DB
765 with session_scope() as session:
766 sub = PushNotificationSubscription(
767 user_id=user.id,
768 platform=PushNotificationPlatform.expo,
769 token="ExponentPushToken[existingtoken]",
770 device_name="Existing Device",
771 device_type=DeviceType.ios,
772 )
773 session.add(sub)
775 # Try to register with the same token - should just return without error
776 with notifications_session(token) as notifications:
777 notifications.RegisterMobilePushNotificationSubscription(
778 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq(
779 token="ExponentPushToken[existingtoken]",
780 device_name="Different Name",
781 )
782 )
784 # Check subscription was NOT modified (already active)
785 with session_scope() as session:
786 sub = session.execute(
787 select(PushNotificationSubscription).where(
788 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]"
789 )
790 ).scalar_one()
791 assert sub.device_name == "Existing Device" # unchanged
794def test_SendTestMobilePushNotification(db, push_collector: PushCollector):
795 user, token = generate_user()
797 with notifications_session(token) as notifications:
798 notifications.SendTestMobilePushNotification(empty_pb2.Empty())
800 push = push_collector.pop_for_user(user.id, last=True)
801 assert push.content.title == "Mobile notifications test"
802 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉"
805def test_get_expo_push_receipts(db):
806 mock_response = Mock()
807 mock_response.status_code = 200
808 mock_response.json.return_value = {
809 "data": {
810 "ticket-1": {"status": "ok"},
811 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
812 }
813 }
815 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post:
816 result = get_expo_push_receipts(["ticket-1", "ticket-2"])
818 mock_post.assert_called_once()
819 call_args = mock_post.call_args
820 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts"
821 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}
823 assert result == {
824 "ticket-1": {"status": "ok"},
825 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}},
826 }
829def test_get_expo_push_receipts_empty(db):
830 result = get_expo_push_receipts([])
831 assert result == {}
834def test_check_expo_push_receipts_success(db):
835 """Test batch receipt checking with successful delivery."""
836 user, token = generate_user()
838 # Create a push subscription and delivery attempt (old enough to be checked)
839 with session_scope() as session:
840 sub = PushNotificationSubscription(
841 user_id=user.id,
842 platform=PushNotificationPlatform.expo,
843 token="ExponentPushToken[testtoken123]",
844 device_name="Test Device",
845 device_type=DeviceType.ios,
846 )
847 session.add(sub)
848 session.flush()
850 attempt = PushNotificationDeliveryAttempt(
851 push_notification_subscription_id=sub.id,
852 outcome=PushNotificationDeliveryOutcome.success,
853 status_code=200,
854 expo_ticket_id="test-ticket-id",
855 )
856 session.add(attempt)
857 session.flush()
858 # Make the attempt old enough to be checked (>15 min)
859 attempt.time = now() - timedelta(minutes=20)
860 attempt_id = attempt.id
861 sub_id = sub.id
863 # Mock the receipt API call
864 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
865 mock_post.return_value.status_code = 200
866 mock_post.return_value.json.return_value = {"data": {"test-ticket-id": {"status": "ok"}}}
868 check_expo_push_receipts(empty_pb2.Empty())
870 # Verify the attempt was updated
871 with session_scope() as session:
872 attempt = session.execute(
873 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
874 ).scalar_one()
875 assert attempt.receipt_checked_at is not None
876 assert attempt.receipt_status == "ok"
877 assert attempt.receipt_error_code is None
879 # Subscription should still be enabled
880 sub = session.execute(
881 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
882 ).scalar_one()
883 assert sub.disabled_at == DATETIME_INFINITY
886def test_check_expo_push_receipts_device_not_registered(db):
887 """Test batch receipt checking with DeviceNotRegistered error disables subscription."""
888 user, token = generate_user()
890 # Create a push subscription and delivery attempt
891 with session_scope() as session:
892 sub = PushNotificationSubscription(
893 user_id=user.id,
894 platform=PushNotificationPlatform.expo,
895 token="ExponentPushToken[devicegone]",
896 device_name="Test Device",
897 device_type=DeviceType.android,
898 )
899 session.add(sub)
900 session.flush()
902 attempt = PushNotificationDeliveryAttempt(
903 push_notification_subscription_id=sub.id,
904 outcome=PushNotificationDeliveryOutcome.success,
905 status_code=200,
906 expo_ticket_id="ticket-device-gone",
907 )
908 session.add(attempt)
909 session.flush()
910 # Make the attempt old enough to be checked
911 attempt.time = now() - timedelta(minutes=15)
912 attempt_id = attempt.id
913 sub_id = sub.id
915 # Mock the receipt API call with DeviceNotRegistered error
916 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
917 mock_post.return_value.status_code = 200
918 mock_post.return_value.json.return_value = {
919 "data": {
920 "ticket-device-gone": {
921 "status": "error",
922 "details": {"error": "DeviceNotRegistered"},
923 }
924 }
925 }
927 check_expo_push_receipts(empty_pb2.Empty())
929 # Verify the attempt was updated and subscription disabled
930 with session_scope() as session:
931 attempt = session.execute(
932 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
933 ).scalar_one()
934 assert attempt.receipt_checked_at is not None
935 assert attempt.receipt_status == "error"
936 assert attempt.receipt_error_code == "DeviceNotRegistered"
938 # Subscription should be disabled
939 sub = session.execute(
940 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
941 ).scalar_one()
942 assert sub.disabled_at <= now()
945def test_check_expo_push_receipts_not_found(db):
946 """Test batch receipt checking when ticket not found (expired)."""
947 user, token = generate_user()
949 with session_scope() as session:
950 sub = PushNotificationSubscription(
951 user_id=user.id,
952 platform=PushNotificationPlatform.expo,
953 token="ExponentPushToken[notfound]",
954 )
955 session.add(sub)
956 session.flush()
958 attempt = PushNotificationDeliveryAttempt(
959 push_notification_subscription_id=sub.id,
960 outcome=PushNotificationDeliveryOutcome.success,
961 status_code=200,
962 expo_ticket_id="unknown-ticket",
963 )
964 session.add(attempt)
965 session.flush()
966 # Make the attempt old enough to be checked
967 attempt.time = now() - timedelta(minutes=15)
968 attempt_id = attempt.id
969 sub_id = sub.id
971 # Mock empty receipt response (ticket not found)
972 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
973 mock_post.return_value.status_code = 200
974 mock_post.return_value.json.return_value = {"data": {}}
976 check_expo_push_receipts(empty_pb2.Empty())
978 with session_scope() as session:
979 attempt = session.execute(
980 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
981 ).scalar_one()
982 assert attempt.receipt_checked_at is not None
983 assert attempt.receipt_status == "not_found"
985 # Subscription should still be enabled
986 sub = session.execute(
987 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id)
988 ).scalar_one()
989 assert sub.disabled_at == DATETIME_INFINITY
992def test_check_expo_push_receipts_skips_already_checked(db):
993 """Test that already-checked receipts are not re-checked."""
994 user, token = generate_user()
996 # Create an attempt that was already checked
997 with session_scope() as session:
998 sub = PushNotificationSubscription(
999 user_id=user.id,
1000 platform=PushNotificationPlatform.expo,
1001 token="ExponentPushToken[alreadychecked]",
1002 )
1003 session.add(sub)
1004 session.flush()
1006 attempt = PushNotificationDeliveryAttempt(
1007 push_notification_subscription_id=sub.id,
1008 outcome=PushNotificationDeliveryOutcome.success,
1009 status_code=200,
1010 expo_ticket_id="already-checked-ticket",
1011 receipt_checked_at=now(),
1012 receipt_status="ok",
1013 )
1014 session.add(attempt)
1015 session.flush()
1016 # Make the attempt old enough
1017 attempt.time = now() - timedelta(minutes=15)
1019 # Should not call the API since the only attempt is already checked
1020 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1021 check_expo_push_receipts(empty_pb2.Empty())
1022 mock_post.assert_not_called()
1025def test_SendDevPushNotification_success(db, push_collector: PushCollector):
1026 """Test SendDevPushNotification sends push with all specified parameters."""
1027 user, token = generate_user()
1029 # Enable dev APIs for this test
1030 config["ENABLE_DEV_APIS"] = True
1032 with notifications_session(token) as notifications:
1033 notifications.SendDevPushNotification(
1034 notifications_pb2.SendDevPushNotificationReq(
1035 title="Test Dev Title",
1036 body="Test dev notification body",
1037 icon="https://example.com/icon.png",
1038 url="https://example.com/action",
1039 key="test-key",
1040 ttl=3600,
1041 )
1042 )
1044 push = push_collector.pop_for_user(user.id, last=True)
1045 assert push.content.title == "Test Dev Title"
1046 assert push.content.body == "Test dev notification body"
1047 assert push.content.action_url == "https://example.com/action"
1048 assert push.content.icon_url == "https://example.com/icon.png"
1049 assert push.topic_action == "adhoc:testing"
1050 assert push.key == "test-key"
1051 assert push.ttl == 3600
1054def test_SendDevPushNotification_minimal(db, push_collector: PushCollector):
1055 """Test SendDevPushNotification with minimal parameters."""
1056 user, token = generate_user()
1058 config["ENABLE_DEV_APIS"] = True
1060 with notifications_session(token) as notifications:
1061 notifications.SendDevPushNotification(
1062 notifications_pb2.SendDevPushNotificationReq(
1063 title="Minimal Title",
1064 body="Minimal body",
1065 )
1066 )
1068 push = push_collector.pop_for_user(user.id, last=True)
1069 assert push.content.title == "Minimal Title"
1070 assert push.content.body == "Minimal body"
1071 assert push.topic_action == "adhoc:testing"
1074def test_SendDevPushNotification_disabled(db, push_collector: PushCollector):
1075 """Test SendDevPushNotification fails when ENABLE_DEV_APIS is disabled."""
1076 user, token = generate_user()
1078 # Ensure dev APIs are disabled (default in tests)
1079 config["ENABLE_DEV_APIS"] = False
1081 with notifications_session(token) as notifications:
1082 with pytest.raises(grpc.RpcError) as e:
1083 notifications.SendDevPushNotification(
1084 notifications_pb2.SendDevPushNotificationReq(
1085 title="Should Fail",
1086 body="This should not be sent",
1087 )
1088 )
1089 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1090 assert "Development APIs are not enabled" in not_none(e.value.details())
1092 assert push_collector.count_for_user(user.id) == 0
1095def test_SendDevPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1096 """Test SendDevPushNotification fails when push notifications are disabled."""
1097 user, token = generate_user()
1099 config["ENABLE_DEV_APIS"] = True
1100 config["PUSH_NOTIFICATIONS_ENABLED"] = False
1102 with notifications_session(token) as notifications:
1103 with pytest.raises(grpc.RpcError) as e:
1104 notifications.SendDevPushNotification(
1105 notifications_pb2.SendDevPushNotificationReq(
1106 title="Should Fail",
1107 body="This should not be sent",
1108 )
1109 )
1110 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1111 assert "Push notifications are currently disabled" in not_none(e.value.details())
1113 assert push_collector.count_for_user(user.id) == 0
1116def test_check_expo_push_receipts_skips_too_recent(db):
1117 """Test that too-recent receipts (<15 min) are not checked."""
1118 user, token = generate_user()
1120 # Create a recent attempt (not old enough to check)
1121 with session_scope() as session:
1122 sub = PushNotificationSubscription(
1123 user_id=user.id,
1124 platform=PushNotificationPlatform.expo,
1125 token="ExponentPushToken[recent]",
1126 )
1127 session.add(sub)
1128 session.flush()
1130 attempt = PushNotificationDeliveryAttempt(
1131 push_notification_subscription_id=sub.id,
1132 outcome=PushNotificationDeliveryOutcome.success,
1133 status_code=200,
1134 expo_ticket_id="recent-ticket",
1135 )
1136 session.add(attempt)
1137 session.flush()
1138 # Make the attempt only 5 minutes old (too recent)
1139 attempt.time = now() - timedelta(minutes=5)
1141 # Should not call the API since the attempt is too recent
1142 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1143 check_expo_push_receipts(empty_pb2.Empty())
1144 mock_post.assert_not_called()
1147def test_check_expo_push_receipts_batch(db):
1148 """Test that multiple receipts are checked in a single batch."""
1149 user, token = generate_user()
1151 # Create multiple delivery attempts
1152 attempt_ids = []
1153 with session_scope() as session:
1154 sub = PushNotificationSubscription(
1155 user_id=user.id,
1156 platform=PushNotificationPlatform.expo,
1157 token="ExponentPushToken[batch]",
1158 )
1159 session.add(sub)
1160 session.flush()
1162 for i in range(3):
1163 attempt = PushNotificationDeliveryAttempt(
1164 push_notification_subscription_id=sub.id,
1165 outcome=PushNotificationDeliveryOutcome.success,
1166 status_code=200,
1167 expo_ticket_id=f"batch-ticket-{i}",
1168 )
1169 session.add(attempt)
1170 session.flush()
1171 attempt.time = now() - timedelta(minutes=20)
1172 attempt_ids.append(attempt.id)
1174 # Mock the batch receipt API call
1175 with patch("couchers.notifications.expo_api.requests.post") as mock_post:
1176 mock_post.return_value.status_code = 200
1177 mock_post.return_value.json.return_value = {
1178 "data": {
1179 "batch-ticket-0": {"status": "ok"},
1180 "batch-ticket-1": {"status": "ok"},
1181 "batch-ticket-2": {"status": "ok"},
1182 }
1183 }
1185 check_expo_push_receipts(empty_pb2.Empty())
1187 # Should only call the API once for all tickets
1188 assert mock_post.call_count == 1
1190 # Verify all attempts were updated
1191 with session_scope() as session:
1192 for attempt_id in attempt_ids:
1193 attempt = session.execute(
1194 select(PushNotificationDeliveryAttempt).where(PushNotificationDeliveryAttempt.id == attempt_id)
1195 ).scalar_one()
1196 assert attempt.receipt_checked_at is not None
1197 assert attempt.receipt_status == "ok"
1200def test_DebugRedeliverPushNotification_success(db, push_collector: PushCollector):
1201 """Test DebugRedeliverPushNotification redelivers an existing notification."""
1202 user, token = generate_user()
1204 config["ENABLE_DEV_APIS"] = True
1206 # Create a notification for the user
1207 with session_scope() as session:
1208 notify(
1209 session,
1210 user_id=user.id,
1211 topic_action=NotificationTopicAction.badge__add,
1212 key="test-badge",
1213 data=notification_data_pb2.BadgeAdd(
1214 badge_id="volunteer",
1215 badge_name="Active Volunteer",
1216 badge_description="This user is an active volunteer for Couchers.org",
1217 ),
1218 )
1220 process_job()
1222 # Pop the initial push notification
1223 push_collector.pop_for_user(user.id, last=True)
1225 # Get the notification_id
1226 with session_scope() as session:
1227 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1228 notification_id = notification.id
1230 # Redeliver the notification
1231 with notifications_session(token) as notifications:
1232 notifications.DebugRedeliverPushNotification(
1233 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1234 )
1236 # Verify a new push was sent
1237 push = push_collector.pop_for_user(user.id, last=True)
1238 assert "Active Volunteer" in push.content.title
1239 assert push.topic_action == "badge:add"
1240 assert push.key == "test-badge"
1243def test_DebugRedeliverPushNotification_not_found(db, push_collector: PushCollector):
1244 """Test DebugRedeliverPushNotification fails when notification doesn't exist."""
1245 user, token = generate_user()
1247 config["ENABLE_DEV_APIS"] = True
1249 with notifications_session(token) as notifications:
1250 with pytest.raises(grpc.RpcError) as e:
1251 notifications.DebugRedeliverPushNotification(
1252 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=999999)
1253 )
1254 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1255 assert "notification not found" in not_none(e.value.details()).lower()
1257 assert push_collector.count_for_user(user.id) == 0
1260def test_DebugRedeliverPushNotification_wrong_user(db, push_collector: PushCollector):
1261 """Test DebugRedeliverPushNotification fails when notification belongs to another user."""
1262 user1, token1 = generate_user()
1263 user2, token2 = generate_user()
1265 config["ENABLE_DEV_APIS"] = True
1267 # Create a notification for user1
1268 with session_scope() as session:
1269 notify(
1270 session,
1271 user_id=user1.id,
1272 topic_action=NotificationTopicAction.badge__add,
1273 key="test-badge",
1274 data=notification_data_pb2.BadgeAdd(
1275 badge_id="volunteer",
1276 badge_name="Active Volunteer",
1277 badge_description="This user is an active volunteer for Couchers.org",
1278 ),
1279 )
1281 process_job()
1283 # Get the notification_id
1284 with session_scope() as session:
1285 notification = session.execute(select(Notification).where(Notification.user_id == user1.id)).scalar_one()
1286 notification_id = notification.id
1288 # user2 tries to redeliver user1's notification
1289 with notifications_session(token2) as notifications:
1290 with pytest.raises(grpc.RpcError) as e:
1291 notifications.DebugRedeliverPushNotification(
1292 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=notification_id)
1293 )
1294 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1295 assert "notification not found" in not_none(e.value.details()).lower()
1297 assert push_collector.count_for_user(user2.id) == 0
1300def test_DebugRedeliverPushNotification_disabled(db, push_collector: PushCollector):
1301 """Test DebugRedeliverPushNotification fails when ENABLE_DEV_APIS is disabled."""
1302 user, token = generate_user()
1304 config["ENABLE_DEV_APIS"] = False
1306 with notifications_session(token) as notifications:
1307 with pytest.raises(grpc.RpcError) as e:
1308 notifications.DebugRedeliverPushNotification(
1309 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1310 )
1311 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1312 assert "Development APIs are not enabled" in not_none(e.value.details())
1314 assert push_collector.count_for_user(user.id) == 0
1317def test_DebugRedeliverPushNotification_push_notifications_disabled(db, push_collector: PushCollector):
1318 """Test DebugRedeliverPushNotification fails when push notifications are disabled."""
1319 user, token = generate_user()
1321 config["ENABLE_DEV_APIS"] = True
1322 config["PUSH_NOTIFICATIONS_ENABLED"] = False
1324 with notifications_session(token) as notifications:
1325 with pytest.raises(grpc.RpcError) as e:
1326 notifications.DebugRedeliverPushNotification(
1327 notifications_pb2.DebugRedeliverPushNotificationReq(notification_id=1)
1328 )
1329 assert e.value.code() == grpc.StatusCode.UNAVAILABLE
1330 assert "Push notifications are currently disabled" in not_none(e.value.details())
1332 assert push_collector.count_for_user(user.id) == 0
1335def test_handle_notification_email_delivery(db):
1336 """Test that email notifications are delivered when email preference is enabled."""
1337 user, token = generate_user()
1339 topic_action = NotificationTopicAction.badge__add
1341 # Enable email notifications for this topic
1342 with notifications_session(token) as notifications:
1343 notifications.SetNotificationSettings(
1344 notifications_pb2.SetNotificationSettingsReq(
1345 preferences=[
1346 notifications_pb2.SingleNotificationPreference(
1347 topic=topic_action.topic,
1348 action=topic_action.action,
1349 delivery_method="email",
1350 enabled=True,
1351 )
1352 ],
1353 )
1354 )
1356 with mock_notification_email() as mock:
1357 with session_scope() as session:
1358 notify(
1359 session,
1360 user_id=user.id,
1361 topic_action=topic_action,
1362 key="test-badge",
1363 data=notification_data_pb2.BadgeAdd(
1364 badge_id="volunteer",
1365 badge_name="Active Volunteer",
1366 badge_description="This user is an active volunteer",
1367 ),
1368 )
1370 assert mock.call_count == 1
1371 assert email_fields(mock).recipient == user.email
1373 with session_scope() as session:
1374 delivery = session.execute(
1375 select(NotificationDelivery)
1376 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1377 .where(Notification.user_id == user.id)
1378 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.email)
1379 ).scalar_one()
1380 assert delivery.delivered is not None
1383def test_handle_notification_push_delivery(db, push_collector: PushCollector):
1384 """Test that push notifications are delivered immediately when push preference is enabled."""
1385 user, token = generate_user()
1387 topic_action = NotificationTopicAction.badge__add
1389 with session_scope() as session:
1390 notify(
1391 session,
1392 user_id=user.id,
1393 topic_action=topic_action,
1394 key="test-badge",
1395 data=notification_data_pb2.BadgeAdd(
1396 badge_id="volunteer",
1397 badge_name="Active Volunteer",
1398 badge_description="This user is an active volunteer",
1399 ),
1400 )
1402 process_job()
1404 push = push_collector.pop_for_user(user.id, last=True)
1405 assert "Active Volunteer" in push.content.title
1407 with session_scope() as session:
1408 delivery = session.execute(
1409 select(NotificationDelivery)
1410 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1411 .where(Notification.user_id == user.id)
1412 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1413 ).scalar_one()
1414 assert delivery.delivered is not None
1417def test_handle_notification_digest_delivery(db):
1418 """Test that digest notifications are queued without a delivered timestamp."""
1419 user, token = generate_user()
1421 topic_action = NotificationTopicAction.badge__add
1423 # Enable only digest notifications for this topic
1424 with notifications_session(token) as notifications:
1425 notifications.SetNotificationSettings(
1426 notifications_pb2.SetNotificationSettingsReq(
1427 preferences=[
1428 notifications_pb2.SingleNotificationPreference(
1429 topic=topic_action.topic,
1430 action=topic_action.action,
1431 delivery_method="push",
1432 enabled=False,
1433 ),
1434 notifications_pb2.SingleNotificationPreference(
1435 topic=topic_action.topic,
1436 action=topic_action.action,
1437 delivery_method="digest",
1438 enabled=True,
1439 ),
1440 ],
1441 )
1442 )
1444 with session_scope() as session:
1445 notify(
1446 session,
1447 user_id=user.id,
1448 topic_action=topic_action,
1449 key="test-badge",
1450 data=notification_data_pb2.BadgeAdd(
1451 badge_id="volunteer",
1452 badge_name="Active Volunteer",
1453 badge_description="This user is an active volunteer",
1454 ),
1455 )
1457 process_job()
1459 # Verify digest NotificationDelivery was created WITHOUT delivered timestamp
1460 with session_scope() as session:
1461 delivery = session.execute(
1462 select(NotificationDelivery)
1463 .join(Notification, Notification.id == NotificationDelivery.notification_id)
1464 .where(Notification.user_id == user.id)
1465 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.digest)
1466 ).scalar_one()
1467 assert delivery.delivered is None
1470def test_handle_notification_banned_user_no_email(db):
1471 """Test that banned users don't receive email notifications."""
1472 user, token = generate_user()
1474 topic_action = NotificationTopicAction.badge__add
1476 # Enable email notifications
1477 with notifications_session(token) as notifications:
1478 notifications.SetNotificationSettings(
1479 notifications_pb2.SetNotificationSettingsReq(
1480 preferences=[
1481 notifications_pb2.SingleNotificationPreference(
1482 topic=topic_action.topic,
1483 action=topic_action.action,
1484 delivery_method="email",
1485 enabled=True,
1486 )
1487 ],
1488 )
1489 )
1491 # Ban the user
1492 with session_scope() as session:
1493 session.execute(update(User).where(User.id == user.id).values(banned_at=now()))
1495 with mock_notification_email() as mock:
1496 with session_scope() as session:
1497 notify(
1498 session,
1499 user_id=user.id,
1500 topic_action=topic_action,
1501 key="test-badge",
1502 data=notification_data_pb2.BadgeAdd(
1503 badge_id="volunteer",
1504 badge_name="Active Volunteer",
1505 badge_description="This user is an active volunteer",
1506 ),
1507 )
1509 # Email should not be sent to the banned user
1510 assert mock.call_count == 0
1513def test_handle_notification_deleted_user_no_regular_email(db):
1514 """Test that deleted users don't receive non-account-deletion email notifications."""
1515 user, token = generate_user()
1517 topic_action = NotificationTopicAction.badge__add
1519 # Enable email notifications
1520 with notifications_session(token) as notifications:
1521 notifications.SetNotificationSettings(
1522 notifications_pb2.SetNotificationSettingsReq(
1523 preferences=[
1524 notifications_pb2.SingleNotificationPreference(
1525 topic=topic_action.topic,
1526 action=topic_action.action,
1527 delivery_method="email",
1528 enabled=True,
1529 )
1530 ],
1531 )
1532 )
1534 # Delete the user
1535 with session_scope() as session:
1536 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1538 with mock_notification_email() as mock:
1539 with session_scope() as session:
1540 notify(
1541 session,
1542 user_id=user.id,
1543 topic_action=topic_action,
1544 key="test-badge",
1545 data=notification_data_pb2.BadgeAdd(
1546 badge_id="volunteer",
1547 badge_name="Active Volunteer",
1548 badge_description="This user is an active volunteer",
1549 ),
1550 )
1552 # Email should not be sent to deleted user for non-account-deletion notification
1553 assert mock.call_count == 0
1556def test_handle_notification_deleted_user_receives_account_deletion_email(db):
1557 """Test that deleted users CAN receive account deletion notifications."""
1558 user, token = generate_user()
1560 topic_action = NotificationTopicAction.account_deletion__complete
1562 # Delete the user
1563 with session_scope() as session:
1564 session.execute(update(User).where(User.id == user.id).values(deleted_at=now()))
1566 with mock_notification_email() as mock:
1567 with session_scope() as session:
1568 notify(
1569 session,
1570 user_id=user.id,
1571 topic_action=topic_action,
1572 key="",
1573 data=notification_data_pb2.AccountDeletionComplete(
1574 undelete_token="test-token",
1575 undelete_days=7,
1576 ),
1577 )
1579 # Email SHOULD be sent to deleted user for account deletion notification
1580 assert mock.call_count == 1
1581 assert email_fields(mock).recipient == user.email
1584def test_handle_notification_do_not_email_respected(db):
1585 """Test that users with do_not_email set don't receive non-critical emails."""
1586 user, token = generate_user()
1588 topic_action = NotificationTopicAction.badge__add
1590 # Enable email notifications
1591 with notifications_session(token) as notifications:
1592 notifications.SetNotificationSettings(
1593 notifications_pb2.SetNotificationSettingsReq(
1594 preferences=[
1595 notifications_pb2.SingleNotificationPreference(
1596 topic=topic_action.topic,
1597 action=topic_action.action,
1598 delivery_method="email",
1599 enabled=True,
1600 )
1601 ],
1602 )
1603 )
1605 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1606 with session_scope() as session:
1607 session.execute(
1608 update(User)
1609 .where(User.id == user.id)
1610 .values(
1611 hosting_status=HostingStatus.cant_host,
1612 meetup_status=MeetupStatus.does_not_want_to_meetup,
1613 do_not_email=True,
1614 )
1615 )
1617 with mock_notification_email() as mock:
1618 with session_scope() as session:
1619 notify(
1620 session,
1621 user_id=user.id,
1622 topic_action=topic_action,
1623 key="test-badge",
1624 data=notification_data_pb2.BadgeAdd(
1625 badge_id="volunteer",
1626 badge_name="Active Volunteer",
1627 badge_description="This user is an active volunteer",
1628 ),
1629 )
1631 # Email should not be sent when do_not_email is True
1632 assert mock.call_count == 0
1635def test_handle_notification_critical_bypasses_do_not_email(db):
1636 """Test that critical notifications bypass do_not_email setting."""
1637 user, token = generate_user()
1639 topic_action = NotificationTopicAction.password__change
1641 # Set do_not_email (requires hosting/meetup status to be set due to DB constraint)
1642 with session_scope() as session:
1643 session.execute(
1644 update(User)
1645 .where(User.id == user.id)
1646 .values(
1647 hosting_status=HostingStatus.cant_host,
1648 meetup_status=MeetupStatus.does_not_want_to_meetup,
1649 do_not_email=True,
1650 )
1651 )
1653 with mock_notification_email() as mock:
1654 with session_scope() as session:
1655 notify(
1656 session,
1657 user_id=user.id,
1658 topic_action=topic_action,
1659 key="",
1660 data=None,
1661 )
1663 # Critical email SHOULD be sent even with do_not_email=True
1664 assert mock.call_count == 1
1665 assert email_fields(mock).recipient == user.email
1668def test_handle_notification_duplicate_delivery_skipped(db, push_collector: PushCollector):
1669 """Test that duplicate deliveries are skipped when NotificationDelivery already exists."""
1670 user, token = generate_user()
1672 topic_action = NotificationTopicAction.badge__add
1674 # Create notification manually
1675 with session_scope() as session:
1676 notification = Notification(
1677 user_id=user.id,
1678 topic_action=topic_action,
1679 key="test-badge",
1680 data=notification_data_pb2.BadgeAdd(
1681 badge_id="volunteer",
1682 badge_name="Active Volunteer",
1683 badge_description="This user is an active volunteer",
1684 ).SerializeToString(),
1685 )
1686 session.add(notification)
1687 session.flush()
1688 notification_id = notification.id
1690 # Manually create a push delivery (simulating it was already delivered)
1691 session.add(
1692 NotificationDelivery(
1693 notification_id=notification_id,
1694 delivery_type=NotificationDeliveryType.push,
1695 delivered=now(),
1696 )
1697 )
1699 # Try to handle the notification again
1700 handle_notification(jobs_pb2.HandleNotificationPayload(notification_id=notification_id))
1702 # No new push should be sent since delivery already exists
1703 assert push_collector.count_for_user(user.id) == 0
1705 # Verify only one delivery exists
1706 with session_scope() as session:
1707 delivery_count = len(
1708 session.execute(
1709 select(NotificationDelivery)
1710 .where(NotificationDelivery.notification_id == notification_id)
1711 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push)
1712 )
1713 .scalars()
1714 .all()
1715 )
1716 assert delivery_count == 1
1719def test_handle_notification_deferred_when_content_not_visible(db, moderator):
1720 """Test that notifications linked to non-visible moderated content are deferred."""
1721 user1, token1 = generate_user(complete_profile=True)
1722 user2, token2 = generate_user(complete_profile=True)
1724 # Create a friend request (which creates a moderation state)
1725 # This also queues a notification via SendFriendRequest
1726 with api_session(token2) as api:
1727 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1729 # Process the queued job (handle_notification)
1730 process_job()
1732 # The notification should exist but have no deliveries because content is shadowed
1733 with session_scope() as session:
1734 notification = session.execute(
1735 select(Notification)
1736 .where(Notification.user_id == user1.id)
1737 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1738 ).scalar_one()
1740 deliveries = (
1741 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1742 .scalars()
1743 .all()
1744 )
1745 # No deliveries because content is not yet visible (shadowed)
1746 assert len(deliveries) == 0
1749def test_handle_notification_delivered_when_content_visible(db, moderator):
1750 """Test that notifications linked to visible moderated content are delivered."""
1751 user1, token1 = generate_user(complete_profile=True)
1752 user2, token2 = generate_user(complete_profile=True)
1754 # Create a friend request
1755 with api_session(token2) as api:
1756 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id))
1757 res = api.ListFriendRequests(empty_pb2.Empty())
1758 fr_id = res.sent[0].friend_request_id
1760 # Process initial job (which is deferred because content is shadowed)
1761 process_job()
1763 # Approve the friend request so it becomes visible (this queues the notification job again)
1764 moderator.approve_friend_request(fr_id)
1766 # Process the notification job that was re-queued after approval
1767 process_jobs()
1769 # Notification should have been delivered
1770 with session_scope() as session:
1771 notification = session.execute(
1772 select(Notification)
1773 .where(Notification.user_id == user1.id)
1774 .where(Notification.topic_action == NotificationTopicAction.friend_request__create)
1775 ).scalar_one()
1777 deliveries = (
1778 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1779 .scalars()
1780 .all()
1781 )
1782 # At least one delivery should exist
1783 assert len(deliveries) > 0
1786def test_handle_notification_multiple_delivery_types(db, push_collector: PushCollector):
1787 """Test that multiple delivery types are processed for a single notification."""
1788 user, token = generate_user()
1790 topic_action = NotificationTopicAction.badge__add
1792 # Enable both email and push notifications
1793 with notifications_session(token) as notifications:
1794 notifications.SetNotificationSettings(
1795 notifications_pb2.SetNotificationSettingsReq(
1796 preferences=[
1797 notifications_pb2.SingleNotificationPreference(
1798 topic=topic_action.topic,
1799 action=topic_action.action,
1800 delivery_method="email",
1801 enabled=True,
1802 ),
1803 notifications_pb2.SingleNotificationPreference(
1804 topic=topic_action.topic,
1805 action=topic_action.action,
1806 delivery_method="push",
1807 enabled=True,
1808 ),
1809 notifications_pb2.SingleNotificationPreference(
1810 topic=topic_action.topic,
1811 action=topic_action.action,
1812 delivery_method="digest",
1813 enabled=True,
1814 ),
1815 ],
1816 )
1817 )
1819 with mock_notification_email() as mock:
1820 with session_scope() as session:
1821 notify(
1822 session,
1823 user_id=user.id,
1824 topic_action=topic_action,
1825 key="test-badge",
1826 data=notification_data_pb2.BadgeAdd(
1827 badge_id="volunteer",
1828 badge_name="Active Volunteer",
1829 badge_description="This user is an active volunteer",
1830 ),
1831 )
1833 # Email should be sent
1834 assert mock.call_count == 1
1836 # Push should be sent
1837 push = push_collector.pop_for_user(user.id, last=True)
1838 assert "Active Volunteer" in push.content.title
1840 # All three delivery types should have deliveries
1841 with session_scope() as session:
1842 notification = session.execute(select(Notification).where(Notification.user_id == user.id)).scalar_one()
1844 deliveries = (
1845 session.execute(select(NotificationDelivery).where(NotificationDelivery.notification_id == notification.id))
1846 .scalars()
1847 .all()
1848 )
1850 delivery_types = {d.delivery_type for d in deliveries}
1851 assert NotificationDeliveryType.email in delivery_types
1852 assert NotificationDeliveryType.push in delivery_types
1853 assert NotificationDeliveryType.digest in delivery_types
1855 # Email and push should have delivered timestamps
1856 for delivery in deliveries:
1857 if delivery.delivery_type in [NotificationDeliveryType.email, NotificationDeliveryType.push]:
1858 assert delivery.delivered is not None
1859 elif delivery.delivery_type == NotificationDeliveryType.digest: 1859 ↛ 1856line 1859 didn't jump to line 1856 because the condition on line 1859 was always true
1860 assert delivery.delivered is None