Coverage for app/backend/src/tests/test_email.py: 100%
273 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1from unittest.mock import patch
2from urllib.parse import parse_qs, urlparse
4import pytest
5from sqlalchemy import func, select, update
7from couchers.config import config
8from couchers.context import make_background_user_context, make_logged_out_context
9from couchers.crypto import b64decode, random_hex, urlsafe_secure_token
10from couchers.db import session_scope
11from couchers.i18n import LocalizationContext
12from couchers.models import (
13 ContentReport,
14 Email,
15 ModerationObjectType,
16 ModerationState,
17 ModerationVisibility,
18 Reference,
19 ReferenceType,
20 SignupFlow,
21 User,
22)
23from couchers.models.notifications import NotificationTopicAction
24from couchers.notifications.notify import notify
25from couchers.proto import api_pb2, auth_pb2, editor_pb2, notification_data_pb2, notifications_pb2
26from couchers.tasks import (
27 enforce_community_memberships,
28 maybe_send_reference_report_email,
29 send_content_report_email,
30 send_email_changed_confirmation_to_new_email,
31 send_signup_email,
32)
33from tests.fixtures.api import events as events_api
34from tests.fixtures.db import generate_user, get_friend_relationship, make_friends
35from tests.fixtures.misc import EmailCollector, Moderator, process_jobs
36from tests.fixtures.sessions import (
37 api_session,
38 auth_api_session,
39 notifications_session,
40 real_editor_session,
41)
42from tests.test_communities import create_community
45def test_signup_verification_email(db, email_collector: EmailCollector):
46 request_email = f"{random_hex(12)}@couchers.org.invalid"
48 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
50 with session_scope() as session:
51 context = make_logged_out_context(LocalizationContext.en_utc())
52 send_signup_email(context, session, flow)
54 email = email_collector.pop_for_recipient(request_email, last=True)
55 assert email.recipient == request_email
56 assert flow.email_token
57 assert flow.email_token in email.html
58 assert flow.email_token in email.html
61def test_signup_verification_email_with_token_ending_in_src(db, email_collector: EmailCollector):
62 """
63 Signup tokens are base64 encoded, so their "=" padding makes roughly one in 65k of them end in "src=".
64 The token goes into the button's `<a href="..." style="...">`, where it must not be mistaken for an image.
65 """
66 request_email = f"{random_hex(12)}@couchers.org.invalid"
67 token = "a-token-ending-in-src="
69 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
71 with session_scope() as session:
72 context = make_logged_out_context(LocalizationContext.en_utc())
73 with patch("couchers.tasks.urlsafe_secure_token", return_value=token):
74 send_signup_email(context, session, flow)
76 email = email_collector.pop_for_recipient(request_email, last=True)
77 assert flow.email_token == token
78 assert token in email.html
81def test_report_email(db, email_collector: EmailCollector):
82 user_reporter, api_token_author = generate_user()
83 user_author, api_token_reported = generate_user()
85 with session_scope() as session:
86 report = ContentReport(
87 reporting_user_id=user_reporter.id,
88 reason="spam",
89 description="I think this is spam and does not belong on couchers",
90 content_ref="comment/123",
91 author_user_id=user_author.id,
92 user_agent="n/a",
93 page="https://couchers.org/comment/123",
94 )
95 session.add(report)
96 session.flush()
98 send_content_report_email(session, report)
100 # Load all data before session closes
101 author_username = report.author_user.username
102 author_id = report.author_user.id
103 author_email = report.author_user.email
104 reporting_username = report.reporting_user.username
105 reporting_id = report.reporting_user.id
106 reporting_email = report.reporting_user.email
107 reason = report.reason
108 description = report.description
110 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
111 assert email.recipient == "reports@couchers.org.invalid"
112 assert author_username in email.plain
113 assert str(author_id) in email.plain
114 assert author_email in email.plain
115 assert reporting_username in email.plain
116 assert str(reporting_id) in email.plain
117 assert reporting_email in email.plain
118 assert reason in email.plain
119 assert description in email.plain
120 assert "report" in email.subject.lower()
123def test_reference_report_email_not_sent(db, email_collector: EmailCollector):
124 from_user, api_token_author = generate_user()
125 to_user, api_token_reported = generate_user()
127 make_friends(from_user, to_user)
129 with session_scope() as session:
130 moderation_state = ModerationState(
131 object_type=ModerationObjectType.reference,
132 object_id=0,
133 visibility=ModerationVisibility.visible,
134 )
135 session.add(moderation_state)
136 session.flush()
137 reference = Reference(
138 from_user_id=from_user.id,
139 to_user_id=to_user.id,
140 reference_type=ReferenceType.friend,
141 text="This person was very nice to me.",
142 rating=0.9,
143 was_appropriate=True,
144 moderation_state_id=moderation_state.id,
145 )
146 session.add(reference)
147 session.flush()
148 moderation_state.object_id = reference.id
150 # no email sent for a positive ref
151 maybe_send_reference_report_email(session, reference)
153 assert email_collector.count_for_recipient("reports@couchers.org.invalid") == 0
156def test_reference_report_email(db, email_collector: EmailCollector):
157 from_user, api_token_author = generate_user()
158 to_user, api_token_reported = generate_user()
160 make_friends(from_user, to_user)
162 with session_scope() as session:
163 moderation_state = ModerationState(
164 object_type=ModerationObjectType.reference,
165 object_id=0,
166 visibility=ModerationVisibility.visible,
167 )
168 session.add(moderation_state)
169 session.flush()
170 reference = Reference(
171 from_user_id=from_user.id,
172 to_user_id=to_user.id,
173 reference_type=ReferenceType.friend,
174 text="This person was not nice to me.",
175 rating=0.3,
176 was_appropriate=False,
177 private_text="This is some private text for support",
178 moderation_state_id=moderation_state.id,
179 )
180 session.add(reference)
181 session.flush()
182 moderation_state.object_id = reference.id
184 maybe_send_reference_report_email(session, reference)
186 reference_text = reference.text
187 reference_private_text = reference.private_text
189 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
190 assert email.recipient == "reports@couchers.org.invalid"
191 assert "report" in email.subject.lower()
192 assert "reference" in email.subject.lower()
193 assert from_user.username in email.plain
194 assert str(from_user.id) in email.plain
195 assert from_user.email in email.plain
196 assert to_user.username in email.plain
197 assert str(to_user.id) in email.plain
198 assert to_user.email in email.plain
199 assert reference_text in email.plain
200 assert "friend" in email.plain.lower()
201 assert reference_private_text
202 assert reference_private_text in email.plain
205def test_email_patching_fails(db):
206 """
207 There was a problem where the mocking wasn't happening and the email dev
208 printing function was called instead, this makes sure the patching is
209 actually done
210 """
211 to_user, to_token = generate_user()
212 from_user, from_token = generate_user()
213 # Need a moderator to approve the friend request since UMS defers notification
214 mod_user, mod_token = generate_user(is_superuser=True)
215 moderator = Moderator(mod_user, mod_token)
217 patched_msg = random_hex(64)
219 def mock_queue_email(session, payload):
220 raise Exception(patched_msg)
222 with api_session(from_token) as api:
223 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
225 friend_relationship = get_friend_relationship(from_user, to_user)
226 assert friend_relationship is not None
227 moderator.approve_friend_request(friend_relationship.id)
229 with patch("couchers.email.queuing._queue_email", mock_queue_email):
230 with pytest.raises(Exception) as e:
231 process_jobs()
233 assert str(e.value) == patched_msg
236def test_email_changed_confirmation_sent_to_new_email(db, email_collector: EmailCollector):
237 confirmation_token = urlsafe_secure_token()
238 user, user_token = generate_user()
239 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
240 user.new_email_token = confirmation_token
242 with session_scope() as session:
243 user_context = make_background_user_context(user.id)
244 send_email_changed_confirmation_to_new_email(user_context, session, user)
246 email = email_collector.pop_for_recipient(user.new_email, last=True)
247 assert "new email" in email.subject
248 assert email.recipient == user.new_email
249 assert user.name in email.plain
250 assert user.name in email.html
251 assert user.email in email.plain
252 assert user.email in email.html
253 assert "You requested that your email be changed from " in email.plain
254 assert "You requested that your email be changed from " in email.html
255 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.plain
256 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.html
257 assert "support@couchers.org" in email.plain
258 assert "support@couchers.org" in email.html
261def test_do_not_email_security(db, email_collector: EmailCollector):
262 user, token = generate_user()
264 password_reset_token = urlsafe_secure_token()
266 with notifications_session(token) as notifications:
267 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
269 # make sure we still get security emails
271 with session_scope() as session:
272 notify(
273 session,
274 user_id=user.id,
275 topic_action=NotificationTopicAction.password_reset__start,
276 key="",
277 data=notification_data_pb2.PasswordResetStart(
278 password_reset_token=password_reset_token,
279 ),
280 )
282 email = email_collector.pop_for_recipient(user.email, last=True)
283 assert email.recipient == user.email
284 assert "reset" in email.subject.lower()
285 assert password_reset_token in email.plain
286 assert password_reset_token in email.html
287 unique_string = "You asked for your password to be reset on Couchers.org."
288 assert unique_string in email.plain
289 assert unique_string in email.html
290 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.plain
291 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.html
292 assert "support@couchers.org" in email.plain
293 assert "support@couchers.org" in email.html
295 assert "/quick-link?payload=" not in email.plain
296 assert "/quick-link?payload=" not in email.html
299def test_do_not_email_non_security(db, email_collector: EmailCollector):
300 user, token1 = generate_user(complete_profile=True)
301 from_user, token2 = generate_user(complete_profile=True)
302 # Need a moderator to approve the friend request since UMS defers notification
303 mod_user, mod_token = generate_user(is_superuser=True)
304 moderator = Moderator(mod_user, mod_token)
306 with notifications_session(token1) as notifications:
307 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
309 with api_session(token2) as api:
310 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
312 friend_relationship = get_friend_relationship(from_user, user)
313 assert friend_relationship is not None
314 moderator.approve_friend_request(friend_relationship.id)
316 assert email_collector.count_for_recipient(user.email) == 0
319def test_do_not_email_non_security_unsublink(db, email_collector: EmailCollector):
320 user, _ = generate_user(complete_profile=True)
321 from_user, token2 = generate_user(complete_profile=True)
322 # Need a moderator to approve the friend request since UMS defers notification
323 mod_user, mod_token = generate_user(is_superuser=True)
324 moderator = Moderator(mod_user, mod_token)
326 with api_session(token2) as api:
327 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
329 friend_relationship = get_friend_relationship(from_user, user)
330 assert friend_relationship is not None
331 moderator.approve_friend_request(friend_relationship.id)
333 email = email_collector.pop_for_recipient(user.email, last=True)
335 assert "/quick-link?payload=" in email.plain
336 assert "/quick-link?payload=" in email.html
339def test_email_prefix_config(db, email_collector: EmailCollector):
340 user, _ = generate_user()
342 with session_scope() as session:
343 notify(
344 session,
345 user_id=user.id,
346 topic_action=NotificationTopicAction.donation__received,
347 key="",
348 data=notification_data_pb2.DonationReceived(
349 amount=20,
350 receipt_url="https://example.com/receipt/12345",
351 ),
352 )
354 email1 = email_collector.pop_for_recipient(user.email, last=True)
355 assert email1.sender_name == "Couchers.org"
356 assert email1.sender_email == "notify@couchers.org.invalid"
357 assert email1.subject == "[TEST] Thank you for your donation to Couchers.org!"
359 config.NOTIFICATION_EMAIL_SENDER = "TestCo"
360 config.NOTIFICATION_EMAIL_ADDRESS = "testco@testing.co.invalid"
361 config.NOTIFICATION_PREFIX = ""
363 with session_scope() as session:
364 notify(
365 session,
366 user_id=user.id,
367 topic_action=NotificationTopicAction.donation__received,
368 key="",
369 data=notification_data_pb2.DonationReceived(
370 amount=20,
371 receipt_url="https://example.com/receipt/12345",
372 ),
373 )
375 email2 = email_collector.pop_for_recipient(user.email, last=True)
376 assert email2.sender_name == "TestCo"
377 assert email2.sender_email == "testco@testing.co.invalid"
378 assert email2.subject == "Thank you for your donation to Couchers.org!"
381def test_send_donation_email(db):
382 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
384 config.ENABLE_EMAIL = True
386 with session_scope() as session:
387 notify(
388 session,
389 user_id=user.id,
390 topic_action=NotificationTopicAction.donation__received,
391 key="",
392 data=notification_data_pb2.DonationReceived(
393 amount=20,
394 receipt_url="https://example.com/receipt/12345",
395 ),
396 )
398 with patch("couchers.email.smtp.smtplib.SMTP"):
399 process_jobs()
401 with session_scope() as session:
402 email = session.execute(select(Email)).scalar_one()
403 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
404 assert (
405 email.plain
406 == """Hi Testy von Test,
408Thank you so much for your donation of $20 to Couchers.org.
410Your contribution will go towards building and sustaining the Couchers.org community, and is vital for our goal of a free and non-profit couch surfing platform.
412You can download an invoice and receipt for the donation here:
414Download invoice: https://example.com/receipt/12345
416Couchers, Inc. is a 501(c)(3) nonprofit (EIN: 87-1734577) registered in the United States. No goods or services were provided in exchange for this contribution.
418If you have any questions about your donation, please email us at donations@couchers.org.
420Thank you!
422Aapeli and Itsi,
423Couchers.org Founders
425---
427This is a security email, you cannot unsubscribe from it.
428"""
429 )
431 assert "Thank you so much for your donation of <b>$20</b> to Couchers.org." in email.html
432 assert email.sender_name == "Couchers.org"
433 assert email.sender_email == "notify@couchers.org.invalid"
434 assert email.recipient == "testing@couchers.org.invalid"
435 assert "https://example.com/receipt/12345" in email.html
436 assert not email.list_unsubscribe_header
437 assert email.source_data and ("donation:received" in email.source_data)
440def test_chat_missed_messages_list_unsubscribe_header(db, email_collector: EmailCollector):
441 """
442 Regression test: chat__missed_messages has key="" (it's a summary, not tied to a single chat).
443 The List-Unsubscribe header must use a topic_action unsubscribe link, not a topic_key link.
444 """
445 user, _ = generate_user()
447 with session_scope() as session:
448 notify(
449 session,
450 user_id=user.id,
451 topic_action=NotificationTopicAction.chat__missed_messages,
452 key="",
453 data=notification_data_pb2.ChatMissedMessages(
454 messages=[
455 notification_data_pb2.ChatMessage(
456 author=api_pb2.User(name="Test User", user_id=2, username="testuser"),
457 text="Hello!",
458 group_chat_id=99,
459 unseen_count=1,
460 ),
461 ],
462 ),
463 )
465 email = email_collector.pop_for_recipient(user.email, last=True)
467 assert email.list_unsubscribe_header
469 # Extract the List-Unsubscribe URL and call the Unsubscribe endpoint
470 url = email.list_unsubscribe_header.strip("<>")
471 url_parts = urlparse(url)
472 params = parse_qs(url_parts.query)
474 with auth_api_session() as (auth_api, metadata_interceptor):
475 res = auth_api.Unsubscribe(
476 auth_pb2.UnsubscribeReq(
477 payload=b64decode(params["payload"][0]),
478 sig=b64decode(params["sig"][0]),
479 )
480 )
481 assert res.response
484def test_email_deleted_users_regression(db, email_collector: EmailCollector, moderator: Moderator):
485 """
486 We introduced a bug in notify v2 where we would email deleted/banned users.
487 """
488 super_user, super_token = generate_user(is_superuser=True)
489 creating_user, creating_token = generate_user(complete_profile=True)
491 normal_user, _ = generate_user()
492 ban_user, _ = generate_user()
493 delete_user, _ = generate_user()
495 with session_scope() as session:
496 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
497 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
498 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
499 c_id = create_community(
500 session,
501 0,
502 2,
503 "Non-global Community",
504 [super_user],
505 [creating_user, normal_user, ban_user, delete_user],
506 r,
507 ).id
509 enforce_community_memberships()
511 res = events_api.create_event(token=creating_token, parent_community_id=c_id, approve_by=moderator)
512 event_id = res.event_id
513 assert not res.is_deleted
515 events_api.request_community_invite(token=creating_token, event_id=event_id)
516 email_collector.pop_for_mods(last=True)
518 with real_editor_session(super_token) as editor:
519 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
520 assert len(res.requests) == 1
521 # creating_user organizes the event, so they're excluded; counts super_user, normal_user,
522 # ban_user and delete_user
523 assert res.requests[0].approx_users_to_notify == 4
525 with session_scope() as session:
526 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
527 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
529 with real_editor_session(super_token) as editor:
530 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
531 assert len(res.requests) == 1
532 # the approximate count excludes banned/deleted users and the organizer, leaving super_user
533 # and normal_user
534 assert res.requests[0].approx_users_to_notify == 2
536 editor.DecideEventCommunityInviteRequest(
537 editor_pb2.DecideEventCommunityInviteRequestReq(
538 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
539 approve=True,
540 )
541 )
543 # only super_user and normal_user get emailed: creating_user organizes (and attends) the
544 # event, so they're excluded from the invite fan-out
545 assert email_collector.count() == 2