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

9from couchers import errors

10from couchers.crypto import b64decode

11from couchers.jobs.worker import process_job

12from couchers.models import (

13 HostingStatus,

14 MeetupStatus,

15 Notification,

16 NotificationDelivery,

17 NotificationDeliveryType,

18 NotificationTopicAction,

19 User,

20)

21from couchers.notifications.notify import notify

22from couchers.notifications.settings import get_topic_actions_by_delivery_type

23from couchers.sql import couchers_select as select

24from proto import admin_pb2, api_pb2, auth_pb2, conversations_pb2, notification_data_pb2, notifications_pb2

25from proto.internal import unsubscribe_pb2

26from tests.test_fixtures import ( # noqa

27 api_session,

28 auth_api_session,

29 conversations_session,

30 db,

31 email_fields,

32 generate_user,

33 mock_notification_email,

34 notifications_session,

35 process_jobs,

36 push_collector,

37 real_admin_session,

38 session_scope,

39 testconfig,

40)

43@pytest.fixture(autouse=True)

44def _(testconfig):

45 pass

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

49def test_SetNotificationSettings_preferences_respected_editable(db, enabled):

50 user, token = generate_user()

52 # enable a notification type and check it gets delivered

53 topic_action = NotificationTopicAction.badge__add

55 with notifications_session(token) as notifications:

56 notifications.SetNotificationSettings(

57 notifications_pb2.SetNotificationSettingsReq(

58 preferences=[

59 notifications_pb2.SingleNotificationPreference(

60 topic=topic_action.topic,

61 action=topic_action.action,

62 delivery_method="push",

63 enabled=enabled,

64 )

65 ],

66 )

67 )

69 with session_scope() as session:

70 notify(

71 session,

72 user_id=user.id,

73 topic_action=topic_action.display,

74 data=notification_data_pb2.BadgeAdd(

75 badge_id="volunteer",

76 badge_name="Active Volunteer",

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

78 ),

79 )

81 process_job()

83 with session_scope() as session:

84 deliv = session.execute(

85 select(NotificationDelivery)

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

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

88 .where(Notification.topic_action == topic_action)

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

90 ).scalar_one_or_none()

92 if enabled:

93 assert deliv is not None

94 else:

95 assert deliv is None

98def test_SetNotificationSettings_preferences_not_editable(db):

99 user, token = generate_user()

100

101 # enable a notification type and check it gets delivered

102 topic_action = NotificationTopicAction.password_reset__start

103

104 with notifications_session(token) as notifications:

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

106 notifications.SetNotificationSettings(

107 notifications_pb2.SetNotificationSettingsReq(

108 preferences=[

109 notifications_pb2.SingleNotificationPreference(

110 topic=topic_action.topic,

111 action=topic_action.action,

112 delivery_method="push",

113 enabled=False,

114 )

115 ],

116 )

117 )

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

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

120

121

122def test_unsubscribe(db):

123 # this is the ugliest test i've written

124

125 user, token = generate_user()

126

127 topic_action = NotificationTopicAction.badge__add

128

129 # first enable email notifs

130 with notifications_session(token) as notifications:

131 notifications.SetNotificationSettings(

132 notifications_pb2.SetNotificationSettingsReq(

133 preferences=[

134 notifications_pb2.SingleNotificationPreference(

135 topic=topic_action.topic,

136 action=topic_action.action,

137 delivery_method=method,

138 enabled=enabled,

139 )

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

141 ],

142 )

143 )

144

145 with mock_notification_email() as mock:

146 with session_scope() as session:

147 notify(

148 session,

149 user_id=user.id,

150 topic_action=topic_action.display,

151 data=notification_data_pb2.BadgeAdd(

152 badge_id="volunteer",

153 badge_name="Active Volunteer",

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

155 ),

156 )

157

158 assert mock.call_count == 1

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

160 # very ugly

161 # http://localhost:3000/unsubscribe?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=

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

163 if "payload" not in link:

164 continue

165 print(link)

166 url_parts = urlparse(link)

167 params = parse_qs(url_parts.query)

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

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

170 if payload.HasField("topic_action"):

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

172 res = auth_api.Unsubscribe(

173 auth_pb2.UnsubscribeReq(

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

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

176 )

177 )

178 break

179 else:

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

181

182 with notifications_session(token) as notifications:

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

184

185 for group in res.groups:

186 for topic in group.topics:

187 for item in topic.items:

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

189 assert not item.email

190

191 with mock_notification_email() as mock:

192 with session_scope() as session:

193 notify(

194 session,

195 user_id=user.id,

196 topic_action=topic_action.display,

197 data=notification_data_pb2.BadgeAdd(

198 badge_id="volunteer",

199 badge_name="Active Volunteer",

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

201 ),

202 )

203

204 assert mock.call_count == 0

205

206

207def test_unsubscribe_do_not_email(db):

208 user, token = generate_user()

209

210 _, token2 = generate_user(complete_profile=True)

211 with mock_notification_email() as mock:

212 with api_session(token2) as api:

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

214

215 assert mock.call_count == 1

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

217 # very ugly

218 # http://localhost:3000/unsubscribe?payload=CAEiGAoOZnJpZW5kX3JlcXVlc3QSBmFjY2VwdA==&sig=BQdk024NTATm8zlR0krSXTBhP5U9TlFv7VhJeIHZtUg=

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

220 if "payload" not in link:

221 continue

222 print(link)

223 url_parts = urlparse(link)

224 params = parse_qs(url_parts.query)

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

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

227 if payload.HasField("do_not_email"):

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

229 res = auth_api.Unsubscribe(

230 auth_pb2.UnsubscribeReq(

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

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

233 )

234 )

235 break

236 else:

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

238

239 _, token3 = generate_user(complete_profile=True)

240 with mock_notification_email() as mock:

241 with api_session(token3) as api:

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

243

244 assert mock.call_count == 0

245

246 with session_scope() as session:

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

248 assert user_.do_not_email

249

250

251def test_get_do_not_email(db):

252 _, token = generate_user()

253

254 with session_scope() as session:

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

256 user.do_not_email = False

257

258 with notifications_session(token) as notifications:

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

260 assert not res.do_not_email_enabled

261

262 with session_scope() as session:

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

264 user.do_not_email = True

265 user.hosting_status = HostingStatus.cant_host

266 user.meetup_status = MeetupStatus.does_not_want_to_meetup

267

268 with notifications_session(token) as notifications:

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

270 assert res.do_not_email_enabled

271

272

273def test_set_do_not_email(db):

274 _, token = generate_user()

275

276 with session_scope() as session:

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

278 user.do_not_email = False

279 user.hosting_status = HostingStatus.can_host

280 user.meetup_status = MeetupStatus.wants_to_meetup

281

282 with notifications_session(token) as notifications:

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

284

285 with session_scope() as session:

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

287 assert not user.do_not_email

288

289 with notifications_session(token) as notifications:

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

291

292 with session_scope() as session:

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

294 assert user.do_not_email

295 assert user.hosting_status == HostingStatus.cant_host

296 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup

297

298 with notifications_session(token) as notifications:

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

300

301 with session_scope() as session:

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

303 assert not user.do_not_email

304

305

306def test_list_notifications(db, push_collector):

307 user1, token1 = generate_user()

308 user2, token2 = generate_user()

309

310 with api_session(token2) as api:

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

312

313 with notifications_session(token1) as notifications:

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

315 assert len(res.notifications) == 1

316

317 n = res.notifications[0]

318

319 assert n.topic == "friend_request"

320 assert n.action == "create"

321 assert n.key == "2"

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

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

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

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

326

327 with conversations_session(token2) as c:

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

329 group_chat_id = res.group_chat_id

330 for i in range(17):

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

332

333 process_jobs()

334

335 all_notifs = []

336 with notifications_session(token1) as notifications:

337 page_token = None

338 for _ in range(100):

339 res = notifications.ListNotifications(

340 notifications_pb2.ListNotificationsReq(

341 page_size=5,

342 page_token=page_token,

343 )

344 )

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

346 all_notifs += res.notifications

347 page_token = res.next_page_token

348 if not page_token:

349 break

350

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

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

353

354

355def test_notifications_seen(db, push_collector):

356 user1, token1 = generate_user()

357 user2, token2 = generate_user()

358 user3, token3 = generate_user()

359 user4, token4 = generate_user()

360

361 with api_session(token2) as api:

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

363

364 with api_session(token3) as api:

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

366

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

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

369 assert len(res.notifications) == 2

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

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

372 # should be listed desc time

373 assert notification_ids[0] > notification_ids[1]

374

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

376

377 with api_session(token4) as api:

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

379

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

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

382 notifications.MarkAllNotificationsSeen(

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

384 )

385

386 # last one is still unseen

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

388

389 # mark the first one unseen

390 notifications.MarkNotificationSeen(

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

392 )

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

394

395 # mark the last one seen

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

397 assert len(res.notifications) == 3

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

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

400

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

402

403 notifications.MarkNotificationSeen(

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

405 )

406

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

408 assert len(res.notifications) == 3

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

