Coverage for app / backend / src / tests / test_notifications.py: 99%

1import html 

2import json 

3import re 

4from urllib.parse import parse_qs, urlparse 

5 

6import grpc 

7import pytest 

8from google.protobuf import empty_pb2, timestamp_pb2 

9from sqlalchemy import select 

10 

11from couchers.config import config 

12from couchers.constants import DATETIME_INFINITY 

13from couchers.context import make_background_user_context 

14from couchers.crypto import b64decode 

15from couchers.db import session_scope 

16from couchers.i18n import LocalizationContext 

17from couchers.jobs.worker import process_job 

18from couchers.models import ( 

19 DeviceType, 

20 HostingStatus, 

21 MeetupStatus, 

22 Notification, 

23 NotificationDelivery, 

24 NotificationDeliveryType, 

25 NotificationTopicAction, 

26 PushNotificationPlatform, 

27 PushNotificationSubscription, 

28 User, 

29) 

30from couchers.notifications.notify import notify 

31from couchers.notifications.settings import get_topic_actions_by_delivery_type 

32from couchers.proto import ( 

33 api_pb2, 

34 auth_pb2, 

35 conversations_pb2, 

36 editor_pb2, 

37 events_pb2, 

38 notification_data_pb2, 

39 notifications_pb2, 

40) 

41from couchers.proto.internal import unsubscribe_pb2 

42from couchers.servicers.api import user_model_to_pb 

43from couchers.utils import not_none, now 

44from tests.fixtures.db import generate_user 

45from tests.fixtures.misc import PushCollector, email_fields, mock_notification_email, process_jobs 

46from tests.fixtures.sessions import ( 

47 api_session, 

48 auth_api_session, 

49 conversations_session, 

50 notifications_session, 

51 real_editor_session, 

52) 

53 

54 

55@pytest.fixture(autouse=True) 

56def _(testconfig): 

57 pass 

58 

59 

60@pytest.mark.parametrize("enabled", [True, False]) 

61def test_SetNotificationSettings_preferences_respected_editable(db, enabled): 

62 user, token = generate_user() 

63 

64 # enable a notification type and check it gets delivered 

65 topic_action = NotificationTopicAction.badge__add 

66 

67 with notifications_session(token) as notifications: 

68 notifications.SetNotificationSettings( 

69 notifications_pb2.SetNotificationSettingsReq( 

70 preferences=[ 

71 notifications_pb2.SingleNotificationPreference( 

72 topic=topic_action.topic, 

73 action=topic_action.action, 

74 delivery_method="push", 

75 enabled=enabled, 

76 ) 

77 ], 

78 ) 

79 ) 

80 

81 with session_scope() as session: 

82 notify( 

83 session, 

84 user_id=user.id, 

85 topic_action=topic_action, 

86 key="", 

87 data=notification_data_pb2.BadgeAdd( 

88 badge_id="volunteer", 

89 badge_name="Active Volunteer", 

90 badge_description="This user is an active volunteer for Couchers.org", 

91 ), 

92 ) 

93 

94 process_job() 

95 

96 with session_scope() as session: 

97 deliv = session.execute( 

98 select(NotificationDelivery) 

99 .join(Notification, Notification.id == NotificationDelivery.notification_id) 

100 .where(Notification.user_id == user.id) 

101 .where(Notification.topic_action == topic_action) 

102 .where(NotificationDelivery.delivery_type == NotificationDeliveryType.push) 

103 ).scalar_one_or_none() 

104 

105 if enabled: 

106 assert deliv is not None 

107 else: 

108 assert deliv is None 

109 

110 

111def test_SetNotificationSettings_preferences_not_editable(db): 

112 user, token = generate_user() 

113 

114 # enable a notification type and check it gets delivered 

115 topic_action = NotificationTopicAction.password_reset__start 

116 

117 with notifications_session(token) as notifications: 

118 with pytest.raises(grpc.RpcError) as e: 

119 notifications.SetNotificationSettings( 

120 notifications_pb2.SetNotificationSettingsReq( 

121 preferences=[ 

122 notifications_pb2.SingleNotificationPreference( 

123 topic=topic_action.topic, 

124 action=topic_action.action, 

125 delivery_method="push", 

126 enabled=False, 

127 ) 

128 ], 

129 ) 

130 ) 

