Coverage for src/tests/test_email.py: 100%
235 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-06-01 15:07 +0000
1from unittest.mock import patch
3import pytest
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)
48@pytest.fixture(autouse=True)
49def _(testconfig):
50 pass
53def test_signup_verification_email(db):
54 request_email = f"{random_hex(12)}@couchers.org.invalid"
56 flow = SignupFlow(name="Frodo", email=request_email)
58 with session_scope() as session:
59 with mock_notification_email() as mock:
60 send_signup_email(session, flow)
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
69def test_report_email(db):
70 user_reporter, api_token_author = generate_user()
71 user_author, api_token_reported = generate_user()
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 )
83 with session_scope() as session:
84 with mock_notification_email() as mock:
85 send_content_report_email(session, report)
87 assert mock.call_count == 1
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()
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()
107 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
108 session.add(friend_relationship)
109 session.flush()
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 )
120 # no email sent for a positive ref
122 with mock_notification_email() as mock:
123 maybe_send_reference_report_email(session, reference)
125 assert mock.call_count == 0
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()
133 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
134 session.add(friend_relationship)
135 session.flush()
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 private_text="This is some private text for support",
145 )
147 with mock_notification_email() as mock:
148 maybe_send_reference_report_email(session, reference)
150 assert mock.call_count == 1
151 e = email_fields(mock)
152 assert e.recipient == "reports@couchers.org.invalid"
153 assert "report" in e.subject.lower()
154 assert "reference" in e.subject.lower()
155 assert reference.from_user.username in e.plain
156 assert str(reference.from_user.id) in e.plain
157 assert reference.from_user.email in e.plain
158 assert reference.to_user.username in e.plain
159 assert str(reference.to_user.id) in e.plain
160 assert reference.to_user.email in e.plain
161 assert reference.text in e.plain
162 assert "friend" in e.plain.lower()
163 assert reference.private_text in e.plain
166def test_email_patching_fails(db):
167 """
168 There was a problem where the mocking wasn't happening and the email dev
169 printing function was called instead, this makes sure the patching is
170 actually done
171 """
172 to_user, to_token = generate_user()
173 from_user, from_token = generate_user()
175 patched_msg = random_hex(64)
177 def mock_queue_email(session, **kwargs):
178 raise Exception(patched_msg)
180 with patch("couchers.email._queue_email", mock_queue_email):
181 with pytest.raises(Exception) as e:
182 with api_session(from_token) as api:
183 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
184 process_jobs()
186 assert str(e.value) == patched_msg
189def test_email_changed_confirmation_sent_to_new_email(db):
190 confirmation_token = urlsafe_secure_token()
191 user, user_token = generate_user()
192 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
193 user.new_email_token = confirmation_token
194 with session_scope() as session:
195 with mock_notification_email() as mock:
196 send_email_changed_confirmation_to_new_email(session, user)
198 assert mock.call_count == 1
199 e = email_fields(mock)
200 assert "new email" in e.subject
201 assert e.recipient == user.new_email
202 assert user.name in e.plain
203 assert user.name in e.html
204 assert user.email in e.plain
205 assert user.email in e.html
206 assert "Your old email address is" in e.plain
207 assert "Your old email address is" in e.html
208 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
209 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
210 assert "support@couchers.org" in e.plain
211 assert "support@couchers.org" in e.html
214def test_do_not_email_security(db):
215 user, token = generate_user()
217 password_reset_token = urlsafe_secure_token()
219 with notifications_session(token) as notifications:
220 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
222 # make sure we still get security emails
224 with mock_notification_email() as mock:
225 with session_scope() as session:
226 notify(
227 session,
228 user_id=user.id,
229 topic_action="password_reset:start",
230 data=notification_data_pb2.PasswordResetStart(
231 password_reset_token=password_reset_token,
232 ),
233 )
235 assert mock.call_count == 1
236 e = email_fields(mock)
237 assert e.recipient == user.email
238 assert "reset" in e.subject.lower()
239 assert password_reset_token in e.plain
240 assert password_reset_token in e.html
241 unique_string = "You asked for your password to be reset on Couchers.org."
242 assert unique_string in e.plain
243 assert unique_string in e.html
244 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
245 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
246 assert "support@couchers.org" in e.plain
247 assert "support@couchers.org" in e.html
249 assert "/unsubscribe?payload=" not in e.plain
250 assert "/unsubscribe?payload=" not in e.html
253def test_do_not_email_non_security(db):
254 user, token1 = generate_user(complete_profile=True)
255 _, token2 = generate_user(complete_profile=True)
257 with notifications_session(token1) as notifications:
258 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
260 with mock_notification_email() as mock:
261 with api_session(token2) as api:
262 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
264 assert mock.call_count == 0
267def test_do_not_email_non_security_unsublink(db):
268 user, _ = generate_user(complete_profile=True)
269 _, token2 = generate_user(complete_profile=True)
271 with mock_notification_email() as mock:
272 with api_session(token2) as api:
273 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
275 assert mock.call_count == 1
276 e = email_fields(mock)
278 assert "/unsubscribe?payload=" in e.plain
279 assert "/unsubscribe?payload=" in e.html
282def test_email_prefix_config(db, monkeypatch):
283 user, _ = generate_user()
285 with mock_notification_email() as mock:
286 with session_scope() as session:
287 notify(
288 session,
289 user_id=user.id,
290 topic_action="donation:received",
291 data=notification_data_pb2.DonationReceived(
292 amount=20,
293 receipt_url="https://example.com/receipt/12345",
294 ),
295 )
297 assert mock.call_count == 1
298 e = email_fields(mock)
299 assert e.sender_name == "Couchers.org"
300 assert e.sender_email == "notify@couchers.org.invalid"
301 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
303 new_config = config.copy()
304 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
305 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
306 new_config["NOTIFICATION_PREFIX"] = ""
308 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
310 with mock_notification_email() as mock:
311 with session_scope() as session:
312 notify(
313 session,
314 user_id=user.id,
315 topic_action="donation:received",
316 data=notification_data_pb2.DonationReceived(
317 amount=20,
318 receipt_url="https://example.com/receipt/12345",
319 ),
320 )
322 assert mock.call_count == 1
323 e = email_fields(mock)
324 assert e.sender_name == "TestCo"
325 assert e.sender_email == "testco@testing.co.invalid"
326 assert e.subject == "Thank you for your donation to Couchers.org!"
329def test_send_donation_email(db, monkeypatch):
330 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
332 new_config = config.copy()
333 new_config["ENABLE_EMAIL"] = True
335 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
337 with session_scope() as session:
338 notify(
339 session,
340 user_id=user.id,
341 topic_action="donation:received",
342 data=notification_data_pb2.DonationReceived(
343 amount=20,
344 receipt_url="https://example.com/receipt/12345",
345 ),
346 )
348 with patch("couchers.email.smtp.smtplib.SMTP"):
349 process_jobs()
351 with session_scope() as session:
352 email = session.execute(select(Email)).scalar_one()
353 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
354 assert (
355 email.plain
356 == """Dear Testy von Test,
358Thank you so much for your donation of $20 to Couchers.org.
360Your 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.
363You can download an invoice and receipt for the donation here:
365<https://example.com/receipt/12345>
367If you have any questions about your donation, please email us at <donations@couchers.org>.
369Your generosity will help deliver the platform for everyone.
372Thank you!
374Aapeli and Itsi,
375Couchers.org Founders
378---
380This is a security email, you cannot unsubscribe from it."""
381 )
383 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
384 assert email.sender_name == "Couchers.org"
385 assert email.sender_email == "notify@couchers.org.invalid"
386 assert email.recipient == "testing@couchers.org.invalid"
387 assert "https://example.com/receipt/12345" in email.html
388 assert not email.list_unsubscribe_header
389 assert email.source_data == "testing_version/donation_received"
392def test_email_deleted_users_regression(db):
393 """
394 We introduced a bug in notify v2 where we would email deleted/banned users.
395 """
396 super_user, super_token = generate_user(is_superuser=True)
397 creating_user, creating_token = generate_user(complete_profile=True)
399 normal_user, _ = generate_user()
400 ban_user, _ = generate_user()
401 delete_user, _ = generate_user()
403 with session_scope() as session:
404 create_community(session, 10, 2, "Global Community", [super_user], [], None)
405 create_community(
406 session,
407 0,
408 2,
409 "Non-global Community",
410 [super_user],
411 [creating_user, normal_user, ban_user, delete_user],
412 None,
413 )
415 enforce_community_memberships()
417 start_time = now() + timedelta(hours=2)
418 end_time = start_time + timedelta(hours=3)
419 with events_session(creating_token) as api:
420 res = api.CreateEvent(
421 events_pb2.CreateEventReq(
422 title="Dummy Title",
423 content="Dummy content.",
424 photo_key=None,
425 offline_information=events_pb2.OfflineEventInformation(
426 address="Near Null Island",
427 lat=0.1,
428 lng=0.2,
429 ),
430 start_time=Timestamp_from_datetime(start_time),
431 end_time=Timestamp_from_datetime(end_time),
432 timezone="UTC",
433 )
434 )
435 event_id = res.event_id
436 assert not res.is_deleted
438 with mock_notification_email() as mock:
439 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
440 assert mock.call_count == 1
442 with real_admin_session(super_token) as admin:
443 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq())
444 assert len(res.requests) == 1
445 # this will count everyone
446 assert res.requests[0].approx_users_to_notify == 5
448 with session_scope() as session:
449 session.execute(select(User).where(User.id == ban_user.id)).scalar_one().is_banned = True
450 session.execute(select(User).where(User.id == delete_user.id)).scalar_one().is_deleted = True
452 with real_admin_session(super_token) as admin:
453 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq())
454 assert len(res.requests) == 1
455 # should only notify creating_user, super_user and normal_user
456 assert res.requests[0].approx_users_to_notify == 3
458 with mock_notification_email() as mock:
459 admin.DecideEventCommunityInviteRequest(
460 admin_pb2.DecideEventCommunityInviteRequestReq(
461 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
462 approve=True,
463 )
464 )
466 assert mock.call_count == 3