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

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.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)

41@pytest.fixture(autouse=True)

42def _(testconfig):

43 pass

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

47def test_SetNotificationSettings_preferences_respected_editable(db, enabled):

48 user, token = generate_user()

50 # enable a notification type and check it gets delivered

51 topic_action = NotificationTopicAction.badge__add

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 )

67 with session_scope() as session:

68 notify(

69 session,

70 user_id=user.id,

71 topic_action=topic_action.display,

72 data=notification_data_pb2.BadgeAdd(

73 badge_id="volunteer",

74 badge_name="Active Volunteer",

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

76 ),

77 )

79 process_job()

81 with session_scope() as session:

82 deliv = session.execute(

83 select(NotificationDelivery)

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

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

86 .where(Notification.topic_action == topic_action)

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

88 ).scalar_one_or_none()

90 if enabled:

91 assert deliv is not None

92 else:

93 assert deliv is None

96def test_SetNotificationSettings_preferences_not_editable(db):

97 user, token = generate_user()

99 # enable a notification type and check it gets delivered

100 topic_action = NotificationTopicAction.password_reset__start

101

102 with notifications_session(token) as notifications:

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

104 notifications.SetNotificationSettings(

105 notifications_pb2.SetNotificationSettingsReq(

106 preferences=[

107 notifications_pb2.SingleNotificationPreference(

108 topic=topic_action.topic,

109 action=topic_action.action,

110 delivery_method="push",

111 enabled=False,

112 )

113 ],

114 )

115 )

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

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

118

119

120def test_unsubscribe(db):

121 # this is the ugliest test i've written

122

123 user, token = generate_user()

124

125 topic_action = NotificationTopicAction.badge__add

126

127 # first enable email notifs

128 with notifications_session(token) as notifications:

129 notifications.SetNotificationSettings(

130 notifications_pb2.SetNotificationSettingsReq(

131 preferences=[

132 notifications_pb2.SingleNotificationPreference(

133 topic=topic_action.topic,

134 action=topic_action.action,

135 delivery_method=method,

136 enabled=enabled,

137 )

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

139 ],

140 )

141 )

142

143 with mock_notification_email() as mock:

144 with session_scope() as session:

145 notify(

146 session,

147 user_id=user.id,

148 topic_action=topic_action.display,

149 data=notification_data_pb2.BadgeAdd(

150 badge_id="volunteer",

151 badge_name="Active Volunteer",

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

153 ),

154 )

155

156 assert mock.call_count == 1

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

158 # very ugly

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

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

161 if "payload" not in link:

162 continue

163 print(link)

164 url_parts = urlparse(link)

165 params = parse_qs(url_parts.query)

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

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

168 if payload.HasField("topic_action"):

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

170 res = auth_api.Unsubscribe(

171 auth_pb2.UnsubscribeReq(

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

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

174 )

175 )

176 break

177 else:

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

179

180 with notifications_session(token) as notifications:

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

182

183 for group in res.groups:

184 for topic in group.topics:

185 for item in topic.items:

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

187 assert not item.email

188

189 with mock_notification_email() as mock:

190 with session_scope() as session:

191 notify(

192 session,

193 user_id=user.id,

194 topic_action=topic_action.display,

195 data=notification_data_pb2.BadgeAdd(

196 badge_id="volunteer",

197 badge_name="Active Volunteer",

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

199 ),

200 )

201

202 assert mock.call_count == 0

203

204

205def test_unsubscribe_do_not_email(db):

206 user, token = generate_user()

207

208 _, token2 = generate_user(complete_profile=True)

209 with mock_notification_email() as mock:

210 with api_session(token2) as api:

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

212

213 assert mock.call_count == 1

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

215 # very ugly

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

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

218 if "payload" not in link:

219 continue

220 print(link)

221 url_parts = urlparse(link)

222 params = parse_qs(url_parts.query)

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

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

225 if payload.HasField("do_not_email"):

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

227 res = auth_api.Unsubscribe(

228 auth_pb2.UnsubscribeReq(

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

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

231 )

232 )

233 break

234 else:

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

236

237 _, token3 = generate_user(complete_profile=True)

238 with mock_notification_email() as mock:

239 with api_session(token3) as api:

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