131 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

132 assert e.value.details() == "That notification preference is not user editable." 

133 

134 

135def test_unsubscribe(db): 

136 # this is the ugliest test i've written 

137 

138 user, token = generate_user() 

139 

140 topic_action = NotificationTopicAction.badge__add 

141 

142 # first enable email notifs 

143 with notifications_session(token) as notifications: 

144 notifications.SetNotificationSettings( 

145 notifications_pb2.SetNotificationSettingsReq( 

146 preferences=[ 

147 notifications_pb2.SingleNotificationPreference( 

148 topic=topic_action.topic, 

149 action=topic_action.action, 

150 delivery_method=method, 

151 enabled=enabled, 

152 ) 

153 for method, enabled in [("email", True), ("digest", False), ("push", False)] 

154 ], 

155 ) 

156 ) 

157 

158 with mock_notification_email() as mock: 

159 with session_scope() as session: 

160 notify( 

161 session, 

162 user_id=user.id, 

163 topic_action=topic_action, 

164 key="", 

165 data=notification_data_pb2.BadgeAdd( 

166 badge_id="volunteer", 

167 badge_name="Active Volunteer", 

168 badge_description="This user is an active volunteer for Couchers.org", 

169 ), 

170 ) 

171 

172 assert mock.call_count == 1 

173 assert email_fields(mock).recipient == user.email 

174 # very ugly 

175 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg= 

176 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html): 176 ↛ 197line 176 didn't jump to line 197 because the loop on line 176 didn't complete

177 if "payload" not in link: 

178 continue 

179 print(link) 

180 url_parts = urlparse(html.unescape(link)) 

181 params = parse_qs(url_parts.query) 

182 print(params["payload"][0]) 

183 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0])) 

184 if payload.HasField("topic_action"): 184 ↛ 176line 184 didn't jump to line 176 because the condition on line 184 was always true

185 with auth_api_session() as (auth_api, metadata_interceptor): 

186 assert ( 

187 auth_api.Unsubscribe( 

188 auth_pb2.UnsubscribeReq( 

189 payload=b64decode(params["payload"][0]), 

190 sig=b64decode(params["sig"][0]), 

191 ) 

192 ).response 

193 == "You've been unsubscribed from email notifications of that type." 

194 ) 

195 break 

196 else: 

197 raise Exception("Didn't find link") 

198 

199 with notifications_session(token) as notifications: 

200 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq()) 

201 

202 for group in res.groups: 

203 for topic in group.topics: 

204 for item in topic.items: 

205 if topic == topic_action.topic and item == topic_action.action: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true

206 assert not item.email 

207 

208 with mock_notification_email() as mock: 

209 with session_scope() as session: 

210 notify( 

211 session, 

212 user_id=user.id, 

213 topic_action=topic_action, 

214 key="", 

215 data=notification_data_pb2.BadgeAdd( 

216 badge_id="volunteer", 

217 badge_name="Active Volunteer", 

218 badge_description="This user is an active volunteer for Couchers.org", 

219 ), 

220 ) 

221 

222 assert mock.call_count == 0 

223 

224 

225def test_unsubscribe_do_not_email(db): 

226 user, token = generate_user() 

227 

228 _, token2 = generate_user(complete_profile=True) 

229 with mock_notification_email() as mock: 

230 with api_session(token2) as api: 

231 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id)) 

232 

233 assert mock.call_count == 1 

234 assert email_fields(mock).recipient == user.email 

235 # very ugly 

236 # http://localhost:3000/quick-link?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg= 

237 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html): 237 ↛ 258line 237 didn't jump to line 258 because the loop on line 237 didn't complete

238 if "payload" not in link: 

239 continue 

240 print(link) 

241 url_parts = urlparse(html.unescape(link)) 

242 params = parse_qs(url_parts.query) 

243 print(params["payload"][0]) 

244 payload = unsubscribe_pb2.UnsubscribePayload.FromString(b64decode(params["payload"][0])) 

