Coverage for app/backend/src/tests/test_email.py: 100%
275 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 19:52 +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, events_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 import api_helpers
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 events_session,
40 notifications_session,
41 real_editor_session,
42)
43from tests.test_communities import create_community
46def test_signup_verification_email(db, email_collector: EmailCollector):
47 request_email = f"{random_hex(12)}@couchers.org.invalid"
49 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
51 with session_scope() as session:
52 context = make_logged_out_context(LocalizationContext.en_utc())
53 send_signup_email(context, session, flow)
55 email = email_collector.pop_for_recipient(request_email, last=True)
56 assert email.recipient == request_email
57 assert flow.email_token
58 assert flow.email_token in email.html
59 assert flow.email_token in email.html
62def test_signup_verification_email_with_token_ending_in_src(db, email_collector: EmailCollector):
63 """
64 Signup tokens are base64 encoded, so their "=" padding makes roughly one in 65k of them end in "src=".
65 The token goes into the button's `<a href="..." style="...">`, where it must not be mistaken for an image.
66 """
67 request_email = f"{random_hex(12)}@couchers.org.invalid"
68 token = "a-token-ending-in-src="
70 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
72 with session_scope() as session:
73 context = make_logged_out_context(LocalizationContext.en_utc())
74 with patch("couchers.tasks.urlsafe_secure_token", return_value=token):
75 send_signup_email(context, session, flow)
77 email = email_collector.pop_for_recipient(request_email, last=True)
78 assert flow.email_token == token
79 assert token in email.html
82def test_report_email(db, email_collector: EmailCollector):
83 user_reporter, api_token_author = generate_user()
84 user_author, api_token_reported = generate_user()
86 with session_scope() as session:
87 report = ContentReport(
88 reporting_user_id=user_reporter.id,
89 reason="spam",
90 description="I think this is spam and does not belong on couchers",
91 content_ref="comment/123",
92 author_user_id=user_author.id,
93 user_agent="n/a",
94 page="https://couchers.org/comment/123",
95 )
96 session.add(report)
97 session.flush()
99 send_content_report_email(session, report)
101 # Load all data before session closes
102 author_username = report.author_user.username
103 author_id = report.author_user.id
104 author_email = report.author_user.email
105 reporting_username = report.reporting_user.username
106 reporting_id = report.reporting_user.id
107 reporting_email = report.reporting_user.email
108 reason = report.reason
109 description = report.description
111 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
112 assert email.recipient == "reports@couchers.org.invalid"
113 assert author_username in email.plain
114 assert str(author_id) in email.plain
115 assert author_email in email.plain
116 assert reporting_username in email.plain
117 assert str(reporting_id) in email.plain
118 assert reporting_email in email.plain
119 assert reason in email.plain
120 assert description in email.plain
121 assert "report" in email.subject.lower()
124def test_reference_report_email_not_sent(db, email_collector: EmailCollector):
125 from_user, api_token_author = generate_user()
126 to_user, api_token_reported = generate_user()
128 make_friends(from_user, to_user)
130 with session_scope() as session:
131 moderation_state = ModerationState(
132 object_type=ModerationObjectType.reference,
133 object_id=0,
134 visibility=ModerationVisibility.visible,
135 )
136 session.add(moderation_state)
137 session.flush()
138 reference = Reference(
139 from_user_id=from_user.id,
140 to_user_id=to_user.id,
141 reference_type=ReferenceType.friend,
142 text="This person was very nice to me.",
143 rating=0.9,
144 was_appropriate=True,
145 moderation_state_id=moderation_state.id,
146 )
147 session.add(reference)
148 session.flush()
149 moderation_state.object_id = reference.id
151 # no email sent for a positive ref
152 maybe_send_reference_report_email(session, reference)
154 assert email_collector.count_for_recipient("reports@couchers.org.invalid") == 0
157def test_reference_report_email(db, email_collector: EmailCollector):
158 from_user, api_token_author = generate_user()
159 to_user, api_token_reported = generate_user()
161 make_friends(from_user, to_user)
163 with session_scope() as session:
164 moderation_state = ModerationState(
165 object_type=ModerationObjectType.reference,
166 object_id=0,
167 visibility=ModerationVisibility.visible,
168 )
169 session.add(moderation_state)
170 session.flush()
171 reference = Reference(
172 from_user_id=from_user.id,
173 to_user_id=to_user.id,
174 reference_type=ReferenceType.friend,
175 text="This person was not nice to me.",
176 rating=0.3,
177 was_appropriate=False,
178 private_text="This is some private text for support",
179 moderation_state_id=moderation_state.id,
180 )
181 session.add(reference)
182 session.flush()
183 moderation_state.object_id = reference.id
185 maybe_send_reference_report_email(session, reference)
187 reference_text = reference.text
188 reference_private_text = reference.private_text
190 email = email_collector.pop_for_recipient("reports@couchers.org.invalid", last=True)
191 assert email.recipient == "reports@couchers.org.invalid"
192 assert "report" in email.subject.lower()
193 assert "reference" in email.subject.lower()
194 assert from_user.username in email.plain
195 assert str(from_user.id) in email.plain
196 assert from_user.email in email.plain
197 assert to_user.username in email.plain
198 assert str(to_user.id) in email.plain
199 assert to_user.email in email.plain
200 assert reference_text in email.plain
201 assert "friend" in email.plain.lower()
202 assert reference_private_text
203 assert reference_private_text in email.plain
206def test_email_patching_fails(db):
207 """
208 There was a problem where the mocking wasn't happening and the email dev
209 printing function was called instead, this makes sure the patching is
210 actually done
211 """
212 to_user, to_token = generate_user()
213 from_user, from_token = generate_user()
214 # Need a moderator to approve the friend request since UMS defers notification
215 mod_user, mod_token = generate_user(is_superuser=True)
216 moderator = Moderator(mod_user, mod_token)
218 patched_msg = random_hex(64)
220 def mock_queue_email(session, payload):
221 raise Exception(patched_msg)
223 with api_session(from_token) as api:
224 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
226 friend_relationship = get_friend_relationship(from_user, to_user)
227 assert friend_relationship is not None
228 moderator.approve_friend_request(friend_relationship.id)
230 with patch("couchers.email.queuing._queue_email", mock_queue_email):
231 with pytest.raises(Exception) as e:
232 process_jobs()
234 assert str(e.value) == patched_msg
237def test_email_changed_confirmation_sent_to_new_email(db, email_collector: EmailCollector):
238 confirmation_token = urlsafe_secure_token()
239 user, user_token = generate_user()
240 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
241 user.new_email_token = confirmation_token
243 with session_scope() as session:
244 user_context = make_background_user_context(user.id)
245 send_email_changed_confirmation_to_new_email(user_context, session, user)
247 email = email_collector.pop_for_recipient(user.new_email, last=True)
248 assert "new email" in email.subject
249 assert email.recipient == user.new_email
250 assert user.name in email.plain
251 assert user.name in email.html
252 assert user.email in email.plain
253 assert user.email in email.html
254 assert "You requested that your email be changed from " in email.plain
255 assert "You requested that your email be changed from " in email.html
256 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.plain
257 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in email.html
258 assert "support@couchers.org" in email.plain
259 assert "support@couchers.org" in email.html
262def test_do_not_email_security(db, email_collector: EmailCollector):
263 user, token = generate_user()
265 password_reset_token = urlsafe_secure_token()
267 with notifications_session(token) as notifications:
268 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
270 # make sure we still get security emails
272 with session_scope() as session:
273 notify(
274 session,
275 user_id=user.id,
276 topic_action=NotificationTopicAction.password_reset__start,
277 key="",
278 data=notification_data_pb2.PasswordResetStart(
279 password_reset_token=password_reset_token,
280 ),
281 )
283 email = email_collector.pop_for_recipient(user.email, last=True)
284 assert email.recipient == user.email
285 assert "reset" in email.subject.lower()
286 assert password_reset_token in email.plain
287 assert password_reset_token in email.html
288 unique_string = "You asked for your password to be reset on Couchers.org."
289 assert unique_string in email.plain
290 assert unique_string in email.html
291 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.plain
292 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in email.html
293 assert "support@couchers.org" in email.plain
294 assert "support@couchers.org" in email.html
296 assert "/quick-link?payload=" not in email.plain
297 assert "/quick-link?payload=" not in email.html
300def test_do_not_email_non_security(db, email_collector: EmailCollector):
301 user, token1 = generate_user(complete_profile=True)
302 from_user, token2 = generate_user(complete_profile=True)
303 # Need a moderator to approve the friend request since UMS defers notification
304 mod_user, mod_token = generate_user(is_superuser=True)
305 moderator = Moderator(mod_user, mod_token)
307 with notifications_session(token1) as notifications:
308 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
310 with api_session(token2) as api:
311 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
313 friend_relationship = get_friend_relationship(from_user, user)
314 assert friend_relationship is not None
315 moderator.approve_friend_request(friend_relationship.id)
317 assert email_collector.count_for_recipient(user.email) == 0
320def test_do_not_email_non_security_unsublink(db, email_collector: EmailCollector):
321 user, _ = generate_user(complete_profile=True)
322 from_user, token2 = generate_user(complete_profile=True)
323 # Need a moderator to approve the friend request since UMS defers notification
324 mod_user, mod_token = generate_user(is_superuser=True)
325 moderator = Moderator(mod_user, mod_token)
327 with api_session(token2) as api:
328 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
330 friend_relationship = get_friend_relationship(from_user, user)
331 assert friend_relationship is not None
332 moderator.approve_friend_request(friend_relationship.id)
334 email = email_collector.pop_for_recipient(user.email, last=True)
336 assert "/quick-link?payload=" in email.plain
337 assert "/quick-link?payload=" in email.html
340def test_email_prefix_config(db, email_collector: EmailCollector):
341 user, _ = generate_user()
343 with session_scope() as session:
344 notify(
345 session,
346 user_id=user.id,
347 topic_action=NotificationTopicAction.donation__received,
348 key="",
349 data=notification_data_pb2.DonationReceived(
350 amount=20,
351 receipt_url="https://example.com/receipt/12345",
352 ),
353 )
355 email1 = email_collector.pop_for_recipient(user.email, last=True)
356 assert email1.sender_name == "Couchers.org"
357 assert email1.sender_email == "notify@couchers.org.invalid"
358 assert email1.subject == "[TEST] Thank you for your donation to Couchers.org!"
360 config.NOTIFICATION_EMAIL_SENDER = "TestCo"
361 config.NOTIFICATION_EMAIL_ADDRESS = "testco@testing.co.invalid"
362 config.NOTIFICATION_PREFIX = ""
364 with session_scope() as session:
365 notify(
366 session,
367 user_id=user.id,
368 topic_action=NotificationTopicAction.donation__received,
369 key="",
370 data=notification_data_pb2.DonationReceived(
371 amount=20,
372 receipt_url="https://example.com/receipt/12345",
373 ),
374 )
376 email2 = email_collector.pop_for_recipient(user.email, last=True)
377 assert email2.sender_name == "TestCo"
378 assert email2.sender_email == "testco@testing.co.invalid"
379 assert email2.subject == "Thank you for your donation to Couchers.org!"
382def test_send_donation_email(db):
383 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
385 config.ENABLE_EMAIL = True
387 with session_scope() as session:
388 notify(
389 session,
390 user_id=user.id,
391 topic_action=NotificationTopicAction.donation__received,
392 key="",
393 data=notification_data_pb2.DonationReceived(
394 amount=20,
395 receipt_url="https://example.com/receipt/12345",
396 ),
397 )
399 with patch("couchers.email.smtp.smtplib.SMTP"):
400 process_jobs()
402 with session_scope() as session:
403 email = session.execute(select(Email)).scalar_one()
404 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
405 assert (
406 email.plain
407 == """Hi Testy von Test,
409Thank you so much for your donation of $20 to Couchers.org.
411Your 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.
413You can download an invoice and receipt for the donation here:
415Download invoice: https://example.com/receipt/12345
417Couchers, 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.
419If you have any questions about your donation, please email us at donations@couchers.org.
421Thank you!
423Aapeli and Itsi,
424Couchers.org Founders
426---
428This is a security email, you cannot unsubscribe from it.
429"""
430 )
432 assert "Thank you so much for your donation of <b>$20</b> to Couchers.org." in email.html
433 assert email.sender_name == "Couchers.org"
434 assert email.sender_email == "notify@couchers.org.invalid"
435 assert email.recipient == "testing@couchers.org.invalid"
436 assert "https://example.com/receipt/12345" in email.html
437 assert not email.list_unsubscribe_header
438 assert email.source_data and ("donation:received" in email.source_data)
441def test_chat_missed_messages_list_unsubscribe_header(db, email_collector: EmailCollector):
442 """
443 Regression test: chat__missed_messages has key="" (it's a summary, not tied to a single chat).
444 The List-Unsubscribe header must use a topic_action unsubscribe link, not a topic_key link.
445 """
446 user, _ = generate_user()
448 with session_scope() as session:
449 notify(
450 session,
451 user_id=user.id,
452 topic_action=NotificationTopicAction.chat__missed_messages,
453 key="",
454 data=notification_data_pb2.ChatMissedMessages(
455 messages=[
456 notification_data_pb2.ChatMessage(
457 author=api_pb2.User(name="Test User", user_id=2, username="testuser"),
458 text="Hello!",
459 group_chat_id=99,
460 unseen_count=1,
461 ),
462 ],
463 ),
464 )
466 email = email_collector.pop_for_recipient(user.email, last=True)
468 assert email.list_unsubscribe_header
470 # Extract the List-Unsubscribe URL and call the Unsubscribe endpoint
471 url = email.list_unsubscribe_header.strip("<>")
472 url_parts = urlparse(url)
473 params = parse_qs(url_parts.query)
475 with auth_api_session() as (auth_api, metadata_interceptor):
476 res = auth_api.Unsubscribe(
477 auth_pb2.UnsubscribeReq(
478 payload=b64decode(params["payload"][0]),
479 sig=b64decode(params["sig"][0]),
480 )
481 )
482 assert res.response
485def test_email_deleted_users_regression(db, email_collector: EmailCollector, moderator: Moderator):
486 """
487 We introduced a bug in notify v2 where we would email deleted/banned users.
488 """
489 super_user, super_token = generate_user(is_superuser=True)
490 creating_user, creating_token = generate_user(complete_profile=True)
492 normal_user, _ = generate_user()
493 ban_user, _ = generate_user()
494 delete_user, _ = generate_user()
496 with session_scope() as session:
497 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
498 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
499 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
500 c_id = create_community(
501 session,
502 0,
503 2,
504 "Non-global Community",
505 [super_user],
506 [creating_user, normal_user, ban_user, delete_user],
507 r,
508 ).id
510 enforce_community_memberships()
512 res = api_helpers.create_event(token=creating_token, parent_community_id=c_id)
513 event_id = res.event_id
514 assert not res.is_deleted
516 moderator.approve_event_occurrence(event_id)
518 with events_session(creating_token) as api:
519 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
521 email_collector.pop_for_mods(last=True)
523 with real_editor_session(super_token) as editor:
524 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
525 assert len(res.requests) == 1
526 # creating_user organizes the event, so they're excluded; counts super_user, normal_user,
527 # ban_user and delete_user
528 assert res.requests[0].approx_users_to_notify == 4
530 with session_scope() as session:
531 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
532 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
534 with real_editor_session(super_token) as editor:
535 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
536 assert len(res.requests) == 1
537 # the approximate count excludes banned/deleted users and the organizer, leaving super_user
538 # and normal_user
539 assert res.requests[0].approx_users_to_notify == 2
541 editor.DecideEventCommunityInviteRequest(
542 editor_pb2.DecideEventCommunityInviteRequestReq(
543 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
544 approve=True,
545 )
546 )
548 # only super_user and normal_user get emailed: creating_user organizes (and attends) the
549 # event, so they're excluded from the invite fan-out
550 assert email_collector.count() == 2