Coverage for src/tests/test_email.py: 100%

229 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 16:44 +0000

1from unittest.mock import patch 

2 

3import pytest 

4 

5import couchers.email 

6import couchers.jobs.handlers 

7from couchers.config import config 

8from couchers.crypto import random_hex, urlsafe_secure_token 

9from couchers.db import session_scope 

10from couchers.models import ( 

11 ContentReport, 

12 Email, 

13 FriendRelationship, 

14 FriendStatus, 

15 Reference, 

16 ReferenceType, 

17 SignupFlow, 

18 User, 

19) 

20from couchers.notifications.notify import notify 

21from couchers.sql import couchers_select as select 

22from couchers.tasks import ( 

23 enforce_community_memberships, 

24 maybe_send_reference_report_email, 

25 send_content_report_email, 

26 send_email_changed_confirmation_to_new_email, 

27 send_signup_email, 

28) 

29from couchers.utils import Timestamp_from_datetime, now, timedelta 

30from proto import admin_pb2, api_pb2, events_pb2, notification_data_pb2, notifications_pb2 

31from tests.test_communities import create_community 

32from tests.test_fixtures import ( # noqa 

33 api_session, 

34 db, 

35 email_fields, 

36 events_session, 

37 generate_user, 

38 mock_notification_email, 

39 notifications_session, 

40 process_jobs, 

41 push_collector, 

42 real_admin_session, 

43 session_scope, 

44 testconfig, 

45) 

46 

47 

48@pytest.fixture(autouse=True) 

49def _(testconfig): 

50 pass 

51 

52 

53def test_signup_verification_email(db): 

54 request_email = f"{random_hex(12)}@couchers.org.invalid" 

55 

56 flow = SignupFlow(name="Frodo", email=request_email) 

57 

58 with mock_notification_email() as mock: 

59 send_signup_email(flow) 

60 

61 assert mock.call_count == 1 

62 e = email_fields(mock) 

63 assert e.recipient == request_email 

64 assert flow.email_token in e.plain 

65 assert flow.email_token in e.html 

66 

67 

68def test_report_email(db): 

69 user_reporter, api_token_author = generate_user() 

70 user_author, api_token_reported = generate_user() 

71 

72 report = ContentReport( 

73 reporting_user=user_reporter, 

74 reason="spam", 

75 description="I think this is spam and does not belong on couchers", 

76 content_ref="comment/123", 

77 author_user=user_author, 

78 user_agent="n/a", 

79 page="https://couchers.org/comment/123", 

80 ) 

81 

82 with mock_notification_email() as mock: 

83 send_content_report_email(report) 

84 

85 assert mock.call_count == 1 

86 

87 e = email_fields(mock) 

88 assert e.recipient == "reports@couchers.org.invalid" 

89 assert report.author_user.username in e.plain 

90 assert str(report.author_user.id) in e.plain 

91 assert report.author_user.email in e.plain 

92 assert report.reporting_user.username in e.plain 

93 assert str(report.reporting_user.id) in e.plain 

94 assert report.reporting_user.email in e.plain 

95 assert report.reason in e.plain 

96 assert report.description in e.plain 

97 assert "report" in e.subject.lower() 

98 

99 

100def test_reference_report_email_not_sent(db): 

101 with session_scope() as session: 

102 from_user, api_token_author = generate_user() 

103 to_user, api_token_reported = generate_user() 

104 

105 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted) 

106 session.add(friend_relationship) 

107 session.flush() 

108 

109 reference = Reference( 

110 from_user=from_user, 

111 to_user=to_user, 

112 reference_type=ReferenceType.friend, 

113 text="This person was very nice to me.", 

114 rating=0.9, 

115 was_appropriate=True, 

116 ) 

117 

118 # no email sent for a positive ref 

119 

120 with mock_notification_email() as mock: 

121 maybe_send_reference_report_email(reference) 

122 

123 assert mock.call_count == 0 

124 

125 

126def test_reference_report_email(db): 

127 with session_scope() as session: 

128 from_user, api_token_author = generate_user() 

129 to_user, api_token_reported = generate_user() 

130 

131 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted) 

132 session.add(friend_relationship) 

133 session.flush() 

134 

135 reference = Reference( 

136 from_user=from_user, 

137 to_user=to_user, 

138 reference_type=ReferenceType.friend, 

139 text="This person was not nice to me.", 

140 rating=0.3, 

141 was_appropriate=False, 

142 ) 

143 

144 with mock_notification_email() as mock: 

145 maybe_send_reference_report_email(reference) 

146 

147 assert mock.call_count == 1 

148 e = email_fields(mock) 

149 assert e.recipient == "reports@couchers.org.invalid" 

150 assert "report" in e.subject.lower() 

151 assert "reference" in e.subject.lower() 

152 assert reference.from_user.username in e.plain 

153 assert str(reference.from_user.id) in e.plain 

154 assert reference.from_user.email in e.plain 

155 assert reference.to_user.username in e.plain 

