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

1import json

2import re

3from urllib.parse import parse_qs, urlparse

5import grpc

6import pytest

7from google.protobuf import empty_pb2, timestamp_pb2

9from couchers import errors

10from couchers.context import make_background_user_context

11from couchers.crypto import b64decode

12from couchers.jobs.worker import process_job

13from couchers.models import (

14 HostingStatus,

15 MeetupStatus,

16 Notification,

17 NotificationDelivery,

18 NotificationDeliveryType,

19 NotificationTopicAction,

20 User,

21)

22from couchers.notifications.notify import notify

23from couchers.notifications.settings import get_topic_actions_by_delivery_type

24from couchers.servicers.api import user_model_to_pb

25from couchers.sql import couchers_select as select

26from couchers.templates.v2 import v2timestamp

27from proto import admin_pb2, api_pb2, auth_pb2, conversations_pb2, events_pb2, notification_data_pb2, notifications_pb2

28from proto.internal import unsubscribe_pb2

29from tests.test_fixtures import ( # noqa

30 api_session,

31 auth_api_session,

32 conversations_session,

33 db,

34 email_fields,

35 generate_user,

36 mock_notification_email,

37 notifications_session,

38 process_jobs,

39 push_collector,

40 real_admin_session,

41 session_scope,

42 testconfig,

43)

46@pytest.fixture(autouse=True)

47def _(testconfig):

48 pass

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

52def test_SetNotificationSettings_preferences_respected_editable(db, enabled):

53 user, token = generate_user()

55 # enable a notification type and check it gets delivered

56 topic_action = NotificationTopicAction.badge__add

58 with notifications_session(token) as notifications:

59 notifications.SetNotificationSettings(

60 notifications_pb2.SetNotificationSettingsReq(

61 preferences=[

62 notifications_pb2.SingleNotificationPreference(

63 topic=topic_action.topic,

64 action=topic_action.action,

65 delivery_method="push",

66 enabled=enabled,

67 )

68 ],

69 )

70 )

72 with session_scope() as session:

73 notify(

74 session,

75 user_id=user.id,

76 topic_action=topic_action.display,

77 data=notification_data_pb2.BadgeAdd(

78 badge_id="volunteer",

79 badge_name="Active Volunteer",

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

81 ),

82 )

84 process_job()

86 with session_scope() as session:

87 deliv = session.execute(

88 select(NotificationDelivery)

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

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

91 .where(Notification.topic_action == topic_action)

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

93 ).scalar_one_or_none()

95 if enabled:

96 assert deliv is not None

97 else:

98 assert deliv is None

100

101def test_SetNotificationSettings_preferences_not_editable(db):

102 user, token = generate_user()

103

104 # enable a notification type and check it gets delivered

105 topic_action = NotificationTopicAction.password_reset__start

106

107 with notifications_session(token) as notifications:

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

109 notifications.SetNotificationSettings(

110 notifications_pb2.SetNotificationSettingsReq(

111 preferences=[

112 notifications_pb2.SingleNotificationPreference(

113 topic=topic_action.topic,

114 action=topic_action.action,

115 delivery_method="push",

116 enabled=False,

117 )

118 ],

119 )

120 )

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

122 assert e.value.details() == errors.CANNOT_EDIT_THAT_NOTIFICATION_PREFERENCE

123

124

125def test_unsubscribe(db):

126 # this is the ugliest test i've written

127

128 user, token = generate_user()

129

130 topic_action = NotificationTopicAction.badge__add

131

132 # first enable email notifs

133 with notifications_session(token) as notifications:

134 notifications.SetNotificationSettings(

135 notifications_pb2.SetNotificationSettingsReq(

136 preferences=[

137 notifications_pb2.SingleNotificationPreference(

138 topic=topic_action.topic,

139 action=topic_action.action,

140 delivery_method=method,

141 enabled=enabled,

142 )

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

144 ],

145 )

146 )

147

148 with mock_notification_email() as mock:

149 with session_scope() as session:

150 notify(

151 session,

152 user_id=user.id,

153 topic_action=topic_action.display,

154 data=notification_data_pb2.BadgeAdd(

155 badge_id="volunteer",

156 badge_name="Active Volunteer",

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

158 ),

159 )

160

161 assert mock.call_count == 1

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

163 # very ugly

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

165 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html):

166 if "payload" not in link:

167 continue

168 print(link)

169 url_parts = urlparse(link)

170 params = parse_qs(url_parts.query)

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

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

173 if payload.HasField("topic_action"):

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

175 res = auth_api.Unsubscribe(

176 auth_pb2.UnsubscribeReq(

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

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

179 )

180 )

181 break

182 else:

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

184

185 with notifications_session(token) as notifications:

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

187

188 for group in res.groups:

189 for topic in group.topics:

190 for item in topic.items:

191 if topic == topic_action.topic and item == topic_action.action:

192 assert not item.email

193

194 with mock_notification_email() as mock:

195 with session_scope() as session:

196 notify(

197 session,

198 user_id=user.id,

199 topic_action=topic_action.display,

200 data=notification_data_pb2.BadgeAdd(

201 badge_id="volunteer",

202 badge_name="Active Volunteer",

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

204 ),

205 )

