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

234 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-11-04 02:51 +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 session_scope() as session: 

59 with mock_notification_email() as mock: 

60 send_signup_email(session, flow) 

61 

62 assert mock.call_count == 1 

63 e = email_fields(mock) 

64 assert e.recipient == request_email 

65 assert flow.email_token in e.plain 

66 assert flow.email_token in e.html 

67 

68 

69def test_report_email(db): 

70 user_reporter, api_token_author = generate_user() 

71 user_author, api_token_reported = generate_user() 

72 

73 report = ContentReport( 

74 reporting_user=user_reporter, 

75 reason="spam", 

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

77 content_ref="comment/123", 

78 author_user=user_author, 

79 user_agent="n/a", 

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

81 ) 

82 

83 with session_scope() as session: 

84 with mock_notification_email() as mock: 

85 send_content_report_email(session, report) 

86 

87 assert mock.call_count == 1 

88 

89 e = email_fields(mock) 

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

91 assert report.author_user.username in e.plain 

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

93 assert report.author_user.email in e.plain 

94 assert report.reporting_user.username in e.plain 

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

96 assert report.reporting_user.email in e.plain 

97 assert report.reason in e.plain 

98 assert report.description in e.plain 

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

100 

101 

102def test_reference_report_email_not_sent(db): 

103 with session_scope() as session: 

104 from_user, api_token_author = generate_user() 

105 to_user, api_token_reported = generate_user() 

106 

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

108 session.add(friend_relationship) 

109 session.flush() 

110 

111 reference = Reference( 

112 from_user=from_user, 

113 to_user=to_user, 

114 reference_type=ReferenceType.friend, 

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

116 rating=0.9, 

117 was_appropriate=True, 

118 ) 

119 

120 # no email sent for a positive ref 

121 

122 with mock_notification_email() as mock: 

123 maybe_send_reference_report_email(session, reference) 

124 

125 assert mock.call_count == 0 

126 

127 

128def test_reference_report_email(db): 

129 with session_scope() as session: 

130 from_user, api_token_author = generate_user() 

131 to_user, api_token_reported = generate_user() 

132 

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

134 session.add(friend_relationship) 

135 session.flush() 

136 

137 reference = Reference( 

138 from_user=from_user, 

139 to_user=to_user, 

140 reference_type=ReferenceType.friend, 

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

142 rating=0.3, 

143 was_appropriate=False, 

144 ) 

145 

146 with mock_notification_email() as mock: 

147 maybe_send_reference_report_email(session, reference) 

148 

149 assert mock.call_count == 1 

150 e = email_fields(mock) 

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

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

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

154 assert reference.from_user.username in e.plain 

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

156 assert reference.from_user.email in e.plain 

157 assert reference.to_user.username in e.plain 

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

159 assert reference.to_user.email in e.plain 

160 assert reference.text in e.plain 

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

162 

163 

164def test_email_patching_fails(db): 

165 """ 

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

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

168 actually done 

169 """ 

170 to_user, to_token = generate_user() 

171 from_user, from_token = generate_user() 

172 

173 patched_msg = random_hex(64) 

174 

175 def mock_queue_email(session, **kwargs): 

176 raise Exception(patched_msg) 

177 

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

179 with pytest.raises(Exception) as e: 

180 with api_session(from_token) as api: 

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

182 process_jobs() 

183 

184 assert str(e.value) == patched_msg 

185 

186 

187def test_email_changed_confirmation_sent_to_new_email(db): 

188 confirmation_token = urlsafe_secure_token() 

189 user, user_token = generate_user() 

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

191 user.new_email_token = confirmation_token 

192 with session_scope() as session: 

193 with mock_notification_email() as mock: 

194 send_email_changed_confirmation_to_new_email(session, user) 

195 

196 assert mock.call_count == 1 

197 e = email_fields(mock) 

198 assert "new email" in e.subject 

199 assert e.recipient == user.new_email 

200 assert user.name in e.plain 

201 assert user.name in e.html 

202 assert user.email in e.plain 

203 assert user.email in e.html 

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

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

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

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

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

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

210 

211 

212def test_do_not_email_security(db): 

213 user, token = generate_user() 

214 

215 password_reset_token = urlsafe_secure_token() 

216 

217 with notifications_session(token) as notifications: 

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

219 

220 # make sure we still get security emails 

221 

222 with mock_notification_email() as mock: 

223 with session_scope() as session: 

224 notify( 

225 session, 

226 user_id=user.id, 

227 topic_action="password_reset:start", 

228 data=notification_data_pb2.PasswordResetStart( 

229 password_reset_token=password_reset_token, 

230 ), 

231 ) 

232 

233 assert mock.call_count == 1 

234 e = email_fields(mock) 

235 assert e.recipient == user.email 

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

237 assert password_reset_token in e.plain 

238 assert password_reset_token in e.html 

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

240 assert unique_string in e.plain 

241 assert unique_string in e.html 

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

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

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

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