156 assert str(reference.to_user.id) in e.plain 

157 assert reference.to_user.email in e.plain 

158 assert reference.text in e.plain 

159 assert "friend" in e.plain.lower() 

160 

161 

162def test_email_patching_fails(db): 

163 """ 

164 There was a problem where the mocking wasn't happening and the email dev 

165 printing function was called instead, this makes sure the patching is 

166 actually done 

167 """ 

168 to_user, to_token = generate_user() 

169 from_user, from_token = generate_user() 

170 

171 patched_msg = random_hex(64) 

172 

173 def mock_queue_email(**kwargs): 

174 raise Exception(patched_msg) 

175 

176 with patch("couchers.email._queue_email", mock_queue_email): 

177 with pytest.raises(Exception) as e: 

178 with api_session(from_token) as api: 

179 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id)) 

180 process_jobs() 

181 

182 assert str(e.value) == patched_msg 

183 

184 

185def test_email_changed_confirmation_sent_to_new_email(db): 

186 confirmation_token = urlsafe_secure_token() 

187 user, user_token = generate_user() 

188 user.new_email = f"{random_hex(12)}@couchers.org.invalid" 

189 user.new_email_token = confirmation_token 

190 with mock_notification_email() as mock: 

191 send_email_changed_confirmation_to_new_email(user) 

192 

193 assert mock.call_count == 1 

194 e = email_fields(mock) 

195 assert "new email" in e.subject 

196 assert e.recipient == user.new_email 

197 assert user.name in e.plain 

198 assert user.name in e.html 

199 assert user.email in e.plain 

200 assert user.email in e.html 

201 assert "Your old email address is" in e.plain 

202 assert "Your old email address is" in e.html 

203 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain 

204 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html 

205 assert "support@couchers.org" in e.plain 

206 assert "support@couchers.org" in e.html 

207 

208 

209def test_do_not_email_security(db): 

210 _, token = generate_user() 

211 

212 password_reset_token = urlsafe_secure_token() 

213 

214 with notifications_session(token) as notifications: 

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

216 

217 # make sure we still get security emails 

218 with session_scope() as session: 

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

220 

221 with mock_notification_email() as mock: 

222 notify( 

223 user_id=user.id, 

224 topic_action="password_reset:start", 

225 data=notification_data_pb2.PasswordResetStart( 

226 password_reset_token=password_reset_token, 

227 ), 

228 ) 

229 

230 assert mock.call_count == 1 

231 e = email_fields(mock) 

232 assert e.recipient == user.email 

233 assert "reset" in e.subject.lower() 

234 assert password_reset_token in e.plain 

235 assert password_reset_token in e.html 

236 unique_string = "You asked for your password to be reset on Couchers.org." 

237 assert unique_string in e.plain 

238 assert unique_string in e.html 

239 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain 

240 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html 

241 assert "support@couchers.org" in e.plain 

242 assert "support@couchers.org" in e.html 

243 

244 assert "/unsubscribe?payload=" not in e.plain 

245 assert "/unsubscribe?payload=" not in e.html 

246 

247 

248def test_do_not_email_non_security(db): 

249 user, token1 = generate_user(complete_profile=True) 

250 _, token2 = generate_user(complete_profile=True) 

251 

252 with notifications_session(token1) as notifications: 

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

254 

255 with mock_notification_email() as mock: 

256 with api_session(token2) as api: 

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

258 

259 assert mock.call_count == 0 

260 

261 

262def test_do_not_email_non_security_unsublink(db): 

263 user, _ = generate_user(complete_profile=True) 

264 _, token2 = generate_user(complete_profile=True) 

265 

266 with mock_notification_email() as mock: 

267 with api_session(token2) as api: 

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

269 

270 assert mock.call_count == 1 

271 e = email_fields(mock) 

272 

273 assert "/unsubscribe?payload=" in e.plain 

274 assert "/unsubscribe?payload=" in e.html 

275 

276 

277def test_email_prefix_config(db, monkeypatch): 

278 user, _ = generate_user() 

279 

280 with mock_notification_email() as mock: 

281 notify( 

282 user_id=user.id, 

283 topic_action="donation:received", 

284 data=notification_data_pb2.DonationReceived( 

285 amount=20, 

286 receipt_url="https://example.com/receipt/12345", 

287 ), 

288 ) 

289 

290 assert mock.call_count == 1 

291 e = email_fields(mock) 

292 assert e.sender_name == "Couchers.org" 

293 assert e.sender_email == "notify@couchers.org.invalid" 

294 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!" 

295 

296 new_config = config.copy() 

297 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo" 

298 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid" 

299 new_config["NOTIFICATION_EMAIL_PREFIX"] = "" 

300 

301 monkeypatch.setattr(couchers.notifications.background, "config", new_config) 

302 

303 with mock_notification_email() as mock: 