245 if payload.HasField("do_not_email"): 

246 with auth_api_session() as (auth_api, metadata_interceptor): 

247 assert ( 

248 auth_api.Unsubscribe( 

249 auth_pb2.UnsubscribeReq( 

250 payload=b64decode(params["payload"][0]), 

251 sig=b64decode(params["sig"][0]), 

252 ) 

253 ).response 

254 == "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." 

255 ) 

256 break 

257 else: 

258 raise Exception("Didn't find link") 

259 

260 _, token3 = generate_user(complete_profile=True) 

261 with mock_notification_email() as mock: 

262 with api_session(token3) as api: 

263 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id)) 

264 

265 assert mock.call_count == 0 

266 

267 with session_scope() as session: 

268 user_ = session.execute(select(User).where(User.id == user.id)).scalar_one() 

269 assert user_.do_not_email 

270 

271 

272def test_get_do_not_email(db): 

273 _, token = generate_user() 

274 

275 with session_scope() as session: 

276 user = session.execute(select(User)).scalar_one() 

277 user.do_not_email = False 

278 

279 with notifications_session(token) as notifications: 

280 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq()) 

281 assert not res.do_not_email_enabled 

282 

283 with session_scope() as session: 

284 user = session.execute(select(User)).scalar_one() 

285 user.do_not_email = True 

286 user.hosting_status = HostingStatus.cant_host 

287 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

288 

289 with notifications_session(token) as notifications: 

290 res = notifications.GetNotificationSettings(notifications_pb2.GetNotificationSettingsReq()) 

291 assert res.do_not_email_enabled 

292 

293 

294def test_set_do_not_email(db): 

295 _, token = generate_user() 

296 

297 with session_scope() as session: 

298 user = session.execute(select(User)).scalar_one() 

299 user.do_not_email = False 

300 user.hosting_status = HostingStatus.can_host 

301 user.meetup_status = MeetupStatus.wants_to_meetup 

302 

303 with notifications_session(token) as notifications: 

304 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False)) 

305 

306 with session_scope() as session: 

307 user = session.execute(select(User)).scalar_one() 

308 assert not user.do_not_email 

309 

310 with notifications_session(token) as notifications: 

311 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True)) 

312 

313 with session_scope() as session: 

314 user = session.execute(select(User)).scalar_one() 

315 assert user.do_not_email 

316 assert user.hosting_status == HostingStatus.cant_host 

317 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup 

318 

319 with notifications_session(token) as notifications: 

320 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=False)) 

321 

322 with session_scope() as session: 

323 user = session.execute(select(User)).scalar_one() 

324 assert not user.do_not_email 

325 

326 

327def test_list_notifications(db, push_collector: PushCollector, moderator): 

328 user1, token1 = generate_user() 

329 user2, token2 = generate_user() 

330 

331 with api_session(token2) as api: 

332 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id)) 

333 

334 with notifications_session(token1) as notifications: 

335 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq()) 

336 assert len(res.notifications) == 1 

337 

338 n = res.notifications[0] 

339 

340 assert n.topic == "friend_request" 

341 assert n.action == "create" 

342 assert n.key == str(user2.id) 

343 assert n.title == f"Friend request from {user2.name}" 

344 assert n.body == f"{user2.name} wants to be your friend." 

345 assert n.icon.startswith("http://localhost:5001/img/thumbnail/") 

346 assert n.url == "http://localhost:3000/connections/friends/" 

347 

348 with conversations_session(token2) as c: 

349 res = c.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[user1.id])) 

350 group_chat_id = res.group_chat_id 

351 moderator.approve_group_chat(group_chat_id) 

352 for i in range(17): 

353 c.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=group_chat_id, text=f"Test message {i}")) 

354 

355 process_jobs() 

356 

357 all_notifs = [] 

358 with notifications_session(token1) as notifications: 

359 page_token = None 

360 for _ in range(100): 360 ↛ 373line 360 didn't jump to line 373

361 res = notifications.ListNotifications( 

362 notifications_pb2.ListNotificationsReq( 

363 page_size=5, 

364 page_token=page_token, 

365 ) 

366 ) 