206

207 assert mock.call_count == 0

208

209

210def test_unsubscribe_do_not_email(db):

211 user, token = generate_user()

212

213 _, token2 = generate_user(complete_profile=True)

214 with mock_notification_email() as mock:

215 with api_session(token2) as api:

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

217

218 assert mock.call_count == 1

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

220 # very ugly

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

222 for link in re.findall(r'<a href="(.*?)"', email_fields(mock).html):

223 if "payload" not in link:

224 continue

225 print(link)

226 url_parts = urlparse(link)

227 params = parse_qs(url_parts.query)

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

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

230 if payload.HasField("do_not_email"):

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

232 res = auth_api.Unsubscribe(

233 auth_pb2.UnsubscribeReq(

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

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

236 )

237 )

238 break

239 else:

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

241

242 _, token3 = generate_user(complete_profile=True)

243 with mock_notification_email() as mock:

244 with api_session(token3) as api:

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

246

247 assert mock.call_count == 0

248

249 with session_scope() as session:

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

251 assert user_.do_not_email

252

253

254def test_get_do_not_email(db):

255 _, token = generate_user()

256

257 with session_scope() as session:

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

259 user.do_not_email = False

260

261 with notifications_session(token) as notifications:

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

263 assert not res.do_not_email_enabled

264

265 with session_scope() as session:

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

267 user.do_not_email = True

268 user.hosting_status = HostingStatus.cant_host

269 user.meetup_status = MeetupStatus.does_not_want_to_meetup

270

271 with notifications_session(token) as notifications:

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

273 assert res.do_not_email_enabled

274

275

276def test_set_do_not_email(db):

277 _, token = generate_user()

278

279 with session_scope() as session:

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

281 user.do_not_email = False

282 user.hosting_status = HostingStatus.can_host

283 user.meetup_status = MeetupStatus.wants_to_meetup

284

285 with notifications_session(token) as notifications:

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

287

288 with session_scope() as session:

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

290 assert not user.do_not_email

291

292 with notifications_session(token) as notifications:

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

294

295 with session_scope() as session:

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

297 assert user.do_not_email

298 assert user.hosting_status == HostingStatus.cant_host

299 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup

300

301 with notifications_session(token) as notifications:

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

303

304 with session_scope() as session:

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

306 assert not user.do_not_email

307

308

309def test_list_notifications(db, push_collector):

310 user1, token1 = generate_user()

311 user2, token2 = generate_user()

312

313 with api_session(token2) as api:

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

315

316 with notifications_session(token1) as notifications:

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

318 assert len(res.notifications) == 1

319

320 n = res.notifications[0]

321

322 assert n.topic == "friend_request"

323 assert n.action == "create"

324 assert n.key == "2"

325 assert n.title == f"{user2.name} wants to be your friend"

326 assert n.body == f"You've received a friend request from {user2.name}"

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

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

329

330 with conversations_session(token2) as c:

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

332 group_chat_id = res.group_chat_id

333 for i in range(17):

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

335

336 process_jobs()

337

338 all_notifs = []

339 with notifications_session(token1) as notifications:

340 page_token = None

341 for _ in range(100):

342 res = notifications.ListNotifications(

343 notifications_pb2.ListNotificationsReq(

344 page_size=5,

345 page_token=page_token,

346 )

347 )

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

349 all_notifs += res.notifications

350 page_token = res.next_page_token

351 if not page_token:

352 break

353

354 bodys = [f"Test message {16 - i}" for i in range(17)] + [f"You've received a friend request from {user2.name}"]

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

356

357

358def test_notifications_seen(db, push_collector):

359 user1, token1 = generate_user()

360 user2, token2 = generate_user()

361 user3, token3 = generate_user()

362 user4, token4 = generate_user()

363

364 with api_session(token2) as api:

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

366

367 with api_session(token3) as api:

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

369

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

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

372 assert len(res.notifications) == 2

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

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

375 # should be listed desc time

376 assert notification_ids[0] > notification_ids[1]

377

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

379

380 with api_session(token4) as api:

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

382

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

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

385 notifications.MarkAllNotificationsSeen(

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

387 )

388

389 # last one is still unseen

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

391

392 # mark the first one unseen

393 notifications.MarkNotificationSeen(

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

395 )

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

397

398 # mark the last one seen

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

400 assert len(res.notifications) == 3

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

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

403

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

405

406 notifications.MarkNotificationSeen(

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

408 )

409

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

411 assert len(res.notifications) == 3

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

413

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

415

416

417def test_GetVapidPublicKey(db):

418 _, token = generate_user()

419

420 with notifications_session(token) as notifications:

421 assert (

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

423 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"

424 )

425

426

427def test_RegisterPushNotificationSubscription(db):

428 _, token = generate_user()

429

430 subscription_info = {

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

432 "expirationTime": None,

433 "keys": {

434 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",

435 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",

436 },

437 }

438

439 with notifications_session(token) as notifications:

440 res = notifications.RegisterPushNotificationSubscription(

441 notifications_pb2.RegisterPushNotificationSubscriptionReq(

442 full_subscription_json=json.dumps(subscription_info),

443 )

444 )

445

446

447def test_SendTestPushNotification(db, push_collector):

448 user, token = generate_user()

449

450 with notifications_session(token) as notifications:

451 notifications.SendTestPushNotification(empty_pb2.Empty())

452

453 push_collector.assert_user_has_count(user.id, 1)

454 push_collector.assert_user_push_matches_fields(

455 user.id,

456 title="Checking push notifications work!",

457 body="If you see this, then it's working :)",

458 )

459

460 # the above two are equivalent to this

461

462 push_collector.assert_user_has_single_matching(

463 user.id,

464 title="Checking push notifications work!",

465 body="If you see this, then it's working :)",

466 )

467

468

469def test_SendBlogPostNotification(db, push_collector):

470 super_user, super_token = generate_user(is_superuser=True)

471

472 user1, user1_token = generate_user()

473 # enabled email

474 user2, user2_token = generate_user()

475 # disabled push

476 user3, user3_token = generate_user()

477

478 topic_action = NotificationTopicAction.general__new_blog_post

479

480 with notifications_session(user2_token) as notifications:

481 notifications.SetNotificationSettings(

482 notifications_pb2.SetNotificationSettingsReq(

483 preferences=[

484 notifications_pb2.SingleNotificationPreference(

485 topic=topic_action.topic,

486 action=topic_action.action,

487 delivery_method="email",

488 enabled=True,

489 )

490 ],

491 )

492 )

493

494 with notifications_session(user3_token) as notifications:

495 notifications.SetNotificationSettings(

496 notifications_pb2.SetNotificationSettingsReq(

497 preferences=[

498 notifications_pb2.SingleNotificationPreference(

499 topic=topic_action.topic,

500 action=topic_action.action,

501 delivery_method="push",

502 enabled=False,

503 )

504 ],

505 )

506 )

507

508 with mock_notification_email() as mock:

509 with real_admin_session(super_token) as admin_api:

510 admin_api.SendBlogPostNotification(

511 admin_pb2.SendBlogPostNotificationReq(

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

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

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

515 )

516 )

517

518 process_jobs()

519

520 assert mock.call_count == 1

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

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

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

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

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

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

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

528

529 push_collector.assert_user_has_count(user1.id, 1)

530 push_collector.assert_user_push_matches_fields(

531 user1.id,

532 title="New blog post: Couchers.org v0.9.9 Release Notes",

533 body="Read about last major updates before v1!",

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

535 )

536

537 push_collector.assert_user_has_count(user2.id, 1)

538 push_collector.assert_user_push_matches_fields(

539 user2.id,

540 title="New blog post: Couchers.org v0.9.9 Release Notes",

541 body="Read about last major updates before v1!",

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

543 )

544

545 push_collector.assert_user_has_count(user3.id, 0)

546

547

548def test_get_topic_actions_by_delivery_type(db):

549 user, token = generate_user()

550

551 # these are enabled by default

552 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults

553 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults

554

555 # these are disabled by default

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

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

558

559 with notifications_session(token) as notifications:

560 notifications.SetNotificationSettings(

561 notifications_pb2.SetNotificationSettingsReq(

562 preferences=[

563 notifications_pb2.SingleNotificationPreference(

564 topic=NotificationTopicAction.reference__receive_friend.topic,

565 action=NotificationTopicAction.reference__receive_friend.action,

566 delivery_method="push",

567 enabled=False,

568 ),

569 notifications_pb2.SingleNotificationPreference(

570 topic=NotificationTopicAction.event__create_any.topic,

571 action=NotificationTopicAction.event__create_any.action,

572 delivery_method="push",

573 enabled=True,

574 ),

575 ],

576 )

577 )

578

579 with session_scope() as session:

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

581 assert NotificationTopicAction.reference__receive_friend not in deliver

582 assert NotificationTopicAction.host_request__accept in deliver

583 assert NotificationTopicAction.event__create_any in deliver

584 assert NotificationTopicAction.discussion__create not in deliver

585 assert NotificationTopicAction.account_deletion__start in deliver

586

587

588def test_event_reminder_email_sent(db):

589 user, token = generate_user()

590 title = "Board Game Night"

591 start_event_time = timestamp_pb2.Timestamp(seconds=1751690400)

592 expected_time_str = v2timestamp(start_event_time, user)

593

594 with mock_notification_email() as mock:

595 with session_scope() as session:

596 user_in_session = session.get(User, user.id)

597

598 notify(

599 session,

600 user_id=user.id,

601 topic_action="event:reminder",

602 data=notification_data_pb2.EventReminder(

603 event=events_pb2.Event(

604 event_id=1,

605 slug="board-game-night",

606 title=title,

607 start_time=start_event_time,

608 ),

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

610 ),

611 )

612

613 assert mock.call_count == 1

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

615 assert title in email_fields(mock).html

616 assert title in email_fields(mock).plain

617 assert expected_time_str in email_fields(mock).html

618 assert expected_time_str in email_fields(mock).plain