304 notify( 

305 user_id=user.id, 

306 topic_action="donation:received", 

307 data=notification_data_pb2.DonationReceived( 

308 amount=20, 

309 receipt_url="https://example.com/receipt/12345", 

310 ), 

311 ) 

312 

313 assert mock.call_count == 1 

314 e = email_fields(mock) 

315 assert e.sender_name == "TestCo" 

316 assert e.sender_email == "testco@testing.co.invalid" 

317 assert e.subject == "Thank you for your donation to Couchers.org!" 

318 

319 

320def test_send_donation_email(db, monkeypatch): 

321 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid") 

322 

323 new_config = config.copy() 

324 new_config["ENABLE_EMAIL"] = True 

325 

326 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config) 

327 

328 notify( 

329 user_id=user.id, 

330 topic_action="donation:received", 

331 data=notification_data_pb2.DonationReceived( 

332 amount=20, 

333 receipt_url="https://example.com/receipt/12345", 

334 ), 

335 ) 

336 

337 with patch("couchers.email.smtp.smtplib.SMTP"): 

338 process_jobs() 

339 

340 with session_scope() as session: 

341 email = session.execute(select(Email)).scalar_one() 

342 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!" 

343 assert ( 

344 email.plain 

345 == """Dear Testy von Test, 

346 

347Thank you so much for your donation of $20 to Couchers.org. 

348 

349Your contribution will go towards building and sustaining the Couchers.org platform and community, and is vital for our goal of a completely free and non-profit generation of couch surfing. 

350 

351 

352You can download an invoice and receipt for the donation here: 

353 

354<https://example.com/receipt/12345> 

355 

356If you have any questions about your donation, please email us at <donations@couchers.org>. 

357 

358Your generosity will help deliver the platform for everyone. 

359 

360 

361Thank you! 

362 

363Aapeli and Itsi, 

364Couchers.org Founders 

365 

366 

367--- 

368 

369This is a security email, you cannot unsubscribe from it.""" 

370 ) 

371 

372 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html 

373 assert email.sender_name == "Couchers.org" 

374 assert email.sender_email == "notify@couchers.org.invalid" 

375 assert email.recipient == "testing@couchers.org.invalid" 

376 assert "https://example.com/receipt/12345" in email.html 

377 assert not email.list_unsubscribe_header 

378 assert email.source_data == "testing_version/donation_received" 

379 

380 

381def test_email_deleted_users_regression(db): 

382 """ 

383 We introduced a bug in notify v2 where we would email deleted/banned users. 

384 """ 

385 super_user, super_token = generate_user(is_superuser=True) 

386 creating_user, creating_token = generate_user(complete_profile=True) 

387 

388 normal_user, _ = generate_user() 

389 ban_user, _ = generate_user() 

390 delete_user, _ = generate_user() 

391 

392 with session_scope() as session: 

393 create_community(session, 10, 2, "Global Community", [super_user], [], None) 

394 create_community( 

395 session, 

396 0, 

397 2, 

398 "Non-global Community", 

399 [super_user], 

400 [creating_user, normal_user, ban_user, delete_user], 

401 None, 

402 ) 

403 

404 enforce_community_memberships() 

405 

406 start_time = now() + timedelta(hours=2) 

407 end_time = start_time + timedelta(hours=3) 

408 with events_session(creating_token) as api: 

409 res = api.CreateEvent( 

410 events_pb2.CreateEventReq( 

411 title="Dummy Title", 

412 content="Dummy content.", 

413 photo_key=None, 

414 offline_information=events_pb2.OfflineEventInformation( 

415 address="Near Null Island", 

416 lat=0.1, 

417 lng=0.2, 

418 ), 

419 start_time=Timestamp_from_datetime(start_time), 

420 end_time=Timestamp_from_datetime(end_time), 

421 timezone="UTC", 

422 ) 

423 ) 

424 event_id = res.event_id 

425 assert not res.is_deleted 

426 

427 with mock_notification_email() as mock: 

428 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id)) 

429 assert mock.call_count == 1 

430 

431 with real_admin_session(super_token) as admin: 

432 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq()) 

433 assert len(res.requests) == 1 

434 # this will count everyone 

435 assert res.requests[0].approx_users_to_notify == 5 

436 

437 with session_scope() as session: 

438 session.execute(select(User).where(User.id == ban_user.id)).scalar_one().is_banned = True 

439 session.execute(select(User).where(User.id == delete_user.id)).scalar_one().is_deleted = True 

440 

441 with real_admin_session(super_token) as admin: 

442 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq()) 

443 assert len(res.requests) == 1 

444 # should only notify creating_user, super_user and normal_user 

445 assert res.requests[0].approx_users_to_notify == 3 

446 

447 with mock_notification_email() as mock: 

448 admin.DecideEventCommunityInviteRequest( 

449 admin_pb2.DecideEventCommunityInviteRequestReq( 

450 event_community_invite_request_id=res.requests[0].event_community_invite_request_id, 

451 approve=True, 

452 ) 

453 ) 

454 

455 assert mock.call_count == 3