367 assert len(res.notifications) == 5 or not res.next_page_token 

368 all_notifs += res.notifications 

369 page_token = res.next_page_token 

370 if not page_token: 

371 break 

372 

373 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"{user2.name} wants to be your friend."] 

374 assert bodys == [n.body for n in all_notifs] 

375 

376 

377def test_notifications_seen(db, push_collector: PushCollector): 

378 user1, token1 = generate_user() 

379 user2, token2 = generate_user() 

380 user3, token3 = generate_user() 

381 user4, token4 = generate_user() 

382 

383 with api_session(token2) as api: 

384 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id)) 

385 

386 with api_session(token3) as api: 

387 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id)) 

388 

389 with notifications_session(token1) as notifications, api_session(token1) as api: 

390 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq()) 

391 assert len(res.notifications) == 2 

392 assert [n.is_seen for n in res.notifications] == [False, False] 

393 notification_ids = [n.notification_id for n in res.notifications] 

394 # should be listed desc time 

395 assert notification_ids[0] > notification_ids[1] 

396 

397 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2 

398 

399 with api_session(token4) as api: 

400 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user1.id)) 

401 

402 with notifications_session(token1) as notifications, api_session(token1) as api: 

403 # mark everything before just the last one as seen (pretend we didn't load the last one yet in the api) 

404 notifications.MarkAllNotificationsSeen( 

405 notifications_pb2.MarkAllNotificationsSeenReq(latest_notification_id=notification_ids[0]) 

406 ) 

407 

408 # last one is still unseen 

409 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1 

410 

411 # mark the first one unseen 

412 notifications.MarkNotificationSeen( 

413 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids[1], set_seen=False) 

414 ) 

415 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2 

416 

417 # mark the last one seen 

418 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq()) 

419 assert len(res.notifications) == 3 

420 assert [n.is_seen for n in res.notifications] == [False, True, False] 

421 notification_ids2 = [n.notification_id for n in res.notifications] 

422 

423 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 2 

424 

425 notifications.MarkNotificationSeen( 

426 notifications_pb2.MarkNotificationSeenReq(notification_id=notification_ids2[0], set_seen=True) 

427 ) 

428 

429 res = notifications.ListNotifications(notifications_pb2.ListNotificationsReq()) 

430 assert len(res.notifications) == 3 

431 assert [n.is_seen for n in res.notifications] == [True, True, False] 

432 

433 assert api.Ping(api_pb2.PingReq()).unseen_notification_count == 1 

434 

435 

436def test_GetVapidPublicKey(db): 

437 _, token = generate_user() 

438 

439 with notifications_session(token) as notifications: 

440 assert ( 

441 notifications.GetVapidPublicKey(empty_pb2.Empty()).vapid_public_key 

442 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A" 

443 ) 

444 

445 

446def test_RegisterPushNotificationSubscription(db): 

447 _, token = generate_user() 

448 

449 subscription_info = { 

450 "endpoint": "https://updates.push.services.mozilla.com/wpush/v2/gAAAAABmW2_iYKVyZRJPhAhktbkXd6Bc8zjIUvtVi5diYL7ZYn8FHka94kIdF46Mp8DwCDWlACnbKOEo97ikaa7JYowGLiGz3qsWL7Vo19LaV4I71mUDUOIKxWIsfp_kM77MlRJQKDUddv-sYyiffOyg63d1lnc_BMIyLXt69T5SEpfnfWTNb6I", 

451 "expirationTime": None, 

452 "keys": { 

453 "auth": "TnuEJ1OdfEkf6HKcUovl0Q", 

454 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0", 

455 }, 

456 } 

457 

458 with notifications_session(token) as notifications: 

459 res = notifications.RegisterPushNotificationSubscription( 

460 notifications_pb2.RegisterPushNotificationSubscriptionReq( 

461 full_subscription_json=json.dumps(subscription_info), 

462 ) 

463 ) 

464 

465 

466def test_SendTestPushNotification(db, push_collector: PushCollector): 

467 user, token = generate_user() 

468 

469 with notifications_session(token) as notifications: 

470 notifications.SendTestPushNotification(empty_pb2.Empty()) 

471 