241

242 assert mock.call_count == 0

243

244 with session_scope() as session:

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

246 assert user_.do_not_email

247

248

249def test_get_do_not_email(db):

250 _, token = generate_user()

251

252 with session_scope() as session:

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

254 user.do_not_email = False

255

256 with notifications_session(token) as notifications:

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

258 assert not res.do_not_email_enabled

259

260 with session_scope() as session:

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

262 user.do_not_email = True

263 user.hosting_status = HostingStatus.cant_host

264 user.meetup_status = MeetupStatus.does_not_want_to_meetup

265

266 with notifications_session(token) as notifications:

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

268 assert res.do_not_email_enabled

269

270

271def test_set_do_not_email(db):

272 _, token = generate_user()

273

274 with session_scope() as session:

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

276 user.do_not_email = False

277 user.hosting_status = HostingStatus.can_host

278 user.meetup_status = MeetupStatus.wants_to_meetup

279

280 with notifications_session(token) as notifications:

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

282

283 with session_scope() as session:

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

285 assert not user.do_not_email

286

287 with notifications_session(token) as notifications:

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

289

290 with session_scope() as session:

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

292 assert user.do_not_email

293 assert user.hosting_status == HostingStatus.cant_host

294 assert user.meetup_status == MeetupStatus.does_not_want_to_meetup

295

296 with notifications_session(token) as notifications:

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

298

299 with session_scope() as session:

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

301 assert not user.do_not_email

302

303

304def test_list_notifications(db, push_collector):

305 user1, token1 = generate_user()

306 user2, token2 = generate_user()

307

308 with api_session(token2) as api:

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

310

311 with notifications_session(token1) as notifications:

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

313 assert len(res.notifications) == 1

314

315 n = res.notifications[0]

316

317 assert n.topic == "friend_request"

318 assert n.action == "create"

319 assert n.key == "2"

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

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

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

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

324

325 with conversations_session(token2) as c:

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

327 group_chat_id = res.group_chat_id

328 for i in range(17):

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

330

331 process_jobs()

332

333 all_notifs = []

334 with notifications_session(token1) as notifications:

335 page_token = None

336 for _ in range(100):

337 res = notifications.ListNotifications(

338 notifications_pb2.ListNotificationsReq(

339 page_size=5,

340 page_token=page_token,

341 )

342 )

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

344 all_notifs += res.notifications

345 page_token = res.next_page_token

346 if not page_token:

347 break

348

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

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

351

352

353def test_GetVapidPublicKey(db):

354 _, token = generate_user()

355

356 with notifications_session(token) as notifications:

357 assert (

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

359 == "BApMo2tGuon07jv-pEaAKZmVo6E_d4HfcdDeV6wx2k9wV8EovJ0ve00bdLzZm9fizDrGZXRYJFqCcRJUfBcgA0A"

360 )

361

362

363def test_RegisterPushNotificationSubscription(db):

364 _, token = generate_user()

365

366 subscription_info = {

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

368 "expirationTime": None,

369 "keys": {

370 "auth": "TnuEJ1OdfEkf6HKcUovl0Q",

371 "p256dh": "BK7Rp8og3eFJPqm0ofR8F-l2mtNCCCWYo6f_5kSs8jPEFiKetnZHNOglvC6IrgU9vHmgFHlG7gHGtB1HM599sy0",

372 },

373 }

374

375 with notifications_session(token) as notifications:

376 res = notifications.RegisterPushNotificationSubscription(

377 notifications_pb2.RegisterPushNotificationSubscriptionReq(

378 full_subscription_json=json.dumps(subscription_info),

379 )

380 )

381

382

383def test_SendTestPushNotification(db, push_collector):

384 user, token = generate_user()

385

386 with notifications_session(token) as notifications:

387 notifications.SendTestPushNotification(empty_pb2.Empty())

388

389 push_collector.assert_user_has_count(user.id, 1)

390 push_collector.assert_user_push_matches_fields(

391 user.id,

392 title="Checking push notifications work!",

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

394 )

395

396 # the above two are equivalent to this

397

398 push_collector.assert_user_has_single_matching(

399 user.id,

400 title="Checking push notifications work!",

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

402 )