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

190 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-20 21:46 +0000

1import json 

2import re 

3from urllib.parse import parse_qs, urlparse 

4 

5import grpc 

6import pytest 

7from google.protobuf import empty_pb2 

8 

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.sql import couchers_select as select 

23from proto import api_pb2, auth_pb2, conversations_pb2, notification_data_pb2, notifications_pb2 

24from proto.internal import unsubscribe_pb2 

25from tests.test_fixtures import ( # noqa 

26 api_session, 

27 auth_api_session, 

28 conversations_session, 

29 db, 

30 email_fields, 

31 generate_user, 

32 mock_notification_email, 

33 notifications_session, 

34 process_jobs, 

35 push_collector, 

36 session_scope, 

37 testconfig, 

38) 

39 

40 

41@pytest.fixture(autouse=True) 

42def _(testconfig): 

43 pass 

44 

45 

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

47def test_SetNotificationSettings_preferences_respected_editable(db, enabled): 

48 user, token = generate_user() 

49 

50 # enable a notification type and check it gets delivered 

51 topic_action = NotificationTopicAction.badge__add 

52 

53 with notifications_session(token) as notifications: 

54 notifications.SetNotificationSettings( 

55 notifications_pb2.SetNotificationSettingsReq( 

56 preferences=[ 

57 notifications_pb2.SingleNotificationPreference( 

58 topic=topic_action.topic, 

59 action=topic_action.action, 

60 delivery_method="push", 

61 enabled=enabled, 

62 ) 

63 ], 

64 ) 

65 ) 

66 

67 notify( 

68 user_id=user.id, 

69 topic_action=topic_action.display, 

70 data=notification_data_pb2.BadgeAdd( 

71 badge_id="volunteer", 

72 badge_name="Active Volunteer", 

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

74 ), 

75 ) 

76 

77 process_job() 

78 

79 with session_scope() as session: 

80 deliv = session.execute( 

81 select(NotificationDelivery) 

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

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

84 .where(Notification.topic_action == topic_action) 

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

86 ).scalar_one_or_none() 

87 

88 if enabled: 

89 assert deliv is not None 

90 else: 

91 assert deliv is None 

92 

93 

94def test_SetNotificationSettings_preferences_not_editable(db): 

95 user, token = generate_user() 

96 

97 # enable a notification type and check it gets delivered 

98 topic_action = NotificationTopicAction.password_reset__start 

99 

100 with notifications_session(token) as notifications: 

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

102 notifications.SetNotificationSettings( 

103 notifications_pb2.SetNotificationSettingsReq( 

104 preferences=[ 

105 notifications_pb2.SingleNotificationPreference( 

106 topic=topic_action.topic, 

107 action=topic_action.action, 

108 delivery_method="push", 

109 enabled=False, 

110 ) 

111 ], 

112 ) 

113 ) 

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

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

116 

117 

118def test_unsubscribe(db): 

119 # this is the ugliest test i've written 

120 

121 user, token = generate_user() 

122 

123 topic_action = NotificationTopicAction.badge__add 

124 

125 # first enable email notifs 

126 with notifications_session(token) as notifications: 

127 notifications.SetNotificationSettings( 

128 notifications_pb2.SetNotificationSettingsReq( 

129 preferences=[ 

130 notifications_pb2.SingleNotificationPreference( 

131 topic=topic_action.topic, 

132 action=topic_action.action, 

133 delivery_method=method, 

134 enabled=enabled, 

135 ) 

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

137 ], 

138 ) 

139 ) 

140 

141 with mock_notification_email() as mock: 

142 notify( 

143 user_id=user.id, 

144 topic_action=topic_action.display, 

145 data=notification_data_pb2.BadgeAdd( 

146 badge_id="volunteer", 

147 badge_name="Active Volunteer", 

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

149 ), 

150 ) 

151 

152 assert mock.call_count == 1 

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

154 # very ugly 

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

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

157 if "payload" not in link: 

158 continue 

159 print(link) 

160 url_parts = urlparse(link) 

161 params = parse_qs(url_parts.query) 

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

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

164 if payload.HasField("topic_action"): 

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

166 res = auth_api.Unsubscribe( 

167 auth_pb2.UnsubscribeReq( 

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

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

170 ) 

171 ) 

172 break 

173 else: 

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

175 

176 with notifications_session(token) as notifications: 

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

178 

179 for group in res.groups: 

180 for topic in group.topics: 

181 for item in topic.items: 

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

183 assert not item.email 

184 

185 with mock_notification_email() as mock: 

186 notify( 

187 user_id=user.id, 

188 topic_action=topic_action.display, 

189 data=notification_data_pb2.BadgeAdd( 

190 badge_id="volunteer", 

191 badge_name="Active Volunteer", 

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

193 ), 

194 ) 

195 

196 assert mock.call_count == 0 

197 

198 

199def test_unsubscribe_do_not_email(db): 

200 user, token = generate_user() 

201 