472 assert push_collector.count_for_user(user.id) == 1 

473 push = push_collector.pop_for_user(user.id, last=True) 

474 assert push.content.title == "Push notifications test" 

475 assert push.content.body == "If you see this, then it's working :)" 

476 

477 

478def test_SendBlogPostNotification(db, push_collector: PushCollector): 

479 super_user, super_token = generate_user(is_superuser=True) 

480 

481 user1, user1_token = generate_user() 

482 # enabled email 

483 user2, user2_token = generate_user() 

484 # disabled push 

485 user3, user3_token = generate_user() 

486 

487 topic_action = NotificationTopicAction.general__new_blog_post 

488 

489 with notifications_session(user2_token) as notifications: 

490 notifications.SetNotificationSettings( 

491 notifications_pb2.SetNotificationSettingsReq( 

492 preferences=[ 

493 notifications_pb2.SingleNotificationPreference( 

494 topic=topic_action.topic, 

495 action=topic_action.action, 

496 delivery_method="email", 

497 enabled=True, 

498 ) 

499 ], 

500 ) 

501 ) 

502 

503 with notifications_session(user3_token) as notifications: 

504 notifications.SetNotificationSettings( 

505 notifications_pb2.SetNotificationSettingsReq( 

506 preferences=[ 

507 notifications_pb2.SingleNotificationPreference( 

508 topic=topic_action.topic, 

509 action=topic_action.action, 

510 delivery_method="push", 

511 enabled=False, 

512 ) 

513 ], 

514 ) 

515 ) 

516 

517 with mock_notification_email() as mock: 

518 with real_editor_session(super_token) as editor_api: 

519 editor_api.SendBlogPostNotification( 

520 editor_pb2.SendBlogPostNotificationReq( 

521 title="Couchers.org v0.9.9 Release Notes", 

522 blurb="Read about last major updates before v1!", 

523 url="https://couchers.org/blog/2025/05/11/v0.9.9-release", 

524 ) 

525 ) 

526 

527 process_jobs() 

528 

529 assert mock.call_count == 1 

530 assert email_fields(mock).recipient == user2.email 

531 assert "Couchers.org v0.9.9 Release Notes" in email_fields(mock).html 

532 assert "Couchers.org v0.9.9 Release Notes" in email_fields(mock).plain 

533 assert "Read about last major updates before v1!" in email_fields(mock).html 

534 assert "Read about last major updates before v1!" in email_fields(mock).plain 

535 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email_fields(mock).html 

536 assert "https://couchers.org/blog/2025/05/11/v0.9.9-release" in email_fields(mock).plain 

537 

538 push = push_collector.pop_for_user(user1.id, last=True) 

539 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes" 

540 assert push.content.body == "Read about last major updates before v1!" 

541 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release" 

542 

543 push = push_collector.pop_for_user(user2.id, last=True) 

544 assert push.content.title == "New blog post: Couchers.org v0.9.9 Release Notes" 

545 assert push.content.body == "Read about last major updates before v1!" 

546 assert push.content.action_url == "https://couchers.org/blog/2025/05/11/v0.9.9-release" 

547 

548 assert push_collector.count_for_user(user3.id) == 0 

549 

550 

551def test_get_topic_actions_by_delivery_type(db): 

552 user, token = generate_user() 

553 

554 # these are enabled by default 

555 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults 

556 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults 

557 

558 # these are disabled by default 

559 assert NotificationDeliveryType.push not in NotificationTopicAction.event__create_any.defaults 

560 assert NotificationDeliveryType.push not in NotificationTopicAction.discussion__create.defaults 

561 

562 with notifications_session(token) as notifications: 

563 notifications.SetNotificationSettings( 

564 notifications_pb2.SetNotificationSettingsReq( 

565 preferences=[ 

566 notifications_pb2.SingleNotificationPreference( 

567 topic=NotificationTopicAction.reference__receive_friend.topic, 

568 action=NotificationTopicAction.reference__receive_friend.action, 

569 delivery_method="push", 

570 enabled=False, 

571 ), 

572 notifications_pb2.SingleNotificationPreference( 

573 topic=NotificationTopicAction.event__create_any.topic, 

574 action=NotificationTopicAction.event__create_any.action, 

575 delivery_method="push", 

576 enabled=True, 

577 ), 

578 ], 

579 ) 

580 ) 

