Coverage for src/tests/test_email.py: 100%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from datetime import timedelta
2from unittest.mock import patch
4import pytest
6import couchers.email
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 Conversation,
13 FriendRelationship,
14 FriendStatus,
15 HostRequest,
16 HostRequestStatus,
17 Message,
18 MessageType,
19 Reference,
20 ReferenceType,
21 SignupFlow,
22 Upload,
23 User,
24)
25from couchers.sql import couchers_select as select
26from couchers.tasks import (
27 maybe_send_reference_report_email,
28 send_account_deletion_confirmation_email,
29 send_account_deletion_successful_email,
30 send_account_recovered_email,
31 send_api_key_email,
32 send_content_report_email,
33 send_email_changed_confirmation_to_new_email,
34 send_email_changed_confirmation_to_old_email,
35 send_email_changed_notification_email,
36 send_friend_request_accepted_email,
37 send_friend_request_email,
38 send_login_email,
39 send_new_host_request_email,
40 send_password_reset_email,
41 send_signup_email,
42)
43from couchers.utils import now
44from tests.test_fixtures import db, generate_user, testconfig # noqa
47@pytest.fixture(autouse=True)
48def _(testconfig):
49 pass
52def test_login_email(db):
53 user, api_token = generate_user()
55 with session_scope() as session:
56 with patch("couchers.email.queue_email") as mock:
57 login_token = send_login_email(session, user)
59 assert mock.call_count == 1
60 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
61 assert recipient == user.email
62 assert "login" in subject.lower()
63 assert login_token.token in plain
64 assert login_token.token in html
67def test_signup_verification_email(db):
68 request_email = f"{random_hex(12)}@couchers.org.invalid"
70 with session_scope() as session:
71 flow = SignupFlow(name="Frodo", email=request_email)
73 with patch("couchers.email.queue_email") as mock:
74 send_signup_email(flow)
76 assert mock.call_count == 1
77 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
78 assert recipient == request_email
79 assert flow.email_token in plain
80 assert flow.email_token in html
83def test_report_email(db):
84 with session_scope():
85 user_reporter, api_token_author = generate_user()
86 user_author, api_token_reported = generate_user()
88 report = ContentReport(
89 reporting_user=user_reporter,
90 reason="spam",
91 description="I think this is spam and does not belong on couchers",
92 content_ref="comment/123",
93 author_user=user_author,
94 user_agent="n/a",
95 page="https://couchers.org/comment/123",
96 )
98 with patch("couchers.email.queue_email") as mock:
99 send_content_report_email(report)
101 assert mock.call_count == 1
103 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
104 assert recipient == "reports@couchers.org.invalid"
105 assert report.author_user.username in plain
106 assert str(report.author_user.id) in plain
107 assert report.author_user.email in plain
108 assert report.author_user.username in html
109 assert str(report.author_user.id) in html
110 assert report.author_user.email in html
111 assert report.reporting_user.username in plain
112 assert str(report.reporting_user.id) in plain
113 assert report.reporting_user.email in plain
114 assert report.reporting_user.username in html
115 assert str(report.reporting_user.id) in html
116 assert report.reporting_user.email in html
117 assert report.reason in plain
118 assert report.reason in html
119 assert report.description in plain
120 assert report.description in html
121 assert "report" in subject.lower()
124def test_reference_report_email_not_sent(db):
125 with session_scope() as session:
126 from_user, api_token_author = generate_user()
127 to_user, api_token_reported = generate_user()
129 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
130 session.add(friend_relationship)
131 session.flush()
133 reference = Reference(
134 from_user=from_user,
135 to_user=to_user,
136 reference_type=ReferenceType.friend,
137 text="This person was very nice to me.",
138 rating=0.9,
139 was_appropriate=True,
140 )
142 # no email sent for a positive ref
144 with patch("couchers.email.queue_email") as mock:
145 maybe_send_reference_report_email(reference)
147 assert mock.call_count == 0
150def test_reference_report_email(db):
151 with session_scope() as session:
152 from_user, api_token_author = generate_user()
153 to_user, api_token_reported = generate_user()
155 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
156 session.add(friend_relationship)
157 session.flush()
159 reference = Reference(
160 from_user=from_user,
161 to_user=to_user,
162 reference_type=ReferenceType.friend,
163 text="This person was not nice to me.",
164 rating=0.3,
165 was_appropriate=False,
166 )
168 with patch("couchers.email.queue_email") as mock:
169 maybe_send_reference_report_email(reference)
171 assert mock.call_count == 1
173 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
174 assert recipient == "reports@couchers.org.invalid"
175 assert "report" in subject.lower()
176 assert "reference" in subject.lower()
177 assert reference.from_user.username in plain
178 assert str(reference.from_user.id) in plain
179 assert reference.from_user.email in plain
180 assert reference.from_user.username in html
181 assert str(reference.from_user.id) in html
182 assert reference.from_user.email in html
183 assert reference.to_user.username in plain
184 assert str(reference.to_user.id) in plain
185 assert reference.to_user.email in plain
186 assert reference.to_user.username in html
187 assert str(reference.to_user.id) in html
188 assert reference.to_user.email in html
189 assert reference.text in plain
190 assert reference.text in html
191 assert "friend" in plain.lower()
192 assert "friend" in html.lower()
195def test_host_request_email(db):
196 with session_scope() as session:
197 host, api_token_to = generate_user()
198 # little trick here to get the upload correctly without invalidating users
199 key = random_hex(32)
200 filename = random_hex(32) + ".jpg"
201 session.add(
202 Upload(
203 key=key,
204 filename=filename,
205 creator_user_id=host.id,
206 )
207 )
208 session.commit()
209 surfer, api_token_from = generate_user(avatar_key=key)
210 from_date = "2020-01-01"
211 to_date = "2020-01-05"
213 conversation = Conversation()
214 message = Message(
215 conversation=conversation,
216 author_id=surfer.id,
217 text=random_hex(64),
218 message_type=MessageType.text,
219 )
221 host_request = HostRequest(
222 conversation=conversation,
223 surfer=surfer,
224 host=host,
225 from_date=from_date,
226 to_date=to_date,
227 status=HostRequestStatus.pending,
228 surfer_last_seen_message_id=message.id,
229 )
231 session.add(host_request)
233 with patch("couchers.email.queue_email") as mock:
234 send_new_host_request_email(host_request)
236 assert mock.call_count == 1
237 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
238 assert recipient == host.email
239 assert "host request" in subject.lower()
240 assert host.name in plain
241 assert host.name in html
242 assert surfer.name in plain
243 assert surfer.name in html
244 assert from_date in plain
245 assert from_date in html
246 assert to_date in plain
247 assert to_date in html
248 assert surfer.avatar.thumbnail_url not in plain
249 assert surfer.avatar.thumbnail_url in html
250 assert f"{config['BASE_URL']}/messages/hosting/" in plain
251 assert f"{config['BASE_URL']}/messages/hosting/" in html
254def test_friend_request_email(db):
255 with session_scope() as session:
256 to_user, api_token_to = generate_user()
257 # little trick here to get the upload correctly without invalidating users
258 key = random_hex(32)
259 filename = random_hex(32) + ".jpg"
260 session.add(
261 Upload(
262 key=key,
263 filename=filename,
264 creator_user_id=to_user.id,
265 )
266 )
267 session.commit()
268 from_user, api_token_from = generate_user(avatar_key=key)
269 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.pending)
270 session.add(friend_relationship)
272 with patch("couchers.email.queue_email") as mock:
273 send_friend_request_email(friend_relationship)
275 assert mock.call_count == 1
276 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
277 assert recipient == to_user.email
278 assert "friend" in subject.lower()
279 assert to_user.name in plain
280 assert to_user.name in html
281 assert from_user.name in subject
282 assert from_user.name in plain
283 assert from_user.name in html
284 assert from_user.avatar.thumbnail_url not in plain
285 assert from_user.avatar.thumbnail_url in html
286 assert f"{config['BASE_URL']}/connections/friends/" in plain
287 assert f"{config['BASE_URL']}/connections/friends/" in html
290def test_friend_request_accepted_email(db):
291 with session_scope() as session:
292 from_user, api_token_from = generate_user()
293 to_user, api_token_to = generate_user()
294 key = random_hex(32)
295 filename = random_hex(32) + ".jpg"
296 session.add(
297 Upload(
298 key=key,
299 filename=filename,
300 creator_user_id=from_user.id,
301 )
302 )
303 session.commit()
304 to_user, api_token_to = generate_user(avatar_key=key)
305 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
306 session.add(friend_relationship)
308 with patch("couchers.email.queue_email") as mock:
309 send_friend_request_accepted_email(friend_relationship)
311 assert mock.call_count == 1
312 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
313 assert recipient == from_user.email
314 assert "friend" in subject.lower()
315 assert from_user.name in plain
316 assert from_user.name in html
317 assert to_user.name in subject
318 assert to_user.name in plain
319 assert to_user.name in html
320 assert to_user.avatar.thumbnail_url not in plain
321 assert to_user.avatar.thumbnail_url in html
322 assert f"{config['BASE_URL']}/user/{to_user.username}" in plain
323 assert f"{config['BASE_URL']}/user/{to_user.username}" in html
326def test_email_patching_fails(db):
327 """
328 There was a problem where the mocking wasn't happening and the email dev
329 printing function was called instead, this makes sure the patching is
330 actually done
331 """
332 with session_scope() as session:
333 from_user, api_token_from = generate_user()
334 to_user, api_token_to = generate_user()
335 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.pending)
336 session.add(friend_relationship)
338 patched_msg = random_hex(64)
340 def mock_queue_email(sender_name, sender_email, recipient, subject, plain, html):
341 raise Exception(patched_msg)
343 with pytest.raises(Exception) as e:
344 with patch("couchers.email.queue_email", mock_queue_email):
345 send_friend_request_email(friend_relationship)
346 assert str(e.value) == patched_msg
349def test_email_changed_notification_email(db):
350 user, token = generate_user()
351 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
352 with patch("couchers.email.queue_email") as mock:
353 send_email_changed_notification_email(user)
355 assert mock.call_count == 1
356 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
357 assert "change requested" in subject
358 assert recipient == user.email
359 assert user.name in plain
360 assert user.name in html
361 assert user.new_email in plain
362 assert user.new_email in html
363 assert "A confirmation was sent to that email address" in plain
364 assert "A confirmation was sent to that email address" in html
365 assert "support@couchers.org" in plain
366 assert "support@couchers.org" in html
369def test_email_changed_confirmation_sent_to_old_email(db):
370 confirmation_token = urlsafe_secure_token()
371 user, user_token = generate_user()
372 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
373 user.old_email_token = confirmation_token
374 with patch("couchers.email.queue_email") as mock:
375 send_email_changed_confirmation_to_old_email(user)
377 assert mock.call_count == 1
378 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
379 assert "new email" in subject
380 assert recipient == user.email
381 assert user.name in plain
382 assert user.name in html
383 assert user.new_email in plain
384 assert user.new_email in html
385 assert "via a similar email sent to your new email address" in plain
386 assert "via a similar email sent to your new email address" in html
387 assert f"{config['BASE_URL']}/confirm-email?token={confirmation_token}" in plain
388 assert f"{config['BASE_URL']}/confirm-email?token={confirmation_token}" in html
389 assert "support@couchers.org" in plain
390 assert "support@couchers.org" in html
393def test_email_changed_confirmation_sent_to_new_email(db):
394 confirmation_token = urlsafe_secure_token()
395 user, user_token = generate_user()
396 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
397 user.new_email_token = confirmation_token
398 with patch("couchers.email.queue_email") as mock:
399 send_email_changed_confirmation_to_new_email(user)
401 assert mock.call_count == 1
402 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
403 assert "new email" in subject
404 assert recipient == user.new_email
405 assert user.name in plain
406 assert user.name in html
407 assert user.email in plain
408 assert user.email in html
409 assert "via a similar email sent to your old email address" in plain
410 assert "via a similar email sent to your old email address" in html
411 assert f"{config['BASE_URL']}/confirm-email?token={confirmation_token}" in plain
412 assert f"{config['BASE_URL']}/confirm-email?token={confirmation_token}" in html
413 assert "support@couchers.org" in plain
414 assert "support@couchers.org" in html
417def test_password_reset_email(db):
418 user, api_token = generate_user()
420 with session_scope() as session:
421 with patch("couchers.email.queue_email") as mock:
422 password_reset_token = send_password_reset_email(session, user)
424 assert mock.call_count == 1
425 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
426 assert recipient == user.email
427 assert "reset" in subject.lower()
428 assert password_reset_token.token in plain
429 assert password_reset_token.token in html
430 unique_string = "You asked for your password to be reset on Couchers.org."
431 assert unique_string in plain
432 assert unique_string in html
433 assert f"{config['BASE_URL']}/complete-password-reset?token={password_reset_token.token}" in plain
434 assert f"{config['BASE_URL']}/complete-password-reset?token={password_reset_token.token}" in html
435 assert "support@couchers.org" in plain
436 assert "support@couchers.org" in html
439def test_account_deletion_confirmation_email(db):
440 user, api_token = generate_user()
442 with patch("couchers.email.queue_email") as mock:
443 account_deletion_token = send_account_deletion_confirmation_email(user)
444 assert mock.call_count == 1
445 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
446 assert recipient == user.email
447 assert "account deletion" in subject.lower()
448 assert account_deletion_token.token in plain
449 assert account_deletion_token.token in html
450 unique_string = "You requested that we delete your account from Couchers.org."
451 assert unique_string in plain
452 assert unique_string in html
453 url = f"{config['BASE_URL']}/delete-account?token={account_deletion_token.token}"
454 assert url in plain
455 assert url in html
456 assert "support@couchers.org" in plain
457 assert "support@couchers.org" in html
460def test_api_key_email(db):
461 user, api_token = generate_user()
463 token = random_hex(64)
464 expiry = now() + timedelta(days=90)
466 with session_scope() as session:
467 with patch("couchers.email.queue_email") as mock:
468 send_api_key_email(session, user, token, expiry)
470 assert mock.call_count == 1
471 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
472 assert recipient == user.email
473 assert "api key" in subject.lower()
474 assert token in plain
475 assert token in html
476 assert str(expiry) in plain
477 assert str(expiry) in html
478 unique_string = "We've issued you with the following API key:"
479 assert unique_string in plain
480 assert unique_string in html
481 assert "support@couchers.org" in plain
482 assert "support@couchers.org" in html
485def test_account_deletion_successful_email(db):
486 user, api_token = generate_user()
488 with session_scope() as session:
489 user_ = session.execute(select(User)).scalar_one()
490 user.undelete_token = "token"
491 user.undelete_until = now()
492 user.is_deleted = True
494 with patch("couchers.email.queue_email") as mock:
495 send_account_deletion_successful_email(user, 7)
497 assert mock.call_count == 1
498 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
499 assert recipient == user.email
500 assert "account has been deleted" in subject.lower()
501 unique_string = "You have successfully deleted your account from Couchers.org."
502 assert unique_string in plain
503 assert unique_string in html
504 assert "7 days" in plain
505 assert "7 days" in html
506 url = f"{config['BASE_URL']}/recover-account?token={user.undelete_token}"
507 assert url in plain
508 assert url in html
509 assert "support@couchers.org" in plain
510 assert "support@couchers.org" in html
513def test_account_recovery_successful_email(db):
514 user, api_token = generate_user()
516 with patch("couchers.email.queue_email") as mock:
517 send_account_recovered_email(user)
519 assert mock.call_count == 1
520 (sender_name, sender_email, recipient, subject, plain, html), _ = mock.call_args
521 assert recipient == user.email
522 assert "account has been recovered" in subject.lower()
523 unique_string = "You have successfully recovered your account on Couchers.org!"
524 assert unique_string in plain
525 assert unique_string in html
526 assert "support@couchers.org" in plain
527 assert "support@couchers.org" in html
530def test_email_prefix_config(db, monkeypatch):
531 user, token = generate_user()
532 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
534 with patch("couchers.email.queue_email") as mock:
535 send_email_changed_notification_email(user)
537 assert mock.call_count == 1
538 (sender_name, sender_email, _, subject, _, _), _ = mock.call_args
540 assert sender_name == "Couchers.org"
541 assert sender_email == "notify@couchers.org.invalid"
542 assert subject == "[TEST] Couchers.org email change requested"
544 new_config = config.copy()
545 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
546 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
547 new_config["NOTIFICATION_EMAIL_PREFIX"] = ""
549 monkeypatch.setattr(couchers.email, "config", new_config)
551 with patch("couchers.email.queue_email") as mock:
552 send_email_changed_notification_email(user)
554 assert mock.call_count == 1
555 (sender_name, sender_email, _, subject, _, _), _ = mock.call_args
557 assert sender_name == "TestCo"
558 assert sender_email == "testco@testing.co.invalid"
559 assert subject == "Couchers.org email change requested"