Coverage for app / backend / src / tests / test_email.py: 100%
285 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 11:04 +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
8import couchers.email
9import couchers.jobs.handlers
10from couchers.config import config
11from couchers.crypto import b64decode, random_hex, urlsafe_secure_token
12from couchers.db import session_scope
13from couchers.models import (
14 ContentReport,
15 Email,
16 Reference,
17 ReferenceType,
18 SignupFlow,
19 User,
20)
21from couchers.models.notifications import NotificationTopicAction
22from couchers.notifications.notify import notify
23from couchers.proto import api_pb2, auth_pb2, editor_pb2, events_pb2, notification_data_pb2, notifications_pb2
24from couchers.tasks import (
25 enforce_community_memberships,
26 maybe_send_reference_report_email,
27 send_content_report_email,
28 send_email_changed_confirmation_to_new_email,
29 send_signup_email,
30)
31from couchers.utils import Timestamp_from_datetime, now
32from tests.fixtures.db import generate_user, get_friend_relationship, make_friends
33from tests.fixtures.misc import Moderator, email_fields, mock_notification_email, process_jobs
34from tests.fixtures.sessions import (
35 api_session,
36 auth_api_session,
37 events_session,
38 notifications_session,
39 real_editor_session,
40)
41from tests.test_communities import create_community
44@pytest.fixture(autouse=True)
45def _(testconfig):
46 pass
49def test_signup_verification_email(db):
50 request_email = f"{random_hex(12)}@couchers.org.invalid"
52 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
54 with session_scope() as session:
55 with mock_notification_email() as mock:
56 send_signup_email(session, flow)
58 assert mock.call_count == 1
59 e = email_fields(mock)
60 assert e.recipient == request_email
61 assert flow.email_token
62 assert flow.email_token in e.html
63 assert flow.email_token in e.html
66def test_report_email(db):
67 user_reporter, api_token_author = generate_user()
68 user_author, api_token_reported = generate_user()
70 with session_scope() as session:
71 report = ContentReport(
72 reporting_user_id=user_reporter.id,
73 reason="spam",
74 description="I think this is spam and does not belong on couchers",
75 content_ref="comment/123",
76 author_user_id=user_author.id,
77 user_agent="n/a",
78 page="https://couchers.org/comment/123",
79 )
80 session.add(report)
81 session.flush()
83 with mock_notification_email() as mock:
84 send_content_report_email(session, report)
86 # Load all data before session closes
87 author_username = report.author_user.username
88 author_id = report.author_user.id
89 author_email = report.author_user.email
90 reporting_username = report.reporting_user.username
91 reporting_id = report.reporting_user.id
92 reporting_email = report.reporting_user.email
93 reason = report.reason
94 description = report.description
96 assert mock.call_count == 1
98 e = email_fields(mock)
99 assert e.recipient == "reports@couchers.org.invalid"
100 assert author_username in e.plain
101 assert str(author_id) in e.plain
102 assert author_email in e.plain
103 assert reporting_username in e.plain
104 assert str(reporting_id) in e.plain
105 assert reporting_email in e.plain
106 assert reason in e.plain
107 assert description in e.plain
108 assert "report" in e.subject.lower()
111def test_reference_report_email_not_sent(db):
112 from_user, api_token_author = generate_user()
113 to_user, api_token_reported = generate_user()
115 make_friends(from_user, to_user)
117 with session_scope() as session:
118 reference = Reference(
119 from_user_id=from_user.id,
120 to_user_id=to_user.id,
121 reference_type=ReferenceType.friend,
122 text="This person was very nice to me.",
123 rating=0.9,
124 was_appropriate=True,
125 )
127 # no email sent for a positive ref
129 with mock_notification_email() as mock:
130 maybe_send_reference_report_email(session, reference)
132 assert mock.call_count == 0
135def test_reference_report_email(db):
136 from_user, api_token_author = generate_user()
137 to_user, api_token_reported = generate_user()
139 make_friends(from_user, to_user)
141 with session_scope() as session:
142 reference = Reference(
143 from_user_id=from_user.id,
144 to_user_id=to_user.id,
145 reference_type=ReferenceType.friend,
146 text="This person was not nice to me.",
147 rating=0.3,
148 was_appropriate=False,
149 private_text="This is some private text for support",
150 )
151 session.add(reference)
152 session.flush()
154 with mock_notification_email() as mock:
155 maybe_send_reference_report_email(session, reference)
157 assert mock.call_count == 1
158 e = email_fields(mock)
159 assert e.recipient == "reports@couchers.org.invalid"
160 assert "report" in e.subject.lower()
161 assert "reference" in e.subject.lower()
162 assert from_user.username in e.plain
163 assert str(from_user.id) in e.plain
164 assert from_user.email in e.plain
165 assert to_user.username in e.plain
166 assert str(to_user.id) in e.plain
167 assert to_user.email in e.plain
168 assert reference.text in e.plain
169 assert "friend" in e.plain.lower()
170 assert reference.private_text
171 assert reference.private_text in e.plain
174def test_email_patching_fails(db):
175 """
176 There was a problem where the mocking wasn't happening and the email dev
177 printing function was called instead, this makes sure the patching is
178 actually done
179 """
180 to_user, to_token = generate_user()
181 from_user, from_token = generate_user()
182 # Need a moderator to approve the friend request since UMS defers notification
183 mod_user, mod_token = generate_user(is_superuser=True)
184 moderator = Moderator(mod_user, mod_token)
186 patched_msg = random_hex(64)
188 def mock_queue_email(session, **kwargs):
189 raise Exception(patched_msg)
191 with api_session(from_token) as api:
192 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
194 friend_relationship = get_friend_relationship(from_user, to_user)
195 assert friend_relationship is not None
196 moderator.approve_friend_request(friend_relationship.id)
198 with patch("couchers.email.queuing._queue_email", mock_queue_email):
199 with pytest.raises(Exception) as e:
200 process_jobs()
202 assert str(e.value) == patched_msg
205def test_email_changed_confirmation_sent_to_new_email(db):
206 confirmation_token = urlsafe_secure_token()
207 user, user_token = generate_user()
208 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
209 user.new_email_token = confirmation_token
210 with session_scope() as session:
211 with mock_notification_email() as mock:
212 send_email_changed_confirmation_to_new_email(session, user)
214 assert mock.call_count == 1
215 e = email_fields(mock)
216 assert "new email" in e.subject
217 assert e.recipient == user.new_email
218 assert user.name in e.plain
219 assert user.name in e.html
220 assert user.email in e.plain
221 assert user.email in e.html
222 assert "Your old email address is" in e.plain
223 assert "Your old email address is" in e.html
224 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
225 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
226 assert "support@couchers.org" in e.plain
227 assert "support@couchers.org" in e.html
230def test_do_not_email_security(db):
231 user, token = generate_user()
233 password_reset_token = urlsafe_secure_token()
235 with notifications_session(token) as notifications:
236 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
238 # make sure we still get security emails
240 with mock_notification_email() as mock:
241 with session_scope() as session:
242 notify(
243 session,
244 user_id=user.id,
245 topic_action=NotificationTopicAction.password_reset__start,
246 key="",
247 data=notification_data_pb2.PasswordResetStart(
248 password_reset_token=password_reset_token,
249 ),
250 )
252 assert mock.call_count == 1
253 e = email_fields(mock)
254 assert e.recipient == user.email
255 assert "reset" in e.subject.lower()
256 assert password_reset_token in e.plain
257 assert password_reset_token in e.html
258 unique_string = "You asked for your password to be reset on Couchers.org."
259 assert unique_string in e.plain
260 assert unique_string in e.html
261 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
262 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
263 assert "support@couchers.org" in e.plain
264 assert "support@couchers.org" in e.html
266 assert "/quick-link?payload=" not in e.plain
267 assert "/quick-link?payload=" not in e.html
270def test_do_not_email_non_security(db):
271 user, token1 = generate_user(complete_profile=True)
272 from_user, token2 = generate_user(complete_profile=True)
273 # Need a moderator to approve the friend request since UMS defers notification
274 mod_user, mod_token = generate_user(is_superuser=True)
275 moderator = Moderator(mod_user, mod_token)
277 with notifications_session(token1) as notifications:
278 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
280 with api_session(token2) as api:
281 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
283 friend_relationship = get_friend_relationship(from_user, user)
284 assert friend_relationship is not None
285 moderator.approve_friend_request(friend_relationship.id)
287 with mock_notification_email() as mock:
288 process_jobs()
290 assert mock.call_count == 0
293def test_do_not_email_non_security_unsublink(db):
294 user, _ = generate_user(complete_profile=True)
295 from_user, token2 = generate_user(complete_profile=True)
296 # Need a moderator to approve the friend request since UMS defers notification
297 mod_user, mod_token = generate_user(is_superuser=True)
298 moderator = Moderator(mod_user, mod_token)
300 with api_session(token2) as api:
301 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
303 friend_relationship = get_friend_relationship(from_user, user)
304 assert friend_relationship is not None
305 moderator.approve_friend_request(friend_relationship.id)
307 with mock_notification_email() as mock:
308 process_jobs()
310 assert mock.call_count == 1
311 e = email_fields(mock)
313 assert "/quick-link?payload=" in e.plain
314 assert "/quick-link?payload=" in e.html
317def test_email_prefix_config(db, monkeypatch):
318 user, _ = generate_user()
320 with mock_notification_email() as mock:
321 with session_scope() as session:
322 notify(
323 session,
324 user_id=user.id,
325 topic_action=NotificationTopicAction.donation__received,
326 key="",
327 data=notification_data_pb2.DonationReceived(
328 amount=20,
329 receipt_url="https://example.com/receipt/12345",
330 ),
331 )
333 assert mock.call_count == 1
334 e = email_fields(mock)
335 assert e.sender_name == "Couchers.org"
336 assert e.sender_email == "notify@couchers.org.invalid"
337 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
339 new_config = config.copy()
340 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
341 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
342 new_config["NOTIFICATION_PREFIX"] = ""
344 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
346 with mock_notification_email() as mock:
347 with session_scope() as session:
348 notify(
349 session,
350 user_id=user.id,
351 topic_action=NotificationTopicAction.donation__received,
352 key="",
353 data=notification_data_pb2.DonationReceived(
354 amount=20,
355 receipt_url="https://example.com/receipt/12345",
356 ),
357 )
359 assert mock.call_count == 1
360 e = email_fields(mock)
361 assert e.sender_name == "TestCo"
362 assert e.sender_email == "testco@testing.co.invalid"
363 assert e.subject == "Thank you for your donation to Couchers.org!"
366def test_send_donation_email(db, monkeypatch):
367 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
369 new_config = config.copy()
370 new_config["ENABLE_EMAIL"] = True
372 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
374 with session_scope() as session:
375 notify(
376 session,
377 user_id=user.id,
378 topic_action=NotificationTopicAction.donation__received,
379 key="",
380 data=notification_data_pb2.DonationReceived(
381 amount=20,
382 receipt_url="https://example.com/receipt/12345",
383 ),
384 )
386 with patch("couchers.email.smtp.smtplib.SMTP"):
387 process_jobs()
389 with session_scope() as session:
390 email = session.execute(select(Email)).scalar_one()
391 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
392 assert (
393 email.plain
394 == """Dear Testy von Test,
396Thank you so much for your donation of $20 to Couchers.org.
398Your 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.
401You can download an invoice and receipt for the donation here:
403<https://example.com/receipt/12345>
405If you have any questions about your donation, please email us at <donations@couchers.org>.
407Your generosity will help deliver the platform for everyone.
410Thank you!
412Aapeli and Itsi,
413Couchers.org Founders
416---
418This is a security email, you cannot unsubscribe from it.
419"""
420 )
422 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
423 assert email.sender_name == "Couchers.org"
424 assert email.sender_email == "notify@couchers.org.invalid"
425 assert email.recipient == "testing@couchers.org.invalid"
426 assert "https://example.com/receipt/12345" in email.html
427 assert not email.list_unsubscribe_header
428 assert email.source_data == "testing_version/donation_received"
431def test_chat_missed_messages_list_unsubscribe_header(db):
432 """
433 Regression test: chat__missed_messages has key="" (it's a summary, not tied to a single chat).
434 The List-Unsubscribe header must use a topic_action unsubscribe link, not a topic_key link.
435 """
436 user, _ = generate_user()
438 with mock_notification_email() as mock:
439 with session_scope() as session:
440 notify(
441 session,
442 user_id=user.id,
443 topic_action=NotificationTopicAction.chat__missed_messages,
444 key="",
445 data=notification_data_pb2.ChatMissedMessages(
446 messages=[
447 notification_data_pb2.ChatMessage(
448 author=api_pb2.User(name="Test User", user_id=2, username="testuser"),
449 message="You missed 1 message(s) from Test User",
450 text="Hello!",
451 group_chat_id=99,
452 ),
453 ],
454 ),
455 )
457 assert mock.call_count == 1
458 e = email_fields(mock)
460 assert e.list_unsubscribe_header
462 # Extract the List-Unsubscribe URL and call the Unsubscribe endpoint
463 url = e.list_unsubscribe_header.strip("<>")
464 url_parts = urlparse(url)
465 params = parse_qs(url_parts.query)
467 with auth_api_session() as (auth_api, metadata_interceptor):
468 res = auth_api.Unsubscribe(
469 auth_pb2.UnsubscribeReq(
470 payload=b64decode(params["payload"][0]),
471 sig=b64decode(params["sig"][0]),
472 )
473 )
474 assert res.response
477def test_email_deleted_users_regression(db, moderator: Moderator):
478 """
479 We introduced a bug in notify v2 where we would email deleted/banned users.
480 """
481 super_user, super_token = generate_user(is_superuser=True)
482 creating_user, creating_token = generate_user(complete_profile=True)
484 normal_user, _ = generate_user()
485 ban_user, _ = generate_user()
486 delete_user, _ = generate_user()
488 with session_scope() as session:
489 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
490 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
491 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
492 c_id = create_community(
493 session,
494 0,
495 2,
496 "Non-global Community",
497 [super_user],
498 [creating_user, normal_user, ban_user, delete_user],
499 r,
500 ).id
502 enforce_community_memberships()
504 start_time = now() + timedelta(hours=2)
505 end_time = start_time + timedelta(hours=3)
506 with events_session(creating_token) as api:
507 res = api.CreateEvent(
508 events_pb2.CreateEventReq(
509 title="Dummy Title",
510 content="Dummy content.",
511 photo_key=None,
512 parent_community_id=c_id,
513 offline_information=events_pb2.OfflineEventInformation(
514 address="Near Null Island",
515 lat=0.1,
516 lng=0.2,
517 ),
518 start_time=Timestamp_from_datetime(start_time),
519 end_time=Timestamp_from_datetime(end_time),
520 timezone="UTC",
521 )
522 )
523 event_id = res.event_id
524 assert not res.is_deleted
526 moderator.approve_event_occurrence(event_id)
528 with events_session(creating_token) as api:
529 with mock_notification_email() as mock:
530 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
531 assert mock.call_count == 1
533 with real_editor_session(super_token) as editor:
534 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
535 assert len(res.requests) == 1
536 # this will count everyone
537 assert res.requests[0].approx_users_to_notify == 5
539 with session_scope() as session:
540 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
541 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
543 with real_editor_session(super_token) as editor:
544 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
545 assert len(res.requests) == 1
546 # should only notify creating_user, super_user and normal_user
547 assert res.requests[0].approx_users_to_notify == 3
549 with mock_notification_email() as mock:
550 editor.DecideEventCommunityInviteRequest(
551 editor_pb2.DecideEventCommunityInviteRequestReq(
552 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
553 approve=True,
554 )
555 )
557 assert mock.call_count == 3