Coverage for src/tests/test_email.py: 100%
234 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-12-20 18:03 +0000
1from unittest.mock import patch
3import pytest
5import couchers.email
6import couchers.jobs.handlers
7from couchers.config import config
8from couchers.crypto import random_hex, urlsafe_secure_token
9from couchers.db import session_scope
10from couchers.models import (
11 ContentReport,
12 Email,
13 FriendRelationship,
14 FriendStatus,
15 Reference,
16 ReferenceType,
17 SignupFlow,
18 User,
19)
20from couchers.notifications.notify import notify
21from couchers.sql import couchers_select as select
22from couchers.tasks import (
23 enforce_community_memberships,
24 maybe_send_reference_report_email,
25 send_content_report_email,
26 send_email_changed_confirmation_to_new_email,
27 send_signup_email,
28)
29from couchers.utils import Timestamp_from_datetime, now, timedelta
30from proto import admin_pb2, api_pb2, events_pb2, notification_data_pb2, notifications_pb2
31from tests.test_communities import create_community
32from tests.test_fixtures import ( # noqa
33 api_session,
34 db,
35 email_fields,
36 events_session,
37 generate_user,
38 mock_notification_email,
39 notifications_session,
40 process_jobs,
41 push_collector,
42 real_admin_session,
43 session_scope,
44 testconfig,
45)
48@pytest.fixture(autouse=True)
49def _(testconfig):
50 pass
53def test_signup_verification_email(db):
54 request_email = f"{random_hex(12)}@couchers.org.invalid"
56 flow = SignupFlow(name="Frodo", email=request_email)
58 with session_scope() as session:
59 with mock_notification_email() as mock:
60 send_signup_email(session, flow)
62 assert mock.call_count == 1
63 e = email_fields(mock)
64 assert e.recipient == request_email
65 assert flow.email_token in e.plain
66 assert flow.email_token in e.html
69def test_report_email(db):
70 user_reporter, api_token_author = generate_user()
71 user_author, api_token_reported = generate_user()
73 report = ContentReport(
74 reporting_user=user_reporter,
75 reason="spam",
76 description="I think this is spam and does not belong on couchers",
77 content_ref="comment/123",
78 author_user=user_author,
79 user_agent="n/a",
80 page="https://couchers.org/comment/123",
81 )
83 with session_scope() as session:
84 with mock_notification_email() as mock:
85 send_content_report_email(session, report)
87 assert mock.call_count == 1
89 e = email_fields(mock)
90 assert e.recipient == "reports@couchers.org.invalid"
91 assert report.author_user.username in e.plain
92 assert str(report.author_user.id) in e.plain
93 assert report.author_user.email in e.plain
94 assert report.reporting_user.username in e.plain
95 assert str(report.reporting_user.id) in e.plain
96 assert report.reporting_user.email in e.plain
97 assert report.reason in e.plain
98 assert report.description in e.plain
99 assert "report" in e.subject.lower()
102def test_reference_report_email_not_sent(db):
103 with session_scope() as session:
104 from_user, api_token_author = generate_user()
105 to_user, api_token_reported = generate_user()
107 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
108 session.add(friend_relationship)
109 session.flush()
111 reference = Reference(
112 from_user=from_user,
113 to_user=to_user,
114 reference_type=ReferenceType.friend,
115 text="This person was very nice to me.",
116 rating=0.9,
117 was_appropriate=True,
118 )
120 # no email sent for a positive ref
122 with mock_notification_email() as mock:
123 maybe_send_reference_report_email(session, reference)
125 assert mock.call_count == 0
128def test_reference_report_email(db):
129 with session_scope() as session:
130 from_user, api_token_author = generate_user()
131 to_user, api_token_reported = generate_user()
133 friend_relationship = FriendRelationship(from_user=from_user, to_user=to_user, status=FriendStatus.accepted)
134 session.add(friend_relationship)
135 session.flush()
137 reference = Reference(
138 from_user=from_user,
139 to_user=to_user,
140 reference_type=ReferenceType.friend,
141 text="This person was not nice to me.",
142 rating=0.3,
143 was_appropriate=False,
144 )
146 with mock_notification_email() as mock:
147 maybe_send_reference_report_email(session, reference)
149 assert mock.call_count == 1
150 e = email_fields(mock)
151 assert e.recipient == "reports@couchers.org.invalid"
152 assert "report" in e.subject.lower()
153 assert "reference" in e.subject.lower()
154 assert reference.from_user.username in e.plain
155 assert str(reference.from_user.id) in e.plain
156 assert reference.from_user.email in e.plain
157 assert reference.to_user.username in e.plain
158 assert str(reference.to_user.id) in e.plain
159 assert reference.to_user.email in e.plain
160 assert reference.text in e.plain
161 assert "friend" in e.plain.lower()
164def test_email_patching_fails(db):
165 """
166 There was a problem where the mocking wasn't happening and the email dev
167 printing function was called instead, this makes sure the patching is
168 actually done
169 """
170 to_user, to_token = generate_user()
171 from_user, from_token = generate_user()
173 patched_msg = random_hex(64)
175 def mock_queue_email(session, **kwargs):
176 raise Exception(patched_msg)
178 with patch("couchers.email._queue_email", mock_queue_email):
179 with pytest.raises(Exception) as e:
180 with api_session(from_token) as api:
181 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=to_user.id))
182 process_jobs()
184 assert str(e.value) == patched_msg
187def test_email_changed_confirmation_sent_to_new_email(db):
188 confirmation_token = urlsafe_secure_token()
189 user, user_token = generate_user()
190 user.new_email = f"{random_hex(12)}@couchers.org.invalid"
191 user.new_email_token = confirmation_token
192 with session_scope() as session:
193 with mock_notification_email() as mock:
194 send_email_changed_confirmation_to_new_email(session, user)
196 assert mock.call_count == 1
197 e = email_fields(mock)
198 assert "new email" in e.subject
199 assert e.recipient == user.new_email
200 assert user.name in e.plain
201 assert user.name in e.html
202 assert user.email in e.plain
203 assert user.email in e.html
204 assert "Your old email address is" in e.plain
205 assert "Your old email address is" in e.html
206 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.plain
207 assert f"http://localhost:3000/confirm-email?token={confirmation_token}" in e.html
208 assert "support@couchers.org" in e.plain
209 assert "support@couchers.org" in e.html
212def test_do_not_email_security(db):
213 user, token = generate_user()
215 password_reset_token = urlsafe_secure_token()
217 with notifications_session(token) as notifications:
218 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
220 # make sure we still get security emails
222 with mock_notification_email() as mock:
223 with session_scope() as session:
224 notify(
225 session,
226 user_id=user.id,
227 topic_action="password_reset:start",
228 data=notification_data_pb2.PasswordResetStart(
229 password_reset_token=password_reset_token,
230 ),
231 )
233 assert mock.call_count == 1
234 e = email_fields(mock)
235 assert e.recipient == user.email
236 assert "reset" in e.subject.lower()
237 assert password_reset_token in e.plain
238 assert password_reset_token in e.html
239 unique_string = "You asked for your password to be reset on Couchers.org."
240 assert unique_string in e.plain
241 assert unique_string in e.html
242 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.plain
243 assert f"http://localhost:3000/complete-password-reset?token={password_reset_token}" in e.html
244 assert "support@couchers.org" in e.plain
245 assert "support@couchers.org" in e.html
247 assert "/unsubscribe?payload=" not in e.plain
248 assert "/unsubscribe?payload=" not in e.html
251def test_do_not_email_non_security(db):
252 user, token1 = generate_user(complete_profile=True)
253 _, token2 = generate_user(complete_profile=True)
255 with notifications_session(token1) as notifications:
256 notifications.SetNotificationSettings(notifications_pb2.SetNotificationSettingsReq(enable_do_not_email=True))
258 with mock_notification_email() as mock:
259 with api_session(token2) as api:
260 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
262 assert mock.call_count == 0
265def test_do_not_email_non_security_unsublink(db):
266 user, _ = generate_user(complete_profile=True)
267 _, token2 = generate_user(complete_profile=True)
269 with mock_notification_email() as mock:
270 with api_session(token2) as api:
271 api.SendFriendRequest(api_pb2.SendFriendRequestReq(user_id=user.id))
273 assert mock.call_count == 1
274 e = email_fields(mock)
276 assert "/unsubscribe?payload=" in e.plain
277 assert "/unsubscribe?payload=" in e.html
280def test_email_prefix_config(db, monkeypatch):
281 user, _ = generate_user()
283 with mock_notification_email() as mock:
284 with session_scope() as session:
285 notify(
286 session,
287 user_id=user.id,
288 topic_action="donation:received",
289 data=notification_data_pb2.DonationReceived(
290 amount=20,
291 receipt_url="https://example.com/receipt/12345",
292 ),
293 )
295 assert mock.call_count == 1
296 e = email_fields(mock)
297 assert e.sender_name == "Couchers.org"
298 assert e.sender_email == "notify@couchers.org.invalid"
299 assert e.subject == "[TEST] Thank you for your donation to Couchers.org!"
301 new_config = config.copy()
302 new_config["NOTIFICATION_EMAIL_SENDER"] = "TestCo"
303 new_config["NOTIFICATION_EMAIL_ADDRESS"] = "testco@testing.co.invalid"
304 new_config["NOTIFICATION_PREFIX"] = ""
306 monkeypatch.setattr(couchers.notifications.background, "config", new_config)
308 with mock_notification_email() as mock:
309 with session_scope() as session:
310 notify(
311 session,
312 user_id=user.id,
313 topic_action="donation:received",
314 data=notification_data_pb2.DonationReceived(
315 amount=20,
316 receipt_url="https://example.com/receipt/12345",
317 ),
318 )
320 assert mock.call_count == 1
321 e = email_fields(mock)
322 assert e.sender_name == "TestCo"
323 assert e.sender_email == "testco@testing.co.invalid"
324 assert e.subject == "Thank you for your donation to Couchers.org!"
327def test_send_donation_email(db, monkeypatch):
328 user, _ = generate_user(name="Testy von Test", email="testing@couchers.org.invalid")
330 new_config = config.copy()
331 new_config["ENABLE_EMAIL"] = True
333 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config)
335 with session_scope() as session:
336 notify(
337 session,
338 user_id=user.id,
339 topic_action="donation:received",
340 data=notification_data_pb2.DonationReceived(
341 amount=20,
342 receipt_url="https://example.com/receipt/12345",
343 ),
344 )
346 with patch("couchers.email.smtp.smtplib.SMTP"):
347 process_jobs()
349 with session_scope() as session:
350 email = session.execute(select(Email)).scalar_one()
351 assert email.subject == "[TEST] Thank you for your donation to Couchers.org!"
352 assert (
353 email.plain
354 == """Dear Testy von Test,
356Thank you so much for your donation of $20 to Couchers.org.
358Your 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.
361You can download an invoice and receipt for the donation here:
363<https://example.com/receipt/12345>
365If you have any questions about your donation, please email us at <donations@couchers.org>.
367Your generosity will help deliver the platform for everyone.
370Thank you!
372Aapeli and Itsi,
373Couchers.org Founders
376---
378This is a security email, you cannot unsubscribe from it."""
379 )
381 assert "Thank you so much for your donation of $20 to Couchers.org." in email.html
382 assert email.sender_name == "Couchers.org"
383 assert email.sender_email == "notify@couchers.org.invalid"
384 assert email.recipient == "testing@couchers.org.invalid"
385 assert "https://example.com/receipt/12345" in email.html
386 assert not email.list_unsubscribe_header
387 assert email.source_data == "testing_version/donation_received"
390def test_email_deleted_users_regression(db):
391 """
392 We introduced a bug in notify v2 where we would email deleted/banned users.
393 """
394 super_user, super_token = generate_user(is_superuser=True)
395 creating_user, creating_token = generate_user(complete_profile=True)
397 normal_user, _ = generate_user()
398 ban_user, _ = generate_user()
399 delete_user, _ = generate_user()
401 with session_scope() as session:
402 create_community(session, 10, 2, "Global Community", [super_user], [], None)
403 create_community(
404 session,
405 0,
406 2,
407 "Non-global Community",
408 [super_user],
409 [creating_user, normal_user, ban_user, delete_user],
410 None,
411 )
413 enforce_community_memberships()
415 start_time = now() + timedelta(hours=2)
416 end_time = start_time + timedelta(hours=3)
417 with events_session(creating_token) as api:
418 res = api.CreateEvent(
419 events_pb2.CreateEventReq(
420 title="Dummy Title",
421 content="Dummy content.",
422 photo_key=None,
423 offline_information=events_pb2.OfflineEventInformation(
424 address="Near Null Island",
425 lat=0.1,
426 lng=0.2,
427 ),
428 start_time=Timestamp_from_datetime(start_time),
429 end_time=Timestamp_from_datetime(end_time),
430 timezone="UTC",
431 )
432 )
433 event_id = res.event_id
434 assert not res.is_deleted
436 with mock_notification_email() as mock:
437 api.RequestCommunityInvite(events_pb2.RequestCommunityInviteReq(event_id=event_id))
438 assert mock.call_count == 1
440 with real_admin_session(super_token) as admin:
441 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq())
442 assert len(res.requests) == 1
443 # this will count everyone
444 assert res.requests[0].approx_users_to_notify == 5
446 with session_scope() as session:
447 session.execute(select(User).where(User.id == ban_user.id)).scalar_one().is_banned = True
448 session.execute(select(User).where(User.id == delete_user.id)).scalar_one().is_deleted = True
450 with real_admin_session(super_token) as admin:
451 res = admin.ListEventCommunityInviteRequests(admin_pb2.ListEventCommunityInviteRequestsReq())
452 assert len(res.requests) == 1
453 # should only notify creating_user, super_user and normal_user
454 assert res.requests[0].approx_users_to_notify == 3
456 with mock_notification_email() as mock:
457 admin.DecideEventCommunityInviteRequest(
458 admin_pb2.DecideEventCommunityInviteRequestReq(
459 event_community_invite_request_id=res.requests[0].event_community_invite_request_id,
460 approve=True,
461 )
462 )
464 assert mock.call_count == 3