Coverage for app / backend / src / tests / test_email.py: 100%
285 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-05 09:44 +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>
405Couchers, 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.
407If you have any questions about your donation, please email us at <donations@couchers.org>.
409Your generosity will help deliver the platform for everyone.
412Thank you!
414Aapeli and Itsi,
415Couchers.org Founders
418---
420This is a security email, you cannot unsubscribe from it.
421"""
422 )
424 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
425 assert email.sender_name == "Couchers.org"
426 assert email.sender_email == "notify@couchers.org.invalid"
427 assert email.recipient == "testing@couchers.org.invalid"
428 assert "https://example.com/receipt/12345" in email.html
429 assert not email.list_unsubscribe_header
430 assert email.source_data == "testing_version/donation_received"
433def test_chat_missed_messages_list_unsubscribe_header(db):
434 """
435 Regression test: chat__missed_messages has key="" (it's a summary, not tied to a single chat).
436 The List-Unsubscribe header must use a topic_action unsubscribe link, not a topic_key link.
437 """
438 user, _ = generate_user()
440 with mock_notification_email() as mock:
441 with session_scope() as session:
442 notify(
443 session,
444 user_id=user.id,
445 topic_action=NotificationTopicAction.chat__missed_messages,
446 key="",
447 data=notification_data_pb2.ChatMissedMessages(
448 messages=[
449 notification_data_pb2.ChatMessage(
450 author=api_pb2.User(name="Test User", user_id=2, username="testuser"),
451 message="You missed 1 message(s) from Test User",
452 text="Hello!",
453 group_chat_id=99,
454 ),
455 ],
456 ),
457 )
459 assert mock.call_count == 1
460 e = email_fields(mock)
462 assert e.list_unsubscribe_header
464 # Extract the List-Unsubscribe URL and call the Unsubscribe endpoint
465 url = e.list_unsubscribe_header.strip("<>")
466 url_parts = urlparse(url)
467 params = parse_qs(url_parts.query)
469 with auth_api_session() as (auth_api, metadata_interceptor):
470 res = auth_api.Unsubscribe(
471 auth_pb2.UnsubscribeReq(
472 payload=b64decode(params["payload"][0]),
473 sig=b64decode(params["sig"][0]),
474 )
475 )
476 assert res.response
479def test_email_deleted_users_regression(db, moderator: Moderator):
480 """
481 We introduced a bug in notify v2 where we would email deleted/banned users.
482 """
483 super_user, super_token = generate_user(is_superuser=True)
484 creating_user, creating_token = generate_user(complete_profile=True)
486 normal_user, _ = generate_user()
487 ban_user, _ = generate_user()
488 delete_user, _ = generate_user()
490 with session_scope() as session:
491 w = create_community(session, 0, 2, "Global Community", [super_user], [], None)
492 mr = create_community(session, 0, 2, "Macroregion", [super_user], [], w)
493 r = create_community(session, 0, 2, "Region", [super_user], [], mr)
494 c_id = create_community(
495 session,
496 0,
497 2,
498 "Non-global Community",
499 [super_user],
500 [creating_user, normal_user, ban_user, delete_user],
501 r,
502 ).id
504 enforce_community_memberships()
506 start_time = now() + timedelta(hours=2)
507 end_time = start_time + timedelta(hours=3)
508 with events_session(creating_token) as api:
509 res = api.CreateEvent(
510 events_pb2.CreateEventReq(
511 title="Dummy Title",
512 content="Dummy content.",
513 photo_key=None,
514 parent_community_id=c_id,
515 offline_information=events_pb2.OfflineEventInformation(
516 address="Near Null Island",
517 lat=0.1,
518 lng=0.2,
519 ),
520 start_time=Timestamp_from_datetime(start_time),
521 end_time=Timestamp_from_datetime(end_time),
522 timezone="UTC",
523 )
524 )
525 event_id = res.event_id
526 assert not res.is_deleted
528 moderator.approve_event_occurrence(event_id)
530 with events_session(creating_token) as api:
531 with mock_notification_email() as mock:
532 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
533 assert mock.call_count == 1
535 with real_editor_session(super_token) as editor:
536 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
537 assert len(res.requests) == 1
538 # this will count everyone
539 assert res.requests[0].approx_users_to_notify == 5
541 with session_scope() as session:
542 session.execute(update(User).where(User.id == ban_user.id).values(banned_at=func.now()))
543 session.execute(update(User).where(User.id == delete_user.id).values(deleted_at=func.now()))
545 with real_editor_session(super_token) as editor:
546 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
547 assert len(res.requests) == 1
548 # should only notify creating_user, super_user and normal_user
549 assert res.requests[0].approx_users_to_notify == 3
551 with mock_notification_email() as mock:
552 editor.DecideEventCommunityInviteRequest(
553 editor_pb2.DecideEventCommunityInviteRequestReq(
554 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
555 approve=True,
556 )
557 )
559 assert mock.call_count == 3