581 

582 with session_scope() as session: 

583 deliver = get_topic_actions_by_delivery_type(session, user.id, NotificationDeliveryType.push) 

584 assert NotificationTopicAction.reference__receive_friend not in deliver 

585 assert NotificationTopicAction.host_request__accept in deliver 

586 assert NotificationTopicAction.event__create_any in deliver 

587 assert NotificationTopicAction.discussion__create not in deliver 

588 assert NotificationTopicAction.account_deletion__start in deliver 

589 

590 

591def test_event_reminder_email_sent(db): 

592 user, token = generate_user() 

593 title = "Board Game Night" 

594 start_event_time = timestamp_pb2.Timestamp(seconds=1751690400) 

595 

596 expected_time_str = LocalizationContext.from_user(user).localize_datetime(start_event_time) 

597 

598 with mock_notification_email() as mock: 

599 with session_scope() as session: 

600 user_in_session = session.get_one(User, user.id) 

601 

602 notify( 

603 session, 

604 user_id=user.id, 

605 topic_action=NotificationTopicAction.event__reminder, 

606 key="", 

607 data=notification_data_pb2.EventReminder( 

608 event=events_pb2.Event( 

609 event_id=1, 

610 slug="board-game-night", 

611 title=title, 

612 start_time=start_event_time, 

613 ), 

614 user=user_model_to_pb(user_in_session, session, make_background_user_context(user_id=user.id)), 

615 ), 

616 ) 

617 

618 assert mock.call_count == 1 

619 assert email_fields(mock).recipient == user.email 

620 assert title in email_fields(mock).html 

621 assert title in email_fields(mock).plain 

622 assert expected_time_str in email_fields(mock).html 

623 assert expected_time_str in email_fields(mock).plain 

624 

625 

626def test_RegisterMobilePushNotificationSubscription(db): 

627 user, token = generate_user() 

628 

629 with notifications_session(token) as notifications: 

630 notifications.RegisterMobilePushNotificationSubscription( 

631 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq( 

632 token="ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]", 

633 device_name="My iPhone", 

634 device_type="ios", 

635 ) 

636 ) 

637 

638 # Check subscription was created 

639 with session_scope() as session: 

640 sub = session.execute( 

641 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id) 

642 ).scalar_one() 

643 assert sub.platform == PushNotificationPlatform.expo 

644 assert sub.token == "ExponentPushToken[xxxxxxxxxxxxxxxxxxxxxx]" 

645 assert sub.device_name == "My iPhone" 

646 assert sub.device_type == DeviceType.ios 

647 assert sub.disabled_at == DATETIME_INFINITY 

648 

649 

650def test_RegisterMobilePushNotificationSubscription_android(db): 

651 user, token = generate_user() 

652 

653 with notifications_session(token) as notifications: 

654 notifications.RegisterMobilePushNotificationSubscription( 

655 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq( 

656 token="ExponentPushToken[yyyyyyyyyyyyyyyyyyyyyy]", 

657 device_name="My Android", 

658 device_type="android", 

659 ) 

660 ) 

661 

662 with session_scope() as session: 

663 sub = session.execute( 

664 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id) 

665 ).scalar_one() 

666 assert sub.platform == PushNotificationPlatform.expo 

667 assert sub.device_type == DeviceType.android 

668 

669 

670def test_RegisterMobilePushNotificationSubscription_no_device_type(db): 

671 user, token = generate_user() 

672 

673 with notifications_session(token) as notifications: 

674 notifications.RegisterMobilePushNotificationSubscription( 

675 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq( 

676 token="ExponentPushToken[zzzzzzzzzzzzzzzzzzzzzz]", 

677 ) 

678 ) 

679 

680 with session_scope() as session: 

681 sub = session.execute( 

682 select(PushNotificationSubscription).where(PushNotificationSubscription.user_id == user.id) 

683 ).scalar_one() 

684 assert sub.platform == PushNotificationPlatform.expo 

