Coverage for src/tests/test_email.py: 100%
236 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-06 23:17 +0000
1from unittest.mock import patch
3import pytest
4from sqlalchemy import update
6import couchers.email
7import couchers.jobs.handlers
8from couchers.config import config
9from couchers.crypto import random_hex, urlsafe_secure_token
10from couchers.db import session_scope
11from couchers.models import (
12 ContentReport,
13 Email,
14 FriendRelationship,
15 FriendStatus,
16 Reference,
17 ReferenceType,
18 SignupFlow,
19 User,
20)
21from couchers.notifications.notify import notify
22from couchers.proto import api_pb2, editor_pb2, events_pb2, notification_data_pb2, notifications_pb2
23from couchers.sql import couchers_select as select
24from couchers.tasks import (
25 enforce_community_memberships,
26 maybe_send_reference_report_email,
27 send_content_report_email,
28 send_email_changed_confirmation_to_new_email,
29 send_signup_email,
30)
31from couchers.utils import Timestamp_from_datetime, now, timedelta
32from tests.test_communities import create_community
33from tests.test_fixtures import ( # noqa
34 api_session,
35 db,
36 email_fields,
37 events_session,
38 generate_user,
39 mock_notification_email,
40 notifications_session,
41 process_jobs,
42 push_collector,
43 real_admin_session,
44 real_editor_session,
45 session_scope,
46 testconfig,
47)
50@pytest.fixture(autouse=True)
51def _(testconfig):
52 pass
55def test_signup_verification_email(db):
56 request_email = f"{random_hex(12)}@couchers.org.invalid"
58 flow = SignupFlow(name="Frodo", email=request_email)
60 with session_scope() as session:
61 with mock_notification_email() as mock:
62 send_signup_email(session, flow)
64 assert mock.call_count == 1
65 e = email_fields(mock)
66 assert e.recipient == request_email
67 assert flow.email_token in e.plain
68 assert flow.email_token in e.html
71def test_report_email(db):
72 user_reporter, api_token_author = generate_user()
73 user_author, api_token_reported = generate_user()
75 report = ContentReport(
76 reporting_user=user_reporter,
77 reason="spam",
78 description="I think this is spam and does not belong on couchers",
79 content_ref="comment/123",
80 author_user=user_author,
81 user_agent="n/a",
82 page="https://couchers.org/comment/123",
83 )
85 with session_scope() as session:
86 with mock_notification_email() as mock:
87 send_content_report_email(session, report)
89 assert mock.call_count == 1
91 e = email_fields(mock)
92 assert e.recipient == "reports@couchers.org.invalid"
93 assert report.author_user.username in e.plain
94 assert str(report.author_user.id) in e.plain
95 assert report.author_user.email in e.plain
96 assert report.reporting_user.username in e.plain
97 assert str(report.reporting_user.id) in e.plain
98 assert report.reporting_user.email in e.plain
99 assert report.reason in e.plain
100 assert report.description in e.plain
101 assert "report" in e.subject.lower()
104def test_reference_report_email_not_sent(db):
105 with session_scope() as session:
106 from_user, api_token_author = generate_user()
107 to_user, api_token_reported = generate_user()
109 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
110 session.add(friend_relationship)
111 session.flush()
113 reference = Reference(
114 from_user=from_user,
115 to_user=to_user,
116 reference_type=ReferenceType.friend,
117 text="This person was very nice to me.",
118 rating=0.9,
119 was_appropriate=True,
120 )
122 # no email sent for a positive ref
124 with mock_notification_email() as mock:
125 maybe_send_reference_report_email(session, reference)
127 assert mock.call_count == 0
130def test_reference_report_email(db):
131 with session_scope() as session:
132 from_user, api_token_author = generate_user()
133 to_user, api_token_reported = generate_user()
135 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
136 session.add(friend_relationship)
137 session.flush()
139 reference = Reference(
140 from_user=from_user,
141 to_user=to_user,
142 reference_type=ReferenceType.friend,
143 text="This person was not nice to me.",
144 rating=0.3,
145 was_appropriate=False,
146 private_text="This is some private text for support",
147 )
149 with mock_notification_email() as mock:
150 maybe_send_reference_report_email(session, reference)
152 assert mock.call_count == 1
153 e = email_fields(mock)
154 assert e.recipient == "reports@couchers.org.invalid"
155 assert "report" in e.subject.lower()
156 assert "reference" in e.subject.lower()
157 assert reference.from_user.username in e.plain
158 assert str(reference.from_user.id) in e.plain
159 assert reference.from_user.email in e.plain
160 assert reference.to_user.username in e.plain
161 assert str(reference.to_user.id) in e.plain
162 assert reference.to_user.email in e.plain
163 assert reference.text in e.plain
164 assert "friend" in e.plain.lower()
165 assert reference.private_text in e.plain
168def test_email_patching_fails(db):
169 """
170 There was a problem where the mocking wasn't happening and the email dev
171 printing function was called instead, this makes sure the patching is
172 actually done
173 """
174 to_user, to_token = generate_user()
175 from_user, from_token = generate_user()
177 patched_msg = random_hex(64)
179 def mock_queue_email(session, **kwargs):
180 raise Exception(patched_msg)
182 with patch("couchers.email._queue_email", mock_queue_email):
183 with pytest.raises(Exception) as e:
184 with api_session(from_token) as api:
185 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
186 process_jobs()
188 assert str(e.value) == patched_msg
191def test_email_changed_confirmation_sent_to_new_email(db):
192 confirmation_token = urlsafe_secure_token()
193 user, user_token = generate_user()
194 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
195 user.new_email_token = confirmation_token
196 with session_scope() as session:
197 with mock_notification_email() as mock:
198 send_email_changed_confirmation_to_new_email(session, user)
200 assert mock.call_count == 1
201 e = email_fields(mock)
202 assert "new email" in e.subject
203 assert e.recipient == user.new_email
204 assert user.name in e.plain
205 assert user.name in e.html
206 assert user.email in e.plain
207 assert user.email in e.html
208 assert "Your old email address is" in e.plain
209 assert "Your old email address is" in e.html
210 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
211 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
212 assert "support@couchers.org" in e.plain
213 assert "support@couchers.org" in e.html
216def test_do_not_email_security(db):
217 user, token = generate_user()
219 password_reset_token = urlsafe_secure_token()
221 with notifications_session(token) as notifications:
222 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
224 # make sure we still get security emails
226 with mock_notification_email() as mock:
227 with session_scope() as session:
228 notify(
229 session,
230 user_id=user.id,
231 topic_action="password_reset:start",
232 data=notification_data_pb2.PasswordResetStart(
233 password_reset_token=password_reset_token,
234 ),
235 )
237 assert mock.call_count == 1
238 e = email_fields(mock)
239 assert e.recipient == user.email
240 assert "reset" in e.subject.lower()
241 assert password_reset_token in e.plain
242 assert password_reset_token in e.html
243 unique_string = "You asked for your password to be reset on Couchers.org."
244 assert unique_string in e.plain
245 assert unique_string in e.html
246 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
247 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
248 assert "support@couchers.org" in e.plain
249 assert "support@couchers.org" in e.html
251 assert "/quick-link?payload=" not in e.plain
252 assert "/quick-link?payload=" not in e.html
255def test_do_not_email_non_security(db):
256 user, token1 = generate_user(complete_profile=True)
257 _, token2 = generate_user(complete_profile=True)
259 with notifications_session(token1) as notifications:
260 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
262 with mock_notification_email() as mock:
263 with api_session(token2) as api:
264 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
266 assert mock.call_count == 0
269def test_do_not_email_non_security_unsublink(db):
270 user, _ = generate_user(complete_profile=True)
271 _, token2 = generate_user(complete_profile=True)
273 with mock_notification_email() as mock:
274 with api_session(token2) as api:
275 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
277 assert mock.call_count == 1
278 e = email_fields(mock)
280 assert "/quick-link?payload=" in e.plain
281 assert "/quick-link?payload=" in e.html
284def test_email_prefix_config(db, monkeypatch):
285 user, _ = generate_user()
287 with mock_notification_email() as mock:
288 with session_scope() as session:
289 notify(
290 session,
291 user_id=user.id,
292 topic_action="donation:received",
293 data=notification_data_pb2.DonationReceived(
294 amount=20,
295 receipt_url="https://example.com/receipt/12345",
296 ),
297 )
299 assert mock.call_count == 1
300 e = email_fields(mock)
301 assert e.sender_name == "Couchers.org"
302 assert e.sender_email == "notify@couchers.org.invalid"
303 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
305 new_config = config.copy()
306 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
307 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
308 new_config["NOTIFICATION_PREFIX"] = ""
310 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
312 with mock_notification_email() as mock:
313 with session_scope() as session:
314 notify(
315 session,
316 user_id=user.id,
317 topic_action="donation:received",
318 data=notification_data_pb2.DonationReceived(
319 amount=20,
320 receipt_url="https://example.com/receipt/12345",
321 ),
322 )
324 assert mock.call_count == 1
325 e = email_fields(mock)
326 assert e.sender_name == "TestCo"
327 assert e.sender_email == "testco@testing.co.invalid"
328 assert e.subject == "Thank you for your donation to Couchers.org!"
331def test_send_donation_email(db, monkeypatch):
332 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
334 new_config = config.copy()
335 new_config["ENABLE_EMAIL"] = True
337 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
339 with session_scope() as session:
340 notify(
341 session,
342 user_id=user.id,
343 topic_action="donation:received",
344 data=notification_data_pb2.DonationReceived(
345 amount=20,
346 receipt_url="https://example.com/receipt/12345",
347 ),
348 )
350 with patch("couchers.email.smtp.smtplib.SMTP"):
351 process_jobs()
353 with session_scope() as session:
354 email = session.execute(select(Email)).scalar_one()
355 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
356 assert (
357 email.plain
358 == """Dear Testy von Test,
360Thank you so much for your donation of $20 to Couchers.org.
362Your 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.
365You can download an invoice and receipt for the donation here:
367<https://example.com/receipt/12345>
369If you have any questions about your donation, please email us at <donations@couchers.org>.
371Your generosity will help deliver the platform for everyone.
374Thank you!
376Aapeli and Itsi,
377Couchers.org Founders
380---
382This is a security email, you cannot unsubscribe from it."""
383 )
385 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
386 assert email.sender_name == "Couchers.org"
387 assert email.sender_email == "notify@couchers.org.invalid"
388 assert email.recipient == "testing@couchers.org.invalid"
389 assert "https://example.com/receipt/12345" in email.html
390 assert not email.list_unsubscribe_header
391 assert email.source_data == "testing_version/donation_received"
394def test_email_deleted_users_regression(db):
395 """
396 We introduced a bug in notify v2 where we would email deleted/banned users.
397 """
398 super_user, super_token = generate_user(is_superuser=True)
399 creating_user, creating_token = generate_user(complete_profile=True)
401 normal_user, _ = generate_user()
402 ban_user, _ = generate_user()
403 delete_user, _ = generate_user()
405 with session_scope() as session:
406 create_community(session, 10, 2, "Global Community", [super_user], [], None)
407 create_community(
408 session,
409 0,
410 2,
411 "Non-global Community",
412 [super_user],
413 [creating_user, normal_user, ban_user, delete_user],
414 None,
415 )
417 enforce_community_memberships()
419 start_time = now() + timedelta(hours=2)
420 end_time = start_time + timedelta(hours=3)
421 with events_session(creating_token) as api:
422 res = api.CreateEvent(
423 events_pb2.CreateEventReq(
424 title="Dummy Title",
425 content="Dummy content.",
426 photo_key=None,
427 offline_information=events_pb2.OfflineEventInformation(
428 address="Near Null Island",
429 lat=0.1,
430 lng=0.2,
431 ),
432 start_time=Timestamp_from_datetime(start_time),
433 end_time=Timestamp_from_datetime(end_time),
434 timezone="UTC",
435 )
436 )
437 event_id = res.event_id
438 assert not res.is_deleted
440 with mock_notification_email() as mock:
441 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
442 assert mock.call_count == 1
444 with real_editor_session(super_token) as editor:
445 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
446 assert len(res.requests) == 1
447 # this will count everyone
448 assert res.requests[0].approx_users_to_notify == 5
450 with session_scope() as session:
451 session.execute(update(User).where(User.id == ban_user.id).values(is_banned=True))
452 session.execute(update(User).where(User.id == delete_user.id).values(is_deleted=True))
454 with real_editor_session(super_token) as editor:
455 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
456 assert len(res.requests) == 1
457 # should only notify creating_user, super_user and normal_user
458 assert res.requests[0].approx_users_to_notify == 3
460 with mock_notification_email() as mock:
461 editor.DecideEventCommunityInviteRequest(
462 editor_pb2.DecideEventCommunityInviteRequestReq(
463 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
464 approve=True,
465 )
466 )
468 assert mock.call_count == 3