Coverage for app / backend / src / tests / test_email.py: 100%
270 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 14:14 +0000
1from datetime import timedelta
2from unittest.mock import patch
4import pytest
5from sqlalchemy import func, select, update
7import couchers.email
8import couchers.jobs.handlers
9from couchers.config import config
10from couchers.crypto import random_hex, urlsafe_secure_token
11from couchers.db import session_scope
12from couchers.models import (
13 ContentReport,
14 Email,
15 Reference,
16 ReferenceType,
17 SignupFlow,
18 User,
19)
20from couchers.models.notifications import NotificationTopicAction
21from couchers.notifications.notify import notify
22from couchers.proto import api_pb2, editor_pb2, events_pb2, notification_data_pb2, notifications_pb2
23from couchers.tasks import (
24 enforce_community_memberships,
25 maybe_send_reference_report_email,
26 send_content_report_email,
27 send_email_changed_confirmation_to_new_email,
28 send_signup_email,
29)
30from couchers.utils import Timestamp_from_datetime, now
31from tests.fixtures.db import generate_user, get_friend_relationship, make_friends
32from tests.fixtures.misc import Moderator, email_fields, mock_notification_email, process_jobs
33from tests.fixtures.sessions import api_session, events_session, notifications_session, real_editor_session
34from tests.test_communities import create_community
37@pytest.fixture(autouse=True)
38def _(testconfig):
39 pass
42def test_signup_verification_email(db):
43 request_email = f"{random_hex(12)}@couchers.org.invalid"
45 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
47 with session_scope() as session:
48 with mock_notification_email() as mock:
49 send_signup_email(session, flow)
51 assert mock.call_count == 1
52 e = email_fields(mock)
53 assert e.recipient == request_email
54 assert flow.email_token
55 assert flow.email_token in e.html
56 assert flow.email_token in e.html
59def test_report_email(db):
60 user_reporter, api_token_author = generate_user()
61 user_author, api_token_reported = generate_user()
63 with session_scope() as session:
64 report = ContentReport(
65 reporting_user_id=user_reporter.id,
66 reason="spam",
67 description="I think this is spam and does not belong on couchers",
68 content_ref="comment/123",
69 author_user_id=user_author.id,
70 user_agent="n/a",
71 page="https://couchers.org/comment/123",
72 )
73 session.add(report)
74 session.flush()
76 with mock_notification_email() as mock:
77 send_content_report_email(session, report)
79 # Load all data before session closes
80 author_username = report.author_user.username
81 author_id = report.author_user.id
82 author_email = report.author_user.email
83 reporting_username = report.reporting_user.username
84 reporting_id = report.reporting_user.id
85 reporting_email = report.reporting_user.email
86 reason = report.reason
87 description = report.description
89 assert mock.call_count == 1
91 e = email_fields(mock)
92 assert e.recipient == "reports@couchers.org.invalid"
93 assert author_username in e.plain
94 assert str(author_id) in e.plain
95 assert author_email in e.plain
96 assert reporting_username in e.plain
97 assert str(reporting_id) in e.plain
98 assert reporting_email in e.plain
99 assert reason in e.plain
100 assert description in e.plain
101 assert "report" in e.subject.lower()
104def test_reference_report_email_not_sent(db):
105 from_user, api_token_author = generate_user()
106 to_user, api_token_reported = generate_user()
108 make_friends(from_user, to_user)
110 with session_scope() as session:
111 reference = Reference(
112 from_user_id=from_user.id,
113 to_user_id=to_user.id,
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 from_user, api_token_author = generate_user()
130 to_user, api_token_reported = generate_user()
132 make_friends(from_user, to_user)
134 with session_scope() as session:
135 reference = Reference(
136 from_user_id=from_user.id,
137 to_user_id=to_user.id,
138 reference_type=ReferenceType.friend,
139 text="This person was not nice to me.",
140 rating=0.3,
141 was_appropriate=False,
142 private_text="This is some private text for support",
143 )
144 session.add(reference)
145 session.flush()
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 from_user.username in e.plain
156 assert str(from_user.id) in e.plain
157 assert from_user.email in e.plain
158 assert to_user.username in e.plain
159 assert str(to_user.id) in e.plain
160 assert 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
164 assert reference.private_text in e.plain
167def test_email_patching_fails(db):
168 """
169 There was a problem where the mocking wasn't happening and the email dev
170 printing function was called instead, this makes sure the patching is
171 actually done
172 """
173 to_user, to_token = generate_user()
174 from_user, from_token = generate_user()
175 # Need a moderator to approve the friend request since UMS defers notification
176 mod_user, mod_token = generate_user(is_superuser=True)
177 moderator = Moderator(mod_user, mod_token)
179 patched_msg = random_hex(64)
181 def mock_queue_email(session, **kwargs):
182 raise Exception(patched_msg)
184 with api_session(from_token) as api:
185 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
187 friend_relationship = get_friend_relationship(from_user, to_user)
188 assert friend_relationship is not None
189 moderator.approve_friend_request(friend_relationship.id)
191 with patch("couchers.email.queuing._queue_email", mock_queue_email):
192 with pytest.raises(Exception) as e:
193 process_jobs()
195 assert str(e.value) == patched_msg
198def test_email_changed_confirmation_sent_to_new_email(db):
199 confirmation_token = urlsafe_secure_token()
200 user, user_token = generate_user()
201 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
202 user.new_email_token = confirmation_token
203 with session_scope() as session:
204 with mock_notification_email() as mock:
205 send_email_changed_confirmation_to_new_email(session, user)
207 assert mock.call_count == 1
208 e = email_fields(mock)
209 assert "new email" in e.subject
210 assert e.recipient == user.new_email
211 assert user.name in e.plain
212 assert user.name in e.html
213 assert user.email in e.plain
214 assert user.email in e.html
215 assert "Your old email address is" in e.plain
216 assert "Your old email address is" in e.html
217 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
218 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
219 assert "support@couchers.org" in e.plain
220 assert "support@couchers.org" in e.html
223def test_do_not_email_security(db):
224 user, token = generate_user()
226 password_reset_token = urlsafe_secure_token()
228 with notifications_session(token) as notifications:
229 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
231 # make sure we still get security emails
233 with mock_notification_email() as mock:
234 with session_scope() as session:
235 notify(
236 session,
237 user_id=user.id,
238 topic_action=NotificationTopicAction.password_reset__start,
239 key="",
240 data=notification_data_pb2.PasswordResetStart(
241 password_reset_token=password_reset_token,
242 ),
243 )
245 assert mock.call_count == 1
246 e = email_fields(mock)
247 assert e.recipient == user.email
248 assert "reset" in e.subject.lower()
249 assert password_reset_token in e.plain
250 assert password_reset_token in e.html
251 unique_string = "You asked for your password to be reset on Couchers.org."
252 assert unique_string in e.plain
253 assert unique_string in e.html
254 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
255 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
256 assert "support@couchers.org" in e.plain
257 assert "support@couchers.org" in e.html
259 assert "/quick-link?payload=" not in e.plain
260 assert "/quick-link?payload=" not in e.html
263def test_do_not_email_non_security(db):
264 user, token1 = generate_user(complete_profile=True)
265 from_user, token2 = generate_user(complete_profile=True)
266 # Need a moderator to approve the friend request since UMS defers notification
267 mod_user, mod_token = generate_user(is_superuser=True)
268 moderator = Moderator(mod_user, mod_token)
270 with notifications_session(token1) as notifications:
271 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
273 with api_session(token2) as api:
274 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
276 friend_relationship = get_friend_relationship(from_user, user)
277 assert friend_relationship is not None
278 moderator.approve_friend_request(friend_relationship.id)
280 with mock_notification_email() as mock:
281 process_jobs()
283 assert mock.call_count == 0
286def test_do_not_email_non_security_unsublink(db):
287 user, _ = generate_user(complete_profile=True)
288 from_user, token2 = generate_user(complete_profile=True)
289 # Need a moderator to approve the friend request since UMS defers notification
290 mod_user, mod_token = generate_user(is_superuser=True)
291 moderator = Moderator(mod_user, mod_token)
293 with api_session(token2) as api:
294 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
296 friend_relationship = get_friend_relationship(from_user, user)
297 assert friend_relationship is not None
298 moderator.approve_friend_request(friend_relationship.id)
300 with mock_notification_email() as mock:
301 process_jobs()
303 assert mock.call_count == 1
304 e = email_fields(mock)
306 assert "/quick-link?payload=" in e.plain
307 assert "/quick-link?payload=" in e.html
310def test_email_prefix_config(db, monkeypatch):
311 user, _ = generate_user()
313 with mock_notification_email() as mock:
314 with session_scope() as session:
315 notify(
316 session,
317 user_id=user.id,
318 topic_action=NotificationTopicAction.donation__received,
319 key="",
320 data=notification_data_pb2.DonationReceived(
321 amount=20,
322 receipt_url="https://example.com/receipt/12345",
323 ),
324 )
326 assert mock.call_count == 1
327 e = email_fields(mock)
328 assert e.sender_name == "Couchers.org"
329 assert e.sender_email == "notify@couchers.org.invalid"
330 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
332 new_config = config.copy()
333 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
334 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
335 new_config["NOTIFICATION_PREFIX"] = ""
337 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
339 with mock_notification_email() as mock:
340 with session_scope() as session:
341 notify(
342 session,
343 user_id=user.id,
344 topic_action=NotificationTopicAction.donation__received,
345 key="",
346 data=notification_data_pb2.DonationReceived(
347 amount=20,
348 receipt_url="https://example.com/receipt/12345",
349 ),
350 )
352 assert mock.call_count == 1
353 e = email_fields(mock)
354 assert e.sender_name == "TestCo"
355 assert e.sender_email == "testco@testing.co.invalid"
356 assert e.subject == "Thank you for your donation to Couchers.org!"
359def test_send_donation_email(db, monkeypatch):
360 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
362 new_config = config.copy()
363 new_config["ENABLE_EMAIL"] = True
365 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
367 with session_scope() as session:
368 notify(
369 session,
370 user_id=user.id,
371 topic_action=NotificationTopicAction.donation__received,
372 key="",
373 data=notification_data_pb2.DonationReceived(
374 amount=20,
375 receipt_url="https://example.com/receipt/12345",
376 ),
377 )
379 with patch("couchers.email.smtp.smtplib.SMTP"):
380 process_jobs()
382 with session_scope() as session:
383 email = session.execute(select(Email)).scalar_one()
384 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
385 assert (
386 email.plain
387 == """Dear Testy von Test,
389Thank you so much for your donation of $20 to Couchers.org.
391Your 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.
394You can download an invoice and receipt for the donation here:
396<https://example.com/receipt/12345>
398If you have any questions about your donation, please email us at <donations@couchers.org>.
400Your generosity will help deliver the platform for everyone.
403Thank you!
405Aapeli and Itsi,
406Couchers.org Founders
409---
411This is a security email, you cannot unsubscribe from it.
412"""
413 )
415 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
416 assert email.sender_name == "Couchers.org"
417 assert email.sender_email == "notify@couchers.org.invalid"
418 assert email.recipient == "testing@couchers.org.invalid"
419 assert "https://example.com/receipt/12345" in email.html
420 assert not email.list_unsubscribe_header
421 assert email.source_data == "testing_version/donation_received"
424def test_email_deleted_users_regression(db, moderator: Moderator):
425 """
426 We introduced a bug in notify v2 where we would email deleted/banned users.
427 """
428 super_user, super_token = generate_user(is_superuser=True)
429 creating_user, creating_token = generate_user(complete_profile=True)
431 normal_user, _ = generate_user()
432 ban_user, _ = generate_user()
433 delete_user, _ = generate_user()
435 with session_scope() as session:
436 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
437 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
438 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
439 c_id = create_community(
440 session,
441 0,
442 2,
443 "Non-global Community",
444 [super_user],
445 [creating_user, normal_user, ban_user, delete_user],
446 r,
447 ).id
449 enforce_community_memberships()
451 start_time = now() + timedelta(hours=2)
452 end_time = start_time + timedelta(hours=3)
453 with events_session(creating_token) as api:
454 res = api.CreateEvent(
455 events_pb2.CreateEventReq(
456 title="Dummy Title",
457 content="Dummy content.",
458 photo_key=None,
459 parent_community_id=c_id,
460 offline_information=events_pb2.OfflineEventInformation(
461 address="Near Null Island",
462 lat=0.1,
463 lng=0.2,
464 ),
465 start_time=Timestamp_from_datetime(start_time),
466 end_time=Timestamp_from_datetime(end_time),
467 timezone="UTC",
468 )
469 )
470 event_id = res.event_id
471 assert not res.is_deleted
473 moderator.approve_event_occurrence(event_id)
475 with events_session(creating_token) as api:
476 with mock_notification_email() as mock:
477 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
478 assert mock.call_count == 1
480 with real_editor_session(super_token) as editor:
481 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
482 assert len(res.requests) == 1
483 # this will count everyone
484 assert res.requests[0].approx_users_to_notify == 5
486 with session_scope() as session:
487 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
488 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
490 with real_editor_session(super_token) as editor:
491 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
492 assert len(res.requests) == 1
493 # should only notify creating_user, super_user and normal_user
494 assert res.requests[0].approx_users_to_notify == 3
496 with mock_notification_email() as mock:
497 editor.DecideEventCommunityInviteRequest(
498 editor_pb2.DecideEventCommunityInviteRequestReq(
499 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
500 approve=True,
501 )
502 )
504 assert mock.call_count == 3