410

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

412

413

414def test_GetVapidPublicKey(db):

415 _, token = generate_user()

416

417 with notifications_session(token) as notifications:

418 assert (

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

420 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"

421 )

422

423

424def test_RegisterPushNotificationSubscription(db):

425 _, token = generate_user()

426

427 subscription_info = {

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

429 "expirationTime": None,

430 "keys": {

431 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",

432 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",

433 },

434 }

435

436 with notifications_session(token) as notifications:

437 res = notifications.RegisterPushNotificationSubscription(

438 notifications_pb2.RegisterPushNotificationSubscriptionReq(

439 full_subscription_json=json.dumps(subscription_info),

440 )

441 )

442

443

444def test_SendTestPushNotification(db, push_collector):

445 user, token = generate_user()

446

447 with notifications_session(token) as notifications:

448 notifications.SendTestPushNotification(empty_pb2.Empty())

449

450 push_collector.assert_user_has_count(user.id, 1)

451 push_collector.assert_user_push_matches_fields(

452 user.id,

453 title="Checking push notifications work!",

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

455 )

456

457 # the above two are equivalent to this

458

459 push_collector.assert_user_has_single_matching(

460 user.id,

461 title="Checking push notifications work!",

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

463 )

464

465

466def test_SendBlogPostNotification(db, push_collector):

467 super_user, super_token = generate_user(is_superuser=True)

468

469 user1, user1_token = generate_user()

470 # enabled email

471 user2, user2_token = generate_user()

472 # disabled push

473 user3, user3_token = generate_user()

474

475 topic_action = NotificationTopicAction.general__new_blog_post

476

477 with notifications_session(user2_token) as notifications:

478 notifications.SetNotificationSettings(

479 notifications_pb2.SetNotificationSettingsReq(

480 preferences=[

481 notifications_pb2.SingleNotificationPreference(

482 topic=topic_action.topic,

483 action=topic_action.action,

484 delivery_method="email",

485 enabled=True,

486 )

487 ],

488 )

489 )

490

491 with notifications_session(user3_token) as notifications:

492 notifications.SetNotificationSettings(

493 notifications_pb2.SetNotificationSettingsReq(

494 preferences=[

495 notifications_pb2.SingleNotificationPreference(

496 topic=topic_action.topic,

497 action=topic_action.action,

498 delivery_method="push",

499 enabled=False,

500 )

501 ],

502 )

503 )

504

505 with mock_notification_email() as mock:

506 with real_admin_session(super_token) as admin_api:

507 admin_api.SendBlogPostNotification(

508 admin_pb2.SendBlogPostNotificationReq(

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

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

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

512 )

513 )

514

515 process_jobs()

516

517 assert mock.call_count == 1

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

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

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

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

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

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

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

525

526 push_collector.assert_user_has_count(user1.id, 1)

527 push_collector.assert_user_push_matches_fields(

528 user1.id,

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

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

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

532 )

533

534 push_collector.assert_user_has_count(user2.id, 1)

535 push_collector.assert_user_push_matches_fields(

536 user2.id,

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

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

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

540 )

541

542 push_collector.assert_user_has_count(user3.id, 0)

543

544

545def test_get_topic_actions_by_delivery_type(db):

546 user, token = generate_user()

547

548 # these are enabled by default

549 assert NotificationDeliveryType.push in NotificationTopicAction.reference__receive_friend.defaults

550 assert NotificationDeliveryType.push in NotificationTopicAction.host_request__accept.defaults

551

552 # these are disabled by default

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

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

555

556 with notifications_session(token) as notifications:

557 notifications.SetNotificationSettings(

558 notifications_pb2.SetNotificationSettingsReq(

559 preferences=[

560 notifications_pb2.SingleNotificationPreference(

561 topic=NotificationTopicAction.reference__receive_friend.topic,

562 action=NotificationTopicAction.reference__receive_friend.action,

563 delivery_method="push",

564 enabled=False,

565 ),

566 notifications_pb2.SingleNotificationPreference(

567 topic=NotificationTopicAction.event__create_any.topic,

568 action=NotificationTopicAction.event__create_any.action,

569 delivery_method="push",

570 enabled=True,

571 ),

572 ],

573 )

574 )

575

576 with session_scope() as session:

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

578 assert NotificationTopicAction.reference__receive_friend not in deliver

579 assert NotificationTopicAction.host_request__accept in deliver

580 assert NotificationTopicAction.event__create_any in deliver

581 assert NotificationTopicAction.discussion__create not in deliver

582 assert NotificationTopicAction.account_deletion__start in deliver