246 

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

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

249 

250 

251def test_do_not_email_non_security(db): 

252 user, token1 = generate_user(complete_profile=True) 

253 _, token2 = generate_user(complete_profile=True) 

254 

255 with notifications_session(token1) as notifications: 

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

257 

258 with mock_notification_email() as mock: 

259 with api_session(token2) as api: 

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

261 

262 assert mock.call_count == 0 

263 

264 

265def test_do_not_email_non_security_unsublink(db): 

266 user, _ = generate_user(complete_profile=True) 

267 _, token2 = generate_user(complete_profile=True) 

268 

269 with mock_notification_email() as mock: 

270 with api_session(token2) as api: 

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

272 

273 assert mock.call_count == 1 

274 e = email_fields(mock) 

275 

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

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

278 

279 

280def test_email_prefix_config(db, monkeypatch): 

281 user, _ = generate_user() 

282 

283 with mock_notification_email() as mock: 

284 with session_scope() as session: 

285 notify( 

286 session, 

287 user_id=user.id, 

288 topic_action="donation:received", 

289 data=notification_data_pb2.DonationReceived( 

290 amount=20, 

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

292 ), 

293 ) 

294 

295 assert mock.call_count == 1 

296 e = email_fields(mock) 

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

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

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

300 

301 new_config = config.copy() 

302 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo" 

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

304 new_config["NOTIFICATION_PREFIX"] = "" 

305 

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

307 

308 with mock_notification_email() as mock: 

309 with session_scope() as session: 

310 notify( 

311 session, 

312 user_id=user.id, 

313 topic_action="donation:received", 

314 data=notification_data_pb2.DonationReceived( 

315 amount=20, 

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

317 ), 

318 ) 

319 

320 assert mock.call_count == 1 

321 e = email_fields(mock) 

322 assert e.sender_name == "TestCo" 

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

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

325 

326 

327def test_send_donation_email(db, monkeypatch): 

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

329 

330 new_config = config.copy() 

331 new_config["ENABLE_EMAIL"] = True 

332 

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

334 

335 with session_scope() as session: 

336 notify( 

337 session, 

338 user_id=user.id, 

339 topic_action="donation:received", 

340 data=notification_data_pb2.DonationReceived( 

341 amount=20, 

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

343 ), 

344 ) 

345 

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

347 process_jobs() 

348 

349 with session_scope() as session: 

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

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

352 assert ( 

353 email.plain 

354 == """Dear Testy von Test, 

355 

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

357 

358Your 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. 

359 

360 

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

362 

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

364 

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

366 

367Your generosity will help deliver the platform for everyone. 

368 

369 

370Thank you! 

371 

372Aapeli and Itsi, 

373Couchers.org Founders 

374 

375 

376--- 

377 

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

379 ) 

380 

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

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

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

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

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

386 assert not email.list_unsubscribe_header 

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

388 

389 

390def test_email_deleted_users_regression(db): 

391 """ 

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

393 """ 

394 super_user, super_token = generate_user(is_superuser=True) 

395 creating_user, creating_token = generate_user(complete_profile=True) 

396 

397 normal_user, _ = generate_user() 

398 ban_user, _ = generate_user() 

399 delete_user, _ = generate_user() 

400 

401 with session_scope() as session: 

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

403 create_community( 

404 session, 

405 0, 

406 2, 

407 "Non-global Community", 

408 [super_user], 

409 [creating_user, normal_user, ban_user, delete_user], 

410 None, 

411 ) 

412 

413 enforce_community_memberships() 

414 

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

416 end_time = start_time + timedelta(hours=3) 

417 with events_session(creating_token) as api: 

418 res = api.CreateEvent( 

419 events_pb2.CreateEventReq( 

420 title="Dummy Title", 

421 content="Dummy content.", 

422 photo_key=None, 

423 offline_information=events_pb2.OfflineEventInformation( 

424 address="Near Null Island", 

425 lat=0.1, 

426 lng=0.2, 

427 ), 

428 start_time=Timestamp_from_datetime(start_time), 

429 end_time=Timestamp_from_datetime(end_time), 

430 timezone="UTC", 

431 ) 

432 ) 

433 event_id = res.event_id 

434 assert not res.is_deleted 

435 

436 with mock_notification_email() as mock: 

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

438 assert mock.call_count == 1 

439 

440 with real_admin_session(super_token) as admin: 

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

442 assert len(res.requests) == 1 

443 # this will count everyone 

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

445 

446 with session_scope() as session: 

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

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

449 

450 with real_admin_session(super_token) as admin: 

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

452 assert len(res.requests) == 1 

453 # should only notify creating_user, super_user and normal_user 

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

455 

456 with mock_notification_email() as mock: 

457 admin.DecideEventCommunityInviteRequest( 

458 admin_pb2.DecideEventCommunityInviteRequestReq( 

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

460 approve=True, 

461 ) 

462 ) 

463 

464 assert mock.call_count == 3