Coverage for app/backend/src/tests/test_email.py: 100%
279 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1from datetime import timedelta
2from unittest.mock import patch
3from urllib.parse import parse_qs, urlparse
5import pytest
6from sqlalchemy import func, select, update
8from couchers.config import config
9from couchers.context import make_background_user_context, make_logged_out_context
10from couchers.crypto import b64decode, random_hex, urlsafe_secure_token
11from couchers.db import session_scope
12from couchers.i18n import LocalizationContext
13from couchers.models import (
14 ContentReport,
15 Email,
16 ModerationObjectType,
17 ModerationState,
18 ModerationVisibility,
19 Reference,
20 ReferenceType,
21 SignupFlow,
22 User,
23)
24from couchers.models.notifications import NotificationTopicAction
25from couchers.notifications.notify import notify
26from couchers.proto import api_pb2, auth_pb2, editor_pb2, events_pb2, notification_data_pb2, notifications_pb2
27from couchers.tasks import (
28 enforce_community_memberships,
29 maybe_send_reference_report_email,
30 send_content_report_email,
31 send_email_changed_confirmation_to_new_email,
32 send_signup_email,
33)
34from couchers.utils import datetime_to_iso8601_local, now
35from tests.fixtures.db import generate_user, get_friend_relationship, make_friends
36from tests.fixtures.misc import EmailCollector, Moderator, process_jobs
37from tests.fixtures.sessions import (
38 api_session,
39 auth_api_session,
40 events_session,
41 notifications_session,
42 real_editor_session,
43)
44from tests.test_communities import create_community
47def test_signup_verification_email(db, email_collector: EmailCollector):
48 request_email = f"{random_hex(12)}@couchers.org.invalid"
50 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
52 with session_scope() as session:
53 context = make_logged_out_context(LocalizationContext.en_utc())
54 send_signup_email(context, session, flow)
56 email = email_collector.pop_for_recipient(request_email, last=True)
57 assert email.recipient == request_email
58 assert flow.email_token
59 assert flow.email_token in email.html
60 assert flow.email_token in email.html
63def test_signup_verification_email_with_token_ending_in_src(db, email_collector: EmailCollector):
64 """
65 Signup tokens are base64 encoded, so their "=" padding makes roughly one in 65k of them end in "src=".
66 The token goes into the button's `<a href="..." style="...">`, where it must not be mistaken for an image.
67 """
68 request_email = f"{random_hex(12)}@couchers.org.invalid"
69 token = "a-token-ending-in-src="
71 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
73 with session_scope() as session:
74 context = make_logged_out_context(LocalizationContext.en_utc())
75 with patch("couchers.tasks.urlsafe_secure_token", return_value=token):
76 send_signup_email(context, session, flow)
78 email = email_collector.pop_for_recipient(request_email, last=True)
79 assert flow.email_token == token
80 assert token in email.html
83def test_report_email(db, email_collector: EmailCollector):
84 user_reporter, api_token_author = generate_user()
85 user_author, api_token_reported = generate_user()
87 with session_scope() as session:
88 report = ContentReport(
89 reporting_user_id=user_reporter.id,
90 reason="spam",
91 description="I think this is spam and does not belong on couchers",
92 content_ref="comment/123",
93 author_user_id=user_author.id,
94 user_agent="n/a",
95 page="https://couchers.org/comment/123",
96 )
97 session.add(report)
98 session.flush()
100 send_content_report_email(session, report)
102 # Load all data before session closes
103 author_username = report.author_user.username
104 author_id = report.author_user.id
105 author_email = report.author_user.email
106 reporting_username = report.reporting_user.username
107 reporting_id = report.reporting_user.id
108 reporting_email = report.reporting_user.email
109 reason = report.reason
110 description = report.description
112 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
113 assert email.recipient == "reports@couchers.org.invalid"
114 assert author_username in email.plain
115 assert str(author_id) in email.plain
116 assert author_email in email.plain
117 assert reporting_username in email.plain
118 assert str(reporting_id) in email.plain
119 assert reporting_email in email.plain
120 assert reason in email.plain
121 assert description in email.plain
122 assert "report" in email.subject.lower()
125def test_reference_report_email_not_sent(db, email_collector: EmailCollector):
126 from_user, api_token_author = generate_user()
127 to_user, api_token_reported = generate_user()
129 make_friends(from_user, to_user)
131 with session_scope() as session:
132 moderation_state = ModerationState(
133 object_type=ModerationObjectType.reference,
134 object_id=0,
135 visibility=ModerationVisibility.visible,
136 )
137 session.add(moderation_state)
138 session.flush()
139 reference = Reference(
140 from_user_id=from_user.id,
141 to_user_id=to_user.id,
142 reference_type=ReferenceType.friend,
143 text="This person was very nice to me.",
144 rating=0.9,
145 was_appropriate=True,
146 moderation_state_id=moderation_state.id,
147 )
148 session.add(reference)
149 session.flush()
150 moderation_state.object_id = reference.id
152 # no email sent for a positive ref
153 maybe_send_reference_report_email(session, reference)
155 assert email_collector.count_for_recipient("reports@couchers.org.invalid") == 0
158def test_reference_report_email(db, email_collector: EmailCollector):
159 from_user, api_token_author = generate_user()
160 to_user, api_token_reported = generate_user()
162 make_friends(from_user, to_user)
164 with session_scope() as session:
165 moderation_state = ModerationState(
166 object_type=ModerationObjectType.reference,
167 object_id=0,
168 visibility=ModerationVisibility.visible,
169 )
170 session.add(moderation_state)
171 session.flush()
172 reference = Reference(
173 from_user_id=from_user.id,
174 to_user_id=to_user.id,
175 reference_type=ReferenceType.friend,
176 text="This person was not nice to me.",
177 rating=0.3,
178 was_appropriate=False,
179 private_text="This is some private text for support",
180 moderation_state_id=moderation_state.id,
181 )
182 session.add(reference)
183 session.flush()
184 moderation_state.object_id = reference.id
186 maybe_send_reference_report_email(session, reference)
188 reference_text = reference.text
189 reference_private_text = reference.private_text
191 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
192 assert email.recipient == "reports@couchers.org.invalid"
193 assert "report" in email.subject.lower()
194 assert "reference" in email.subject.lower()
195 assert from_user.username in email.plain
196 assert str(from_user.id) in email.plain
197 assert from_user.email in email.plain
198 assert to_user.username in email.plain
199 assert str(to_user.id) in email.plain
200 assert to_user.email in email.plain
201 assert reference_text in email.plain
202 assert "friend" in email.plain.lower()
203 assert reference_private_text
204 assert reference_private_text in email.plain
207def test_email_patching_fails(db):
208 """
209 There was a problem where the mocking wasn't happening and the email dev
210 printing function was called instead, this makes sure the patching is
211 actually done
212 """
213 to_user, to_token = generate_user()
214 from_user, from_token = generate_user()
215 # Need a moderator to approve the friend request since UMS defers notification
216 mod_user, mod_token = generate_user(is_superuser=True)
217 moderator = Moderator(mod_user, mod_token)
219 patched_msg = random_hex(64)
221 def mock_queue_email(session, payload):
222 raise Exception(patched_msg)
224 with api_session(from_token) as api:
225 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
227 friend_relationship = get_friend_relationship(from_user, to_user)
228 assert friend_relationship is not None
229 moderator.approve_friend_request(friend_relationship.id)
231 with patch("couchers.email.queuing._queue_email", mock_queue_email):
232 with pytest.raises(Exception) as e:
233 process_jobs()
235 assert str(e.value) == patched_msg
238def test_email_changed_confirmation_sent_to_new_email(db, email_collector: EmailCollector):
239 confirmation_token = urlsafe_secure_token()
240 user, user_token = generate_user()
241 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
242 user.new_email_token = confirmation_token
244 with session_scope() as session:
245 user_context = make_background_user_context(user.id)
246 send_email_changed_confirmation_to_new_email(user_context, session, user)
248 email = email_collector.pop_for_recipient(user.new_email, last=True)
249 assert "new email" in email.subject
250 assert email.recipient == user.new_email
251 assert user.name in email.plain
252 assert user.name in email.html
253 assert user.email in email.plain
254 assert user.email in email.html
255 assert "You requested that your email be changed from " in email.plain
256 assert "You requested that your email be changed from " in email.html
257 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.plain
258 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.html
259 assert "support@couchers.org" in email.plain
260 assert "support@couchers.org" in email.html
263def test_do_not_email_security(db, email_collector: EmailCollector):
264 user, token = generate_user()
266 password_reset_token = urlsafe_secure_token()
268 with notifications_session(token) as notifications:
269 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
271 # make sure we still get security emails
273 with session_scope() as session:
274 notify(
275 session,
276 user_id=user.id,
277 topic_action=NotificationTopicAction.password_reset__start,
278 key="",
279 data=notification_data_pb2.PasswordResetStart(
280 password_reset_token=password_reset_token,
281 ),
282 )
284 email = email_collector.pop_for_recipient(user.email, last=True)
285 assert email.recipient == user.email
286 assert "reset" in email.subject.lower()
287 assert password_reset_token in email.plain
288 assert password_reset_token in email.html
289 unique_string = "You asked for your password to be reset on Couchers.org."
290 assert unique_string in email.plain
291 assert unique_string in email.html
292 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.plain
293 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.html
294 assert "support@couchers.org" in email.plain
295 assert "support@couchers.org" in email.html
297 assert "/quick-link?payload=" not in email.plain
298 assert "/quick-link?payload=" not in email.html
301def test_do_not_email_non_security(db, email_collector: EmailCollector):
302 user, token1 = generate_user(complete_profile=True)
303 from_user, token2 = generate_user(complete_profile=True)
304 # Need a moderator to approve the friend request since UMS defers notification
305 mod_user, mod_token = generate_user(is_superuser=True)
306 moderator = Moderator(mod_user, mod_token)
308 with notifications_session(token1) as notifications:
309 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
311 with api_session(token2) as api:
312 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
314 friend_relationship = get_friend_relationship(from_user, user)
315 assert friend_relationship is not None
316 moderator.approve_friend_request(friend_relationship.id)
318 assert email_collector.count_for_recipient(user.email) == 0
321def test_do_not_email_non_security_unsublink(db, email_collector: EmailCollector):
322 user, _ = generate_user(complete_profile=True)
323 from_user, token2 = generate_user(complete_profile=True)
324 # Need a moderator to approve the friend request since UMS defers notification
325 mod_user, mod_token = generate_user(is_superuser=True)
326 moderator = Moderator(mod_user, mod_token)
328 with api_session(token2) as api:
329 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
331 friend_relationship = get_friend_relationship(from_user, user)
332 assert friend_relationship is not None
333 moderator.approve_friend_request(friend_relationship.id)
335 email = email_collector.pop_for_recipient(user.email, last=True)
337 assert "/quick-link?payload=" in email.plain
338 assert "/quick-link?payload=" in email.html
341def test_email_prefix_config(db, email_collector: EmailCollector):
342 user, _ = generate_user()
344 with session_scope() as session:
345 notify(
346 session,
347 user_id=user.id,
348 topic_action=NotificationTopicAction.donation__received,
349 key="",
350 data=notification_data_pb2.DonationReceived(
351 amount=20,
352 receipt_url="https://example.com/receipt/12345",
353 ),
354 )
356 email1 = email_collector.pop_for_recipient(user.email, last=True)
357 assert email1.sender_name == "Couchers.org"
358 assert email1.sender_email == "notify@couchers.org.invalid"
359 assert email1.subject == "[TEST] Thank you for your donation to Couchers.org!"
361 config.NOTIFICATION_EMAIL_SENDER = "TestCo"
362 config.NOTIFICATION_EMAIL_ADDRESS = "testco@testing.co.invalid"
363 config.NOTIFICATION_PREFIX = ""
365 with session_scope() as session:
366 notify(
367 session,
368 user_id=user.id,
369 topic_action=NotificationTopicAction.donation__received,
370 key="",
371 data=notification_data_pb2.DonationReceived(
372 amount=20,
373 receipt_url="https://example.com/receipt/12345",
374 ),
375 )
377 email2 = email_collector.pop_for_recipient(user.email, last=True)
378 assert email2.sender_name == "TestCo"
379 assert email2.sender_email == "testco@testing.co.invalid"
380 assert email2.subject == "Thank you for your donation to Couchers.org!"
383def test_send_donation_email(db):
384 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
386 config.ENABLE_EMAIL = True
388 with session_scope() as session:
389 notify(
390 session,
391 user_id=user.id,
392 topic_action=NotificationTopicAction.donation__received,
393 key="",
394 data=notification_data_pb2.DonationReceived(
395 amount=20,
396 receipt_url="https://example.com/receipt/12345",
397 ),
398 )
400 with patch("couchers.email.smtp.smtplib.SMTP"):
401 process_jobs()
403 with session_scope() as session:
404 email = session.execute(select(Email)).scalar_one()
405 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
406 assert (
407 email.plain
408 == """Hi Testy von Test,
410Thank you so much for your donation of $20 to Couchers.org.
412Your 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.
414You can download an invoice and receipt for the donation here:
416Download invoice: https://example.com/receipt/12345
418Couchers, 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.
420If you have any questions about your donation, please email us at donations@couchers.org.
422Thank you!
424Aapeli and Itsi,
425Couchers.org Founders
427---
429This is a security email, you cannot unsubscribe from it.
430"""
431 )
433 assert "Thank you so much for your donation of <b>$20</b> to Couchers.org." in email.html
434 assert email.sender_name == "Couchers.org"
435 assert email.sender_email == "notify@couchers.org.invalid"
436 assert email.recipient == "testing@couchers.org.invalid"
437 assert "https://example.com/receipt/12345" in email.html
438 assert not email.list_unsubscribe_header
439 assert email.source_data and ("donation:received" in email.source_data)
442def test_chat_missed_messages_list_unsubscribe_header(db, email_collector: EmailCollector):
443 """
444 Regression test: chat__missed_messages has key="" (it's a summary, not tied to a single chat).
445 The List-Unsubscribe header must use a topic_action unsubscribe link, not a topic_key link.
446 """
447 user, _ = generate_user()
449 with session_scope() as session:
450 notify(
451 session,
452 user_id=user.id,
453 topic_action=NotificationTopicAction.chat__missed_messages,
454 key="",
455 data=notification_data_pb2.ChatMissedMessages(
456 messages=[
457 notification_data_pb2.ChatMessage(
458 author=api_pb2.User(name="Test User", user_id=2, username="testuser"),
459 text="Hello!",
460 group_chat_id=99,
461 unseen_count=1,
462 ),
463 ],
464 ),
465 )
467 email = email_collector.pop_for_recipient(user.email, last=True)
469 assert email.list_unsubscribe_header
471 # Extract the List-Unsubscribe URL and call the Unsubscribe endpoint
472 url = email.list_unsubscribe_header.strip("<>")
473 url_parts = urlparse(url)
474 params = parse_qs(url_parts.query)
476 with auth_api_session() as (auth_api, metadata_interceptor):
477 res = auth_api.Unsubscribe(
478 auth_pb2.UnsubscribeReq(
479 payload=b64decode(params["payload"][0]),
480 sig=b64decode(params["sig"][0]),
481 )
482 )
483 assert res.response
486def test_email_deleted_users_regression(db, email_collector: EmailCollector, moderator: Moderator):
487 """
488 We introduced a bug in notify v2 where we would email deleted/banned users.
489 """
490 super_user, super_token = generate_user(is_superuser=True)
491 creating_user, creating_token = generate_user(complete_profile=True)
493 normal_user, _ = generate_user()
494 ban_user, _ = generate_user()
495 delete_user, _ = generate_user()
497 with session_scope() as session:
498 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
499 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
500 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
501 c_id = create_community(
502 session,
503 0,
504 2,
505 "Non-global Community",
506 [super_user],
507 [creating_user, normal_user, ban_user, delete_user],
508 r,
509 ).id
511 enforce_community_memberships()
513 start_time = now() + timedelta(hours=2)
514 end_time = start_time + timedelta(hours=3)
515 with events_session(creating_token) as api:
516 res = api.CreateEvent(
517 events_pb2.CreateEventReq(
518 title="Dummy Title",
519 content="Dummy content.",
520 photo_key=None,
521 parent_community_id=c_id,
522 location=events_pb2.EventLocation(
523 address="Near Null Island",
524 lat=0.1,
525 lng=0.2,
526 ),
527 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time),
528 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time),
529 )
530 )
531 event_id = res.event_id
532 assert not res.is_deleted
534 moderator.approve_event_occurrence(event_id)
536 with events_session(creating_token) as api:
537 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
539 email_collector.pop_for_mods(last=True)
541 with real_editor_session(super_token) as editor:
542 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
543 assert len(res.requests) == 1
544 # creating_user organizes the event, so they're excluded; counts super_user, normal_user,
545 # ban_user and delete_user
546 assert res.requests[0].approx_users_to_notify == 4
548 with session_scope() as session:
549 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
550 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
552 with real_editor_session(super_token) as editor:
553 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
554 assert len(res.requests) == 1
555 # the approximate count excludes banned/deleted users and the organizer, leaving super_user
556 # and normal_user
557 assert res.requests[0].approx_users_to_notify == 2
559 editor.DecideEventCommunityInviteRequest(
560 editor_pb2.DecideEventCommunityInviteRequestReq(
561 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
562 approve=True,
563 )
564 )
566 # only super_user and normal_user get emailed: creating_user organizes (and attends) the
567 # event, so they're excluded from the invite fan-out
568 assert email_collector.count() == 2