Coverage for app / backend / src / tests / test_email.py: 100%
253 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-03 06:18 +0000
1from datetime import timedelta
2from unittest.mock import patch
4import pytest
5from sqlalchemy import select, update
7import couchers.email
8import couchers.jobs.handlers
9from couchers.config import config
10from couchers.crypto import random_hex, urlsafe_secure_token
11from couchers.db import session_scope
12from couchers.models import (
13 ContentReport,
14 Email,
15 FriendRelationship,
16 FriendStatus,
17 Reference,
18 ReferenceType,
19 SignupFlow,
20 User,
21)
22from couchers.models.notifications import NotificationTopicAction
23from couchers.notifications.notify import notify
24from couchers.proto import api_pb2, editor_pb2, events_pb2, notification_data_pb2, notifications_pb2
25from couchers.tasks import (
26 enforce_community_memberships,
27 maybe_send_reference_report_email,
28 send_content_report_email,
29 send_email_changed_confirmation_to_new_email,
30 send_signup_email,
31)
32from couchers.utils import Timestamp_from_datetime, now
33from tests.fixtures.db import generate_user
34from tests.fixtures.misc import email_fields, mock_notification_email, process_jobs
35from tests.fixtures.sessions import api_session, events_session, notifications_session, real_editor_session
36from tests.test_communities import create_community
39@pytest.fixture(autouse=True)
40def _(testconfig):
41 pass
44def test_signup_verification_email(db):
45 request_email = f"{random_hex(12)}@couchers.org.invalid"
47 flow = SignupFlow(name="Frodo", email=request_email, flow_token="")
49 with session_scope() as session:
50 with mock_notification_email() as mock:
51 send_signup_email(session, flow)
53 assert mock.call_count == 1
54 e = email_fields(mock)
55 assert e.recipient == request_email
56 assert flow.email_token
57 assert flow.email_token in e.html
58 assert flow.email_token in e.html
61def test_report_email(db):
62 user_reporter, api_token_author = generate_user()
63 user_author, api_token_reported = generate_user()
65 with session_scope() as session:
66 report = ContentReport(
67 reporting_user_id=user_reporter.id,
68 reason="spam",
69 description="I think this is spam and does not belong on couchers",
70 content_ref="comment/123",
71 author_user_id=user_author.id,
72 user_agent="n/a",
73 page="https://couchers.org/comment/123",
74 )
75 session.add(report)
76 session.flush()
78 with mock_notification_email() as mock:
79 send_content_report_email(session, report)
81 # Load all data before session closes
82 author_username = report.author_user.username
83 author_id = report.author_user.id
84 author_email = report.author_user.email
85 reporting_username = report.reporting_user.username
86 reporting_id = report.reporting_user.id
87 reporting_email = report.reporting_user.email
88 reason = report.reason
89 description = report.description
91 assert mock.call_count == 1
93 e = email_fields(mock)
94 assert e.recipient == "reports@couchers.org.invalid"
95 assert author_username in e.plain
96 assert str(author_id) in e.plain
97 assert author_email in e.plain
98 assert reporting_username in e.plain
99 assert str(reporting_id) in e.plain
100 assert reporting_email in e.plain
101 assert reason in e.plain
102 assert description in e.plain
103 assert "report" in e.subject.lower()
106def test_reference_report_email_not_sent(db):
107 with session_scope() as session:
108 from_user, api_token_author = generate_user()
109 to_user, api_token_reported = generate_user()
111 friend_relationship = FriendRelationship(
112 from_user_id=from_user.id, to_user_id=to_user.id, status=FriendStatus.accepted
113 )
114 session.add(friend_relationship)
115 session.flush()
117 reference = Reference(
118 from_user_id=from_user.id,
119 to_user_id=to_user.id,
120 reference_type=ReferenceType.friend,
121 text="This person was very nice to me.",
122 rating=0.9,
123 was_appropriate=True,
124 )
126 # no email sent for a positive ref
128 with mock_notification_email() as mock:
129 maybe_send_reference_report_email(session, reference)
131 assert mock.call_count == 0
134def test_reference_report_email(db):
135 with session_scope() as session:
136 from_user, api_token_author = generate_user()
137 to_user, api_token_reported = generate_user()
139 friend_relationship = FriendRelationship(
140 from_user_id=from_user.id, to_user_id=to_user.id, status=FriendStatus.accepted
141 )
142 session.add(friend_relationship)
143 session.flush()
145 reference = Reference(
146 from_user_id=from_user.id,
147 to_user_id=to_user.id,
148 reference_type=ReferenceType.friend,
149 text="This person was not nice to me.",
150 rating=0.3,
151 was_appropriate=False,
152 private_text="This is some private text for support",
153 )
154 session.add(reference)
155 session.flush()
157 with mock_notification_email() as mock:
158 maybe_send_reference_report_email(session, reference)
160 assert mock.call_count == 1
161 e = email_fields(mock)
162 assert e.recipient == "reports@couchers.org.invalid"
163 assert "report" in e.subject.lower()
164 assert "reference" in e.subject.lower()
165 assert reference.from_user.username in e.plain
166 assert str(reference.from_user.id) in e.plain
167 assert reference.from_user.email in e.plain
168 assert reference.to_user.username in e.plain
169 assert str(reference.to_user.id) in e.plain
170 assert reference.to_user.email in e.plain
171 assert reference.text in e.plain
172 assert "friend" in e.plain.lower()
173 assert reference.private_text
174 assert reference.private_text in e.plain
177def test_email_patching_fails(db):
178 """
179 There was a problem where the mocking wasn't happening and the email dev
180 printing function was called instead, this makes sure the patching is
181 actually done
182 """
183 to_user, to_token = generate_user()
184 from_user, from_token = generate_user()
186 patched_msg = random_hex(64)
188 def mock_queue_email(session, **kwargs):
189 raise Exception(patched_msg)
191 with patch("couchers.email._queue_email", mock_queue_email):
192 with pytest.raises(Exception) as e:
193 with api_session(from_token) as api:
194 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
195 process_jobs()
197 assert str(e.value) == patched_msg
200def test_email_changed_confirmation_sent_to_new_email(db):
201 confirmation_token = urlsafe_secure_token()
202 user, user_token = generate_user()
203 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
204 user.new_email_token = confirmation_token
205 with session_scope() as session:
206 with mock_notification_email() as mock:
207 send_email_changed_confirmation_to_new_email(session, user)
209 assert mock.call_count == 1
210 e = email_fields(mock)
211 assert "new email" in e.subject
212 assert e.recipient == user.new_email
213 assert user.name in e.plain
214 assert user.name in e.html
215 assert user.email in e.plain
216 assert user.email in e.html
217 assert "Your old email address is" in e.plain
218 assert "Your old email address is" in e.html
219 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
220 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
221 assert "support@couchers.org" in e.plain
222 assert "support@couchers.org" in e.html
225def test_do_not_email_security(db):
226 user, token = generate_user()
228 password_reset_token = urlsafe_secure_token()
230 with notifications_session(token) as notifications:
231 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
233 # make sure we still get security emails
235 with mock_notification_email() as mock:
236 with session_scope() as session:
237 notify(
238 session,
239 user_id=user.id,
240 topic_action=NotificationTopicAction.password_reset__start,
241 key="",
242 data=notification_data_pb2.PasswordResetStart(
243 password_reset_token=password_reset_token,
244 ),
245 )
247 assert mock.call_count == 1
248 e = email_fields(mock)
249 assert e.recipient == user.email
250 assert "reset" in e.subject.lower()
251 assert password_reset_token in e.plain
252 assert password_reset_token in e.html
253 unique_string = "You asked for your password to be reset on Couchers.org."
254 assert unique_string in e.plain
255 assert unique_string in e.html
256 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
257 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
258 assert "support@couchers.org" in e.plain
259 assert "support@couchers.org" in e.html
261 assert "/quick-link?payload=" not in e.plain
262 assert "/quick-link?payload=" not in e.html
265def test_do_not_email_non_security(db):
266 user, token1 = generate_user(complete_profile=True)
267 _, token2 = generate_user(complete_profile=True)
269 with notifications_session(token1) as notifications:
270 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
272 with mock_notification_email() as mock:
273 with api_session(token2) as api:
274 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
276 assert mock.call_count == 0
279def test_do_not_email_non_security_unsublink(db):
280 user, _ = generate_user(complete_profile=True)
281 _, token2 = generate_user(complete_profile=True)
283 with mock_notification_email() as mock:
284 with api_session(token2) as api:
285 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
287 assert mock.call_count == 1
288 e = email_fields(mock)
290 assert "/quick-link?payload=" in e.plain
291 assert "/quick-link?payload=" in e.html
294def test_email_prefix_config(db, monkeypatch):
295 user, _ = generate_user()
297 with mock_notification_email() as mock:
298 with session_scope() as session:
299 notify(
300 session,
301 user_id=user.id,
302 topic_action=NotificationTopicAction.donation__received,
303 key="",
304 data=notification_data_pb2.DonationReceived(
305 amount=20,
306 receipt_url="https://example.com/receipt/12345",
307 ),
308 )
310 assert mock.call_count == 1
311 e = email_fields(mock)
312 assert e.sender_name == "Couchers.org"
313 assert e.sender_email == "notify@couchers.org.invalid"
314 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
316 new_config = config.copy()
317 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
318 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
319 new_config["NOTIFICATION_PREFIX"] = ""
321 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
323 with mock_notification_email() as mock:
324 with session_scope() as session:
325 notify(
326 session,
327 user_id=user.id,
328 topic_action=NotificationTopicAction.donation__received,
329 key="",
330 data=notification_data_pb2.DonationReceived(
331 amount=20,
332 receipt_url="https://example.com/receipt/12345",
333 ),
334 )
336 assert mock.call_count == 1
337 e = email_fields(mock)
338 assert e.sender_name == "TestCo"
339 assert e.sender_email == "testco@testing.co.invalid"
340 assert e.subject == "Thank you for your donation to Couchers.org!"
343def test_send_donation_email(db, monkeypatch):
344 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
346 new_config = config.copy()
347 new_config["ENABLE_EMAIL"] = True
349 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
351 with session_scope() as session:
352 notify(
353 session,
354 user_id=user.id,
355 topic_action=NotificationTopicAction.donation__received,
356 key="",
357 data=notification_data_pb2.DonationReceived(
358 amount=20,
359 receipt_url="https://example.com/receipt/12345",
360 ),
361 )
363 with patch("couchers.email.smtp.smtplib.SMTP"):
364 process_jobs()
366 with session_scope() as session:
367 email = session.execute(select(Email)).scalar_one()
368 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
369 assert (
370 email.plain
371 == """Dear Testy von Test,
373Thank you so much for your donation of $20 to Couchers.org.
375Your 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.
378You can download an invoice and receipt for the donation here:
380<https://example.com/receipt/12345>
382If you have any questions about your donation, please email us at <donations@couchers.org>.
384Your generosity will help deliver the platform for everyone.
387Thank you!
389Aapeli and Itsi,
390Couchers.org Founders
393---
395This is a security email, you cannot unsubscribe from it.
396"""
397 )
399 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
400 assert email.sender_name == "Couchers.org"
401 assert email.sender_email == "notify@couchers.org.invalid"
402 assert email.recipient == "testing@couchers.org.invalid"
403 assert "https://example.com/receipt/12345" in email.html
404 assert not email.list_unsubscribe_header
405 assert email.source_data == "testing_version/donation_received"
408def test_email_deleted_users_regression(db):
409 """
410 We introduced a bug in notify v2 where we would email deleted/banned users.
411 """
412 super_user, super_token = generate_user(is_superuser=True)
413 creating_user, creating_token = generate_user(complete_profile=True)
415 normal_user, _ = generate_user()
416 ban_user, _ = generate_user()
417 delete_user, _ = generate_user()
419 with session_scope() as session:
420 create_community(session, 10, 2, "Global Community", [super_user], [], None)
421 create_community(
422 session,
423 0,
424 2,
425 "Non-global Community",
426 [super_user],
427 [creating_user, normal_user, ban_user, delete_user],
428 None,
429 )
431 enforce_community_memberships()
433 start_time = now() + timedelta(hours=2)
434 end_time = start_time + timedelta(hours=3)
435 with events_session(creating_token) as api:
436 res = api.CreateEvent(
437 events_pb2.CreateEventReq(
438 title="Dummy Title",
439 content="Dummy content.",
440 photo_key=None,
441 offline_information=events_pb2.OfflineEventInformation(
442 address="Near Null Island",
443 lat=0.1,
444 lng=0.2,
445 ),
446 start_time=Timestamp_from_datetime(start_time),
447 end_time=Timestamp_from_datetime(end_time),
448 timezone="UTC",
449 )
450 )
451 event_id = res.event_id
452 assert not res.is_deleted
454 with mock_notification_email() as mock:
455 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
456 assert mock.call_count == 1
458 with real_editor_session(super_token) as editor:
459 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
460 assert len(res.requests) == 1
461 # this will count everyone
462 assert res.requests[0].approx_users_to_notify == 5
464 with session_scope() as session:
465 session.execute(update(User).where(User.id == ban_user.id).values(is_banned=True))
466 session.execute(update(User).where(User.id == delete_user.id).values(is_deleted=True))
468 with real_editor_session(super_token) as editor:
469 res = editor.ListEventCommunityInviteRequests(editor_pb2.ListEventCommunityInviteRequestsReq())
470 assert len(res.requests) == 1
471 # should only notify creating_user, super_user and normal_user
472 assert res.requests[0].approx_users_to_notify == 3
474 with mock_notification_email() as mock:
475 editor.DecideEventCommunityInviteRequest(
476 editor_pb2.DecideEventCommunityInviteRequestReq(
477 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
478 approve=True,
479 )
480 )
482 assert mock.call_count == 3