202 _, token2 = generate_user(complete_profile=True) 

203 with mock_notification_email() as mock: 

204 with api_session(token2) as api: 

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

206 

207 assert mock.call_count == 1 

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

209 # very ugly 

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

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

212 if "payload" not in link: 

213 continue 

214 print(link) 

215 url_parts = urlparse(link) 

216 params = parse_qs(url_parts.query) 

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

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

219 if payload.HasField("do_not_email"): 

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

221 res = auth_api.Unsubscribe( 

222 auth_pb2.UnsubscribeReq( 

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

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

225 ) 

226 ) 

227 break 

228 else: 

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

230 

231 _, token3 = generate_user(complete_profile=True) 

232 with mock_notification_email() as mock: 

233 with api_session(token3) as api: 

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

235 

236 assert mock.call_count == 0 

237 

238 with session_scope() as session: 

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

240 assert user_.do_not_email 

241 

242 

243def test_get_do_not_email(db): 

244 _, token = generate_user() 

245 

246 with session_scope() as session: 

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

248 user.do_not_email = False 

249 

250 with notifications_session(token) as notifications: 

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

252 assert not res.do_not_email_enabled 

253 

254 with session_scope() as session: 

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

256 user.do_not_email = True 

257 user.hosting_status = HostingStatus.cant_host 

258 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

259 

260 with notifications_session(token) as notifications: 

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

262 assert res.do_not_email_enabled 

263 

264 

265def test_set_do_not_email(db): 

266 _, token = generate_user() 

267 

268 with session_scope() as session: 

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

270 user.do_not_email = False 

271 user.hosting_status = HostingStatus.can_host 

272 user.meetup_status = MeetupStatus.wants_to_meetup 

273 

274 with notifications_session(token) as notifications: 

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

276 

277 with session_scope() as session: 

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

279 assert not user.do_not_email 

280 

281 with notifications_session(token) as notifications: 

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

283 

284 with session_scope() as session: 

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

286 assert user.do_not_email 

287 assert user.hosting_status == HostingStatus.cant_host 

288 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup 

289 

290 with notifications_session(token) as notifications: 

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

292 

293 with session_scope() as session: 

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

295 assert not user.do_not_email 

296 

297 

298def test_list_notifications(db, push_collector): 

299 user1, token1 = generate_user() 

300 user2, token2 = generate_user() 

301 

302 with api_session(token2) as api: 

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

304 

305 with notifications_session(token1) as notifications: 

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

307 assert len(res.notifications) == 1 

308 

309 n = res.notifications[0] 

310 

311 assert n.topic == "friend_request" 

312 assert n.action == "create" 

313 assert n.key == "2" 

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

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

316 assert n.icon == "http://localhost:3000/logo512.png" 

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

318 

319 with conversations_session(token2) as c: 

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

321 group_chat_id = res.group_chat_id 

322 for i in range(17): 

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

324 

325 process_jobs() 

326 

327 all_notifs = [] 

328 with notifications_session(token1) as notifications: 

329 page_token = None 

330 for _ in range(100): 

331 res = notifications.ListNotifications( 

332 notifications_pb2.ListNotificationsReq( 

333 page_size=5, 

334 page_token=page_token, 

335 ) 

336 ) 

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

338 all_notifs += res.notifications 

339 page_token = res.next_page_token 

340 if not page_token: 

341 break 

342 

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

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

345 

346 

347def test_GetVapidPublicKey(db): 

348 _, token = generate_user() 

349 

350 with notifications_session(token) as notifications: 

351 assert ( 

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

353 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A" 

354 ) 

355 

356 

357def test_RegisterPushNotification(db): 

358 _, token = generate_user() 

359 

360 subscription_info = { 

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

362 "expirationTime": None, 

363 "keys": { 

364 "auth": "TnuEJ1OdfEkf6HKcUovl0Q", 

365 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0", 

366 }, 

367 } 

368 

369 with notifications_session(token) as notifications: 

370 res = notifications.RegisterPushNotification( 

371 notifications_pb2.RegisterPushNotificationReq( 

372 endpoint=subscription_info["endpoint"], 

373 auth_key=subscription_info["keys"]["auth"], 

374 p256dh_key=subscription_info["keys"]["p256dh"], 

375 full_subscription_json=json.dumps(subscription_info), 

376 ) 

377 ) 

378 

379 

380def test_SendTestPushNotification(db, push_collector): 

381 user, token = generate_user() 

382 

383 with notifications_session(token) as notifications: 

384 notifications.SendTestPushNotification(empty_pb2.Empty()) 

385 

386 push_collector.assert_user_has_count(user.id, 1) 

387 push_collector.assert_user_push_matches_fields( 

388 user.id, 

389 title="Checking push notifications work!", 

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

391 ) 

392 

393 # the above two are equivalent to this 

394 

395 push_collector.assert_user_has_single_matching( 

396 user.id, 

397 title="Checking push notifications work!", 

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

399 )