685 assert sub.device_name is None 

686 assert sub.device_type is None 

687 

688 

689def test_RegisterMobilePushNotificationSubscription_re_enable(db): 

690 user, token = generate_user() 

691 

692 # Create a disabled subscription directly in the DB 

693 with session_scope() as session: 

694 sub = PushNotificationSubscription( 

695 user_id=user.id, 

696 platform=PushNotificationPlatform.expo, 

697 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]", 

698 device_name="Old Device", 

699 device_type=DeviceType.ios, 

700 ) 

701 sub.disabled_at = now() 

702 session.add(sub) 

703 session.flush() 

704 sub_id = sub.id 

705 

706 # Re-register with the same token 

707 with notifications_session(token) as notifications: 

708 notifications.RegisterMobilePushNotificationSubscription( 

709 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq( 

710 token="ExponentPushToken[reeeeeeeeeeeeeeeeeeeee]", 

711 device_name="New Device Name", 

712 device_type="android", 

713 ) 

714 ) 

715 

716 # Check subscription was re-enabled and updated 

717 with session_scope() as session: 

718 sub = session.execute( 

719 select(PushNotificationSubscription).where(PushNotificationSubscription.id == sub_id) 

720 ).scalar_one() 

721 assert sub.disabled_at == DATETIME_INFINITY 

722 assert sub.device_name == "New Device Name" 

723 assert sub.device_type == DeviceType.android 

724 

725 

726def test_RegisterMobilePushNotificationSubscription_already_exists(db): 

727 user, token = generate_user() 

728 

729 # Create an active subscription directly in the DB 

730 with session_scope() as session: 

731 sub = PushNotificationSubscription( 

732 user_id=user.id, 

733 platform=PushNotificationPlatform.expo, 

734 token="ExponentPushToken[existingtoken]", 

735 device_name="Existing Device", 

736 device_type=DeviceType.ios, 

737 ) 

738 session.add(sub) 

739 

740 # Try to register with the same token - should just return without error 

741 with notifications_session(token) as notifications: 

742 notifications.RegisterMobilePushNotificationSubscription( 

743 notifications_pb2.RegisterMobilePushNotificationSubscriptionReq( 

744 token="ExponentPushToken[existingtoken]", 

745 device_name="Different Name", 

746 ) 

747 ) 

748 

749 # Check subscription was NOT modified (already active) 

750 with session_scope() as session: 

751 sub = session.execute( 

752 select(PushNotificationSubscription).where( 

753 PushNotificationSubscription.token == "ExponentPushToken[existingtoken]" 

754 ) 

755 ).scalar_one() 

756 assert sub.device_name == "Existing Device" # unchanged 

757 

758 

759def test_SendTestMobilePushNotification(db, push_collector: PushCollector): 

760 user, token = generate_user() 

761 

762 with notifications_session(token) as notifications: 

763 notifications.SendTestMobilePushNotification(empty_pb2.Empty()) 

764 

765 push = push_collector.pop_for_user(user.id, last=True) 

766 assert push.content.title == "Mobile notifications test" 

767 assert push.content.body == "If you see this on your phone, everything is wired up correctly 🎉" 

768 

769 

770def test_get_expo_push_receipts(db): 

771 from unittest.mock import Mock, patch 

772 

773 from couchers.notifications.expo_api import get_expo_push_receipts 

774 

775 mock_response = Mock() 

776 mock_response.status_code = 200 

777 mock_response.json.return_value = { 

778 "data": { 

779 "ticket-1": {"status": "ok"}, 

780 "ticket-2": {"status": "error", "details": {"error": "DeviceNotRegistered"}}, 

781 } 

782 } 

783 

784 with patch("couchers.notifications.expo_api.requests.post", return_value=mock_response) as mock_post: 

785 result = get_expo_push_receipts(["ticket-1", "ticket-2"]) 

786 

787 mock_post.assert_called_once() 

788 call_args = mock_post.call_args 

789 assert call_args[0][0] == "https://exp.host/--/api/v2/push/getReceipts" 

790 assert call_args[1]["json"] == {"ids": ["ticket-1", "ticket-2"]}