Coverage for app/backend/src/tests/test_account.py: 100%
788 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1from datetime import UTC, date, datetime, timedelta
2from unittest.mock import patch
4import grpc
5import pytest
6from google.protobuf import empty_pb2, wrappers_pb2
7from sqlalchemy import select, update
8from sqlalchemy.sql import func
10from couchers import urls
11from couchers.crypto import hash_password, random_hex
12from couchers.db import session_scope
13from couchers.materialized_views import refresh_materialized_views_rapid
14from couchers.models import (
15 AccountDeletionReason,
16 AccountDeletionToken,
17 BackgroundJob,
18 HostingStatus,
19 InviteCode,
20 PhotoGalleryItem,
21 SleepingArrangement,
22 Upload,
23 User,
24)
25from couchers.proto import account_pb2, api_pb2, auth_pb2, messages_pb2, requests_pb2
26from couchers.utils import now, today
27from tests.fixtures.db import backdate_conversations, generate_user, make_volunteer
28from tests.fixtures.misc import EmailCollector, PushCollector, process_jobs
29from tests.fixtures.sessions import (
30 account_session,
31 auth_api_session,
32 public_session,
33 real_account_session,
34 requests_session,
35)
36from tests.fixtures.timewarp import Timewarp
37from tests.test_references import create_public_trip
38from tests.test_requests import valid_request_text
41def test_GetAccountInfo(db, fast_passwords):
42 # with password
43 user1, token1 = generate_user(hashed_password=hash_password(random_hex()), email="user@couchers.invalid")
45 with account_session(token1) as account:
46 res = account.GetAccountInfo(empty_pb2.Empty())
47 assert res.email == "user@couchers.invalid"
48 assert res.username == user1.username
49 assert not res.has_strong_verification
50 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED
51 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED
52 assert not res.is_superuser
53 assert res.ui_language_preference == ""
54 assert not res.is_volunteer
57def test_donation_banner_no_drive(db):
58 """Test that the banner is not shown when no drive is configured (flag unset)"""
59 # User has donated, but there's no drive, so the banner should not show
60 user, token = generate_user()
62 with account_session(token) as account:
63 res = account.GetAccountInfo(empty_pb2.Empty())
64 assert not res.should_show_donation_banner
67def test_donation_banner_never_donated(db, feature_flags):
68 """Test that banner is shown when user has never donated and drive is active"""
69 # Explicitly set last_donated=None since generate_user defaults to now()
70 user, token = generate_user(last_donated=None)
72 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
73 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
74 with account_session(token) as account:
75 res = account.GetAccountInfo(empty_pb2.Empty())
76 assert res.should_show_donation_banner
79def test_donation_banner_donated_before_drive(db, feature_flags):
80 """Test that banner is shown when user donated before drive start"""
81 user, token = generate_user()
83 # Set donation before drive start
84 with session_scope() as session:
85 last_donated = datetime(2025, 10, 15, tzinfo=UTC) # Before Nov 1
86 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
88 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
89 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
90 with account_session(token) as account:
91 res = account.GetAccountInfo(empty_pb2.Empty())
92 assert res.should_show_donation_banner
95def test_donation_banner_donated_after_drive(db, feature_flags):
96 """Test that banner is not shown when user donated after drive start"""
97 user, token = generate_user()
99 # Set donation after drive start
100 with session_scope() as session:
101 last_donated = datetime(2025, 11, 15, tzinfo=UTC) # After Nov 1
102 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
104 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
105 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
106 with account_session(token) as account:
107 res = account.GetAccountInfo(empty_pb2.Empty())
108 assert not res.should_show_donation_banner
111def test_donation_banner_donated_exactly_at_drive_start(db, feature_flags):
112 """Test that banner is not shown when user donated exactly at drive start time"""
113 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
115 user, token = generate_user()
117 # Set donation exactly at drive start
118 with session_scope() as session:
119 session.execute(update(User).where(User.id == user.id).values(last_donated=drive_start))
121 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
122 with account_session(token) as account:
123 res = account.GetAccountInfo(empty_pb2.Empty())
124 assert not res.should_show_donation_banner
127def test_GetAccountInfo_regression(db):
128 # there was a bug in evaluating `has_completed_profile` on the backend (in python)
129 # when about_me is None but the user has a key, it was failing because len(about_me) doesn't work on None
130 user, token = generate_user(about_me=None, complete_profile=False)
132 # add an avatar photo to the user's profile gallery
133 with session_scope() as session:
134 key = random_hex(32)
135 filename = random_hex(32) + ".jpg"
136 session.add(
137 Upload(
138 key=key,
139 filename=filename,
140 creator_user_id=user.id,
141 )
142 )
143 session.flush()
144 assert user.profile_gallery_id is not None
145 session.add(
146 PhotoGalleryItem(
147 gallery_id=user.profile_gallery_id,
148 upload_key=key,
149 position=0,
150 )
151 )
153 with account_session(token) as account:
154 res = account.GetAccountInfo(empty_pb2.Empty())
157def test_ChangePasswordV2_normal(db, fast_passwords, email_collector: EmailCollector, push_collector: PushCollector):
158 # user has old password and is changing to new password
159 old_password = random_hex()
160 new_password = random_hex()
161 user, token = generate_user(hashed_password=hash_password(old_password))
163 with account_session(token) as account:
164 account.ChangePasswordV2(
165 account_pb2.ChangePasswordV2Req(
166 old_password=old_password,
167 new_password=new_password,
168 )
169 )
171 email = email_collector.pop_for_recipient(user.email, last=True)
172 assert email.subject == "[TEST] Your password was changed"
174 push = push_collector.pop_for_user(user.id, last=True)
175 assert push.content.title == "Password changed"
176 assert push.content.body == "Your password was changed."
178 with session_scope() as session:
179 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
180 assert updated_user.hashed_password == hash_password(new_password)
183def test_ChangePasswordV2_regression(db, fast_passwords):
184 # send_password_changed_email wasn't working
185 # user has old password and is changing to new password
186 old_password = random_hex()
187 new_password = random_hex()
188 user, token = generate_user(hashed_password=hash_password(old_password))
190 with account_session(token) as account:
191 account.ChangePasswordV2(
192 account_pb2.ChangePasswordV2Req(
193 old_password=old_password,
194 new_password=new_password,
195 )
196 )
198 with session_scope() as session:
199 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
200 assert updated_user.hashed_password == hash_password(new_password)
203def test_ChangePasswordV2_normal_short_password(db, fast_passwords):
204 # user has old password and is changing to new password, but used short password
205 old_password = random_hex()
206 new_password = random_hex(length=1)
207 user, token = generate_user(hashed_password=hash_password(old_password))
209 with account_session(token) as account:
210 with pytest.raises(grpc.RpcError) as e:
211 account.ChangePasswordV2(
212 account_pb2.ChangePasswordV2Req(
213 old_password=old_password,
214 new_password=new_password,
215 )
216 )
217 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
218 assert e.value.details() == "The password must be 8 or more characters long."
220 with session_scope() as session:
221 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
222 assert updated_user.hashed_password == hash_password(old_password)
225def test_ChangePasswordV2_normal_long_password(db, fast_passwords):
226 # user has old password and is changing to new password, but used short password
227 old_password = random_hex()
228 new_password = random_hex(length=1000)
229 user, token = generate_user(hashed_password=hash_password(old_password))
231 with account_session(token) as account:
232 with pytest.raises(grpc.RpcError) as e:
233 account.ChangePasswordV2(
234 account_pb2.ChangePasswordV2Req(
235 old_password=old_password,
236 new_password=new_password,
237 )
238 )
239 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
240 assert e.value.details() == "The password must be less than 256 characters."
242 with session_scope() as session:
243 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
244 assert updated_user.hashed_password == hash_password(old_password)
247def test_ChangePasswordV2_normal_insecure_password(db, fast_passwords):
248 # user has old password and is changing to new password, but used insecure password
249 old_password = random_hex()
250 new_password = "12345678"
251 user, token = generate_user(hashed_password=hash_password(old_password))
253 with account_session(token) as account:
254 with pytest.raises(grpc.RpcError) as e:
255 account.ChangePasswordV2(
256 account_pb2.ChangePasswordV2Req(
257 old_password=old_password,
258 new_password=new_password,
259 )
260 )
261 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
262 assert e.value.details() == "The password is insecure. Please use one that is not easily guessable."
264 with session_scope() as session:
265 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
266 assert updated_user.hashed_password == hash_password(old_password)
269def test_ChangePasswordV2_normal_wrong_password(db, fast_passwords):
270 # user has old password and is changing to new password, but used wrong old password
271 old_password = random_hex()
272 new_password = random_hex()
273 user, token = generate_user(hashed_password=hash_password(old_password))
275 with account_session(token) as account:
276 with pytest.raises(grpc.RpcError) as e:
277 account.ChangePasswordV2(
278 account_pb2.ChangePasswordV2Req(
279 old_password="Wrong password",
280 new_password=new_password,
281 )
282 )
283 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
284 assert e.value.details() == "Wrong password."
286 with session_scope() as session:
287 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
288 assert updated_user.hashed_password == hash_password(old_password)
291def test_ChangePasswordV2_normal_no_passwords(db, fast_passwords):
292 # user has old password and called with empty body
293 old_password = random_hex()
294 user, token = generate_user(hashed_password=hash_password(old_password))
296 with account_session(token) as account:
297 with pytest.raises(grpc.RpcError) as e:
298 account.ChangePasswordV2(account_pb2.ChangePasswordV2Req(old_password=old_password))
299 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
300 assert e.value.details() == "The password must be 8 or more characters long."
302 with session_scope() as session:
303 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
304 assert updated_user.hashed_password == hash_password(old_password)
307def test_ChangeEmailV2_wrong_password(db, fast_passwords):
308 password = random_hex()
309 new_email = f"{random_hex()}@couchers.org.invalid"
310 user, token = generate_user(hashed_password=hash_password(password))
312 with account_session(token) as account:
313 with pytest.raises(grpc.RpcError) as e:
314 account.ChangeEmailV2(
315 account_pb2.ChangeEmailV2Req(
316 password="Wrong password",
317 new_email=new_email,
318 )
319 )
320 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
321 assert e.value.details() == "Wrong password."
323 with session_scope() as session:
324 assert (
325 session.execute(
326 select(func.count())
327 .select_from(User)
328 .where(User.new_email_token_created <= func.now())
329 .where(User.new_email_token_expiry >= func.now())
330 )
331 ).scalar_one() == 0
334def test_ChangeEmailV2_wrong_email(db, fast_passwords):
335 password = random_hex()
336 new_email = f"{random_hex()}@couchers.org.invalid"
337 user, token = generate_user(hashed_password=hash_password(password))
339 with account_session(token) as account:
340 with pytest.raises(grpc.RpcError) as e:
341 account.ChangeEmailV2(
342 account_pb2.ChangeEmailV2Req(
343 password="Wrong password",
344 new_email=new_email,
345 )
346 )
347 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
348 assert e.value.details() == "Wrong password."
350 with session_scope() as session:
351 assert (
352 session.execute(
353 select(func.count())
354 .select_from(User)
355 .where(User.new_email_token_created <= func.now())
356 .where(User.new_email_token_expiry >= func.now())
357 )
358 ).scalar_one() == 0
361def test_ChangeEmailV2_invalid_email(db, fast_passwords):
362 password = random_hex()
363 user, token = generate_user(hashed_password=hash_password(password))
365 with account_session(token) as account:
366 with pytest.raises(grpc.RpcError) as e:
367 account.ChangeEmailV2(
368 account_pb2.ChangeEmailV2Req(
369 password=password,
370 new_email="not a real email",
371 )
372 )
373 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
374 assert e.value.details() == "Invalid email."
376 with session_scope() as session:
377 assert (
378 session.execute(
379 select(func.count())
380 .select_from(User)
381 .where(User.new_email_token_created <= func.now())
382 .where(User.new_email_token_expiry >= func.now())
383 )
384 ).scalar_one() == 0
387def test_ChangeEmailV2_email_in_use(db, fast_passwords):
388 password = random_hex()
389 user, token = generate_user(hashed_password=hash_password(password))
390 user2, token2 = generate_user(hashed_password=hash_password(password))
392 with account_session(token) as account:
393 with pytest.raises(grpc.RpcError) as e:
394 account.ChangeEmailV2(
395 account_pb2.ChangeEmailV2Req(
396 password=password,
397 new_email=user2.email,
398 )
399 )
400 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
401 assert e.value.details() == "Invalid email."
403 with session_scope() as session:
404 assert (
405 session.execute(
406 select(func.count())
407 .select_from(User)
408 .where(User.new_email_token_created <= func.now())
409 .where(User.new_email_token_expiry >= func.now())
410 )
411 ).scalar_one() == 0
414def test_ChangeEmailV2_deleted_user_email(db, fast_passwords):
415 password = random_hex()
416 user, token = generate_user(hashed_password=hash_password(password))
417 deleted_user, _ = generate_user(delete_user=True)
418 banned_user, _ = generate_user()
420 with session_scope() as session:
421 session.execute(
422 update(User).where(User.id == banned_user.id).values(deleted_at=func.now(), banned_at=func.now())
423 )
425 with account_session(token) as account:
426 with pytest.raises(grpc.RpcError) as e:
427 account.ChangeEmailV2(account_pb2.ChangeEmailV2Req(password=password, new_email=banned_user.email))
428 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
430 account.ChangeEmailV2(account_pb2.ChangeEmailV2Req(password=password, new_email=deleted_user.email))
433def test_ChangeEmailV2_no_change(db, fast_passwords):
434 password = random_hex()
435 user, token = generate_user(hashed_password=hash_password(password))
437 with account_session(token) as account:
438 with pytest.raises(grpc.RpcError) as e:
439 account.ChangeEmailV2(
440 account_pb2.ChangeEmailV2Req(
441 password=password,
442 new_email=user.email,
443 )
444 )
445 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
446 assert e.value.details() == "Invalid email."
448 with session_scope() as session:
449 assert (
450 session.execute(
451 select(func.count())
452 .select_from(User)
453 .where(User.new_email_token_created <= func.now())
454 .where(User.new_email_token_expiry >= func.now())
455 )
456 ).scalar_one() == 0
459def test_ChangeEmailV2_wrong_token(db, fast_passwords):
460 password = random_hex()
461 new_email = f"{random_hex()}@couchers.org.invalid"
462 user, token = generate_user(hashed_password=hash_password(password))
464 with account_session(token) as account:
465 account.ChangeEmailV2(
466 account_pb2.ChangeEmailV2Req(
467 password=password,
468 new_email=new_email,
469 )
470 )
472 with auth_api_session() as (auth_api, metadata_interceptor):
473 with pytest.raises(grpc.RpcError) as e:
474 res = auth_api.ConfirmChangeEmailV2(
475 auth_pb2.ConfirmChangeEmailV2Req(
476 change_email_token="wrongtoken",
477 )
478 )
479 assert e.value.code() == grpc.StatusCode.NOT_FOUND
480 assert e.value.details() == "Invalid token."
482 with session_scope() as session:
483 user_updated = session.execute(select(User).where(User.id == user.id)).scalar_one()
484 assert user_updated.email == user.email
487def test_ChangeEmailV2_tokens_two_hour_window(db, fast_passwords, timewarp: Timewarp):
488 password = random_hex()
489 new_email = f"{random_hex()}@couchers.org.invalid"
490 user, token = generate_user(hashed_password=hash_password(password))
492 with account_session(token) as account:
493 account.ChangeEmailV2(
494 account_pb2.ChangeEmailV2Req(
495 password=password,
496 new_email=new_email,
497 )
498 )
500 with session_scope() as session:
501 new_email_token = session.execute(select(User.new_email_token).where(User.id == user.id)).scalar_one()
503 # before the token was issued
504 timewarp.advance(-timedelta(minutes=1))
505 with auth_api_session() as (auth_api, metadata_interceptor):
506 with pytest.raises(grpc.RpcError) as e:
507 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
508 assert e.value.code() == grpc.StatusCode.NOT_FOUND
509 assert e.value.details() == "Invalid token."
511 with pytest.raises(grpc.RpcError) as e:
512 auth_api.ConfirmChangeEmailV2(
513 auth_pb2.ConfirmChangeEmailV2Req(
514 change_email_token=new_email_token,
515 )
516 )
517 assert e.value.code() == grpc.StatusCode.NOT_FOUND
518 assert e.value.details() == "Invalid token."
520 # and a minute after the two hour window closed
521 timewarp.advance(timedelta(hours=2, minutes=1))
522 with auth_api_session() as (auth_api, metadata_interceptor):
523 with pytest.raises(grpc.RpcError) as e:
524 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
525 assert e.value.code() == grpc.StatusCode.NOT_FOUND
526 assert e.value.details() == "Invalid token."
528 with pytest.raises(grpc.RpcError) as e:
529 auth_api.ConfirmChangeEmailV2(
530 auth_pb2.ConfirmChangeEmailV2Req(
531 change_email_token=new_email_token,
532 )
533 )
534 assert e.value.code() == grpc.StatusCode.NOT_FOUND
535 assert e.value.details() == "Invalid token."
538def test_ChangeEmailV2(db, fast_passwords, push_collector: PushCollector):
539 password = random_hex()
540 new_email = f"{random_hex()}@couchers.org.invalid"
541 user, token = generate_user(hashed_password=hash_password(password))
542 user_id = user.id
544 with account_session(token) as account:
545 account.ChangeEmailV2(
546 account_pb2.ChangeEmailV2Req(
547 password=password,
548 new_email=new_email,
549 )
550 )
552 with session_scope() as session:
553 user_updated = session.execute(select(User).where(User.id == user_id)).scalar_one()
554 assert user_updated.email == user.email
555 assert user_updated.new_email == new_email
556 assert user_updated.new_email_token is not None
557 assert user_updated.new_email_token_created
558 assert user_updated.new_email_token_created <= now()
559 assert user_updated.new_email_token_expiry
560 assert user_updated.new_email_token_expiry >= now()
562 token = user_updated.new_email_token
564 process_jobs()
565 push = push_collector.pop_for_user(user_id, last=True)
566 assert push.content.title == "Email change requested"
567 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
569 with auth_api_session() as (auth_api, metadata_interceptor):
570 auth_api.ConfirmChangeEmailV2(
571 auth_pb2.ConfirmChangeEmailV2Req(
572 change_email_token=token,
573 )
574 )
576 with session_scope() as session:
577 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
578 assert user.email == new_email
579 assert user.new_email is None
580 assert user.new_email_token is None
581 assert user.new_email_token_created is None
582 assert user.new_email_token_expiry is None
584 process_jobs()
585 push = push_collector.pop_for_user(user_id, last=True)
586 assert push.content.title == "Email verified"
587 assert push.content.body == "Your new email address has been verified."
590def test_ChangeEmailV2_sends_proper_emails(db, fast_passwords, push_collector: PushCollector):
591 password = random_hex()
592 new_email = f"{random_hex()}@couchers.org.invalid"
593 user, token = generate_user(hashed_password=hash_password(password))
595 with account_session(token) as account:
596 account.ChangeEmailV2(
597 account_pb2.ChangeEmailV2Req(
598 password=password,
599 new_email=new_email,
600 )
601 )
603 process_jobs()
605 with session_scope() as session:
606 jobs = session.execute(select(BackgroundJob).where(BackgroundJob.job_type == "send_email")).scalars().all()
607 assert len(jobs) == 2
608 uq_str1 = b"Email address change initiated"
609 uq_str2 = b"You requested that your email be changed from"
610 assert (uq_str1 in jobs[0].payload and uq_str2 in jobs[1].payload) or (
611 uq_str2 in jobs[0].payload and uq_str1 in jobs[1].payload
612 )
614 push = push_collector.pop_for_user(user.id, last=True)
615 assert push.content.title == "Email change requested"
616 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
619def test_ChangeLanguagePreference(db, fast_passwords):
620 # user changes from default to ISO 639-1 language code
621 new_lang = "zh"
622 user, token = generate_user()
624 with real_account_session(token) as account:
625 res = account.GetAccountInfo(empty_pb2.Empty())
626 assert res.ui_language_preference == ""
628 # call will have info about the request
629 res, call = account.ChangeLanguagePreference.with_call(
630 account_pb2.ChangeLanguagePreferenceReq(ui_language_preference=new_lang)
631 )
633 # cookies are sent via initial metadata, so we check for it there
634 # the value of "set-cookie" will be the full cookie string, pull the key value from the string
635 cookie_values = [v.split(";")[0] for k, v in call.initial_metadata() if k == "set-cookie"]
636 assert any(val == "NEXT_LOCALE=zh" for val in cookie_values), (
637 f"Didn't find the right cookie, got {call.initial_metadata()}"
638 )
640 # the changed language preference should also be sent to the backend
641 res = account.GetAccountInfo(empty_pb2.Empty())
642 assert res.ui_language_preference == "zh"
645def test_contributor_form(db):
646 user, token = generate_user()
648 with account_session(token) as account:
649 res = account.GetContributorFormInfo(empty_pb2.Empty())
650 assert not res.filled_contributor_form
652 account.FillContributorForm(account_pb2.FillContributorFormReq(contributor_form=auth_pb2.ContributorForm()))
654 res = account.GetContributorFormInfo(empty_pb2.Empty())
655 assert res.filled_contributor_form
658def test_DeleteAccount_start(db, email_collector: EmailCollector):
659 user, token = generate_user()
661 with account_session(token) as account:
662 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None))
663 email = email_collector.pop_for_recipient(user.email, last=True)
664 assert email.subject == "[TEST] Confirm your account deletion"
666 with session_scope() as session:
667 deletion_token: AccountDeletionToken = session.execute(
668 select(AccountDeletionToken).where(AccountDeletionToken.user_id == user.id)
669 ).scalar_one()
671 assert deletion_token.is_valid
672 assert session.execute(select(User).where(User.id == user.id)).scalar_one().deleted_at is None
675def test_DeleteAccount_message_storage(db):
676 user, token = generate_user()
678 with account_session(token) as account:
679 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None)) # not stored
680 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="")) # not stored
681 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="Reason"))
682 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="0192#(&!&#)*@//)(8"))
683 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="\n\n\t")) # not stored
684 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="1337"))
686 with session_scope() as session:
687 assert session.execute(select(func.count()).select_from(AccountDeletionReason)).scalar_one() == 3
690def test_full_delete_account_with_recovery(db, email_collector: EmailCollector, push_collector: PushCollector):
691 user, token = generate_user()
692 user_id = user.id
694 with account_session(token) as account:
695 with pytest.raises(grpc.RpcError) as err:
696 account.DeleteAccount(account_pb2.DeleteAccountReq())
697 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
698 assert err.value.details() == "Please confirm your account deletion."
700 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
702 email = email_collector.pop_for_recipient(user.email, last=True)
703 assert email.subject == "[TEST] Confirm your account deletion"
704 assert email.recipient == user.email
705 assert "account deletion" in email.subject.lower()
706 unique_string = "You requested that we delete your Couchers.org account."
707 assert unique_string in email.plain
708 assert unique_string in email.html
709 assert "support@couchers.org" in email.plain
710 assert "support@couchers.org" in email.html
712 push = push_collector.pop_for_user(user_id, last=True)
713 assert push.content.title == "Account deletion requested"
714 assert push.content.body == "Use the link we emailed you to confirm."
716 with session_scope() as session:
717 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
718 delete_token = token_o.token
720 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
721 assert token_o.user == user_
722 assert user_.deleted_at is None
723 assert not user_.undelete_token
724 assert not user_.undelete_until
726 assert delete_token in email.plain
727 assert delete_token in email.html
728 delete_url = f"http://localhost:3000/delete-account?token={delete_token}"
729 assert delete_url in email.plain
730 assert delete_url in email.html
732 with auth_api_session() as (auth_api, metadata_interceptor):
733 auth_api.ConfirmDeleteAccount(
734 auth_pb2.ConfirmDeleteAccountReq(
735 token=delete_token,
736 )
737 )
739 email = email_collector.pop_for_recipient(user.email, last=True)
740 assert email.recipient == user.email
741 assert "account has been deleted" in email.subject.lower()
742 unique_string = "You have successfully deleted your Couchers.org account."
743 assert unique_string in email.plain
744 assert unique_string in email.html
745 assert "7 days" in email.plain
746 assert "7 days" in email.html
747 assert "support@couchers.org" in email.plain
748 assert "support@couchers.org" in email.html
750 push = push_collector.pop_for_user(user_id, last=True)
751 assert push.content.title == "Account deleted"
752 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
754 with session_scope() as session:
755 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
757 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
758 assert user_.deleted_at is not None
759 assert user_.undelete_token
760 assert user_.undelete_until
761 assert user_.undelete_until > now()
763 undelete_token = user_.undelete_token
765 undelete_url = f"http://localhost:3000/recover-account?token={undelete_token}"
766 assert undelete_url in email.plain
767 assert undelete_url in email.html
769 with auth_api_session() as (auth_api, metadata_interceptor):
770 auth_api.RecoverAccount(
771 auth_pb2.RecoverAccountReq(
772 token=undelete_token,
773 )
774 )
776 email = email_collector.pop_for_recipient(user.email, last=True)
777 assert email.recipient == user.email
778 assert "account has been recovered" in email.subject.lower()
779 unique_string = "Your Couchers.org account has been successfully recovered."
780 assert unique_string in email.plain
781 assert unique_string in email.html
782 assert "support@couchers.org" in email.plain
783 assert "support@couchers.org" in email.html
785 push = push_collector.pop_for_user(user_id, last=True)
786 assert push.content.title == "Account restored"
787 assert push.content.body == "Welcome back!"
789 with session_scope() as session:
790 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
792 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
793 assert user.deleted_at is None
794 assert not user.undelete_token
795 assert not user.undelete_until
798def test_multiple_delete_tokens(db):
799 """
800 Make sure deletion tokens are deleted on delete
801 """
802 user, token = generate_user()
804 with account_session(token) as account:
805 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
806 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
807 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
809 with session_scope() as session:
810 assert session.execute(select(func.count()).select_from(AccountDeletionToken)).scalar_one() == 3
811 token = session.execute(select(AccountDeletionToken.token).limit(1)).scalar_one()
813 with auth_api_session() as (auth_api, metadata_interceptor):
814 auth_api.ConfirmDeleteAccount(
815 auth_pb2.ConfirmDeleteAccountReq(
816 token=token,
817 )
818 )
820 with session_scope() as session:
821 assert not session.execute(select(AccountDeletionToken.token)).scalar_one_or_none()
824def test_ListActiveSessions_pagination(db, fast_passwords):
825 password = random_hex()
826 user, token = generate_user(hashed_password=hash_password(password))
828 with auth_api_session() as (auth_api, metadata_interceptor):
829 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
830 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
831 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
832 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
834 with real_account_session(token) as account:
835 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_size=3))
836 assert len(res.active_sessions) == 3
837 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_token=res.next_page_token, page_size=3))
838 assert len(res.active_sessions) == 2
839 assert not res.next_page_token
842def test_ListActiveSessions_details(db, fast_passwords):
843 password = random_hex()
844 user, token = generate_user(hashed_password=hash_password(password))
846 ips_user_agents = [
847 (
848 "108.123.33.162",
849 "Mozilla/5.0 (iPhone; CPU iPhone OS 17_6_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.6 Mobile/15E148 Safari/604.1",
850 ),
851 (
852 "8.245.212.28",
853 "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) SamsungBrowser/26.0 Chrome/122.0.0.0 Mobile Safari/537.36",
854 ),
855 (
856 "95.254.140.156",
857 "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:130.0) Gecko/20100101 Firefox/130.0",
858 ),
859 ]
861 for ip, user_agent in ips_user_agents:
862 options = (("grpc.primary_user_agent", user_agent),)
863 with auth_api_session(grpc_channel_options=options) as (auth_api, metadata_interceptor):
864 auth_api.Authenticate(
865 auth_pb2.AuthReq(user=user.username, password=password), metadata=(("x-couchers-real-ip", ip),)
866 )
868 def dummy_geoip(ip_address):
869 return {
870 "108.123.33.162": "Chicago, United States",
871 "8.245.212.28": "Sydney, Australia",
872 }.get(ip_address)
874 with real_account_session(token) as account:
875 with patch("couchers.servicers.account.geoip_approximate_location", dummy_geoip):
876 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
877 print(res)
878 assert len(res.active_sessions) == 4
880 # this one currently making the API call
881 assert res.active_sessions[0].operating_system == "Other"
882 assert res.active_sessions[0].browser == "Other"
883 assert res.active_sessions[0].device == "Other"
884 assert res.active_sessions[0].approximate_location == "Unknown"
885 assert res.active_sessions[0].is_current_session
887 assert res.active_sessions[1].operating_system == "Ubuntu"
888 assert res.active_sessions[1].browser == "Firefox"
889 assert res.active_sessions[1].device == "Other"
890 assert res.active_sessions[1].approximate_location == "Unknown"
891 assert not res.active_sessions[1].is_current_session
893 assert res.active_sessions[2].operating_system == "Android"
894 assert res.active_sessions[2].browser == "Samsung Internet"
895 assert res.active_sessions[2].device == "K"
896 assert res.active_sessions[2].approximate_location == "Sydney, Australia"
897 assert not res.active_sessions[2].is_current_session
899 assert res.active_sessions[3].operating_system == "iOS"
900 assert res.active_sessions[3].browser == "Mobile Safari"
901 assert res.active_sessions[3].device == "iPhone"
902 assert res.active_sessions[3].approximate_location == "Chicago, United States"
903 assert not res.active_sessions[3].is_current_session
906def test_LogOutSession(db, fast_passwords):
907 password = random_hex()
908 user, token = generate_user(hashed_password=hash_password(password))
910 with auth_api_session() as (auth_api, metadata_interceptor):
911 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
912 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
913 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
914 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
916 with real_account_session(token) as account:
917 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
918 assert len(res.active_sessions) == 5
919 account.LogOutSession(account_pb2.LogOutSessionReq(created=res.active_sessions[3].created))
921 res2 = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
922 assert len(res2.active_sessions) == 4
924 # ignore the first session as it changes
925 assert res.active_sessions[1:3] + res.active_sessions[4:] == res2.active_sessions[1:]
928def test_LogOutOtherSessions(db, fast_passwords):
929 password = random_hex()
930 user, token = generate_user(hashed_password=hash_password(password))
932 with auth_api_session() as (auth_api, metadata_interceptor):
933 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
934 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
935 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
936 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
938 with real_account_session(token) as account:
939 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
940 assert len(res.active_sessions) == 5
941 with pytest.raises(grpc.RpcError) as e:
942 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=False))
943 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
944 assert e.value.details() == "Please confirm you want to log out of other sessions."
946 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=True))
947 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
948 assert len(res.active_sessions) == 1
951def test_CreateInviteCode(db):
952 user, token = generate_user()
954 with account_session(token) as account:
955 res = account.CreateInviteCode(account_pb2.CreateInviteCodeReq())
956 code = res.code
957 assert len(code) == 8
959 with session_scope() as session:
960 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
961 assert invite.creator_user_id == user.id
962 assert invite.disabled is None
963 assert res.url == urls.invite_code_link(code=res.code)
966def test_DisableInviteCode(db):
967 user, token = generate_user()
969 with account_session(token) as account:
970 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
971 account.DisableInviteCode(account_pb2.DisableInviteCodeReq(code=code))
973 with session_scope() as session:
974 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
975 assert invite.disabled is not None
978def test_ListInviteCodes(db):
979 user, token = generate_user()
980 another_user, _ = generate_user()
982 with account_session(token) as account:
983 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
985 # simulate another_user having signed up with this invite code
986 with session_scope() as session:
987 session.execute(update(User).where(User.id == another_user.id).values(invite_code_id=code))
989 with account_session(token) as account:
990 res = account.ListInviteCodes(empty_pb2.Empty())
991 assert len(res.invite_codes) == 1
992 assert res.invite_codes[0].code == code
993 assert res.invite_codes[0].uses == 1
994 assert res.invite_codes[0].url == urls.invite_code_link(code=code)
997def test_reminders(db, moderator):
998 # reference writing reminders tested in test_AvailableWriteReferences_and_ListPendingReferencesToWrite
999 # we use LiteUser, so remember to refresh materialized views
1000 user, token = generate_user(complete_profile=False)
1001 complete_user, complete_token = generate_user(complete_profile=True)
1002 req_user1, req_user_token1 = generate_user(complete_profile=True)
1003 req_user2, req_user_token2 = generate_user(complete_profile=True)
1005 refresh_materialized_views_rapid(empty_pb2.Empty())
1006 with account_session(complete_token) as account:
1007 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1008 with account_session(token) as account:
1009 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1010 "complete_profile_reminder",
1011 ]
1013 today_plus_2 = (today() + timedelta(days=2)).isoformat()
1014 today_plus_3 = (today() + timedelta(days=3)).isoformat()
1015 with requests_session(req_user_token1) as api:
1016 host_request1_id = api.CreateHostRequest(
1017 requests_pb2.CreateHostRequestReq(
1018 host_user_id=user.id,
1019 from_date=today_plus_2,
1020 to_date=today_plus_3,
1021 text=valid_request_text("Test request 1"),
1022 )
1023 ).host_request_id
1024 moderator.approve_host_request(host_request1_id)
1026 with account_session(token) as account:
1027 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1028 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1029 "respond_to_host_request_reminder",
1030 "complete_profile_reminder",
1031 ]
1032 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1033 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1035 with requests_session(req_user_token2) as api:
1036 host_request2_id = api.CreateHostRequest(
1037 requests_pb2.CreateHostRequestReq(
1038 host_user_id=user.id,
1039 from_date=today_plus_2,
1040 to_date=today_plus_3,
1041 text=valid_request_text("Test request 2"),
1042 )
1043 ).host_request_id
1044 moderator.approve_host_request(host_request2_id)
1046 refresh_materialized_views_rapid(empty_pb2.Empty())
1047 with account_session(token) as account:
1048 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1049 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1050 "respond_to_host_request_reminder",
1051 "respond_to_host_request_reminder",
1052 "complete_profile_reminder",
1053 ]
1054 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1055 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1056 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1057 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1059 backdate_conversations()
1060 with requests_session(req_user_token1) as api:
1061 host_request3_id = api.CreateHostRequest(
1062 requests_pb2.CreateHostRequestReq(
1063 host_user_id=user.id,
1064 from_date=today_plus_2,
1065 to_date=today_plus_3,
1066 text=valid_request_text("Test request 3"),
1067 )
1068 ).host_request_id
1069 moderator.approve_host_request(host_request3_id)
1071 refresh_materialized_views_rapid(empty_pb2.Empty())
1072 with account_session(token) as account:
1073 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1074 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1075 "respond_to_host_request_reminder",
1076 "respond_to_host_request_reminder",
1077 "respond_to_host_request_reminder",
1078 "complete_profile_reminder",
1079 ]
1080 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1081 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1082 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1083 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1084 assert reminders[2].respond_to_host_request_reminder.host_request_id == host_request3_id
1085 assert reminders[2].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1087 # accept req
1088 with requests_session(token) as api:
1089 api.RespondHostRequest(
1090 requests_pb2.RespondHostRequestReq(
1091 host_request_id=host_request1_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
1092 )
1093 )
1095 refresh_materialized_views_rapid(empty_pb2.Empty())
1096 with account_session(token) as account:
1097 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1098 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1099 "respond_to_host_request_reminder",
1100 "respond_to_host_request_reminder",
1101 "complete_profile_reminder",
1102 ]
1103 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request2_id
1104 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1105 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request3_id
1106 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1108 # host replies to req2 with a message: reminder should clear even though it's still pending
1109 with requests_session(token) as api:
1110 api.SendHostRequestMessage(
1111 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request2_id, text="Let me think about it")
1112 )
1114 refresh_materialized_views_rapid(empty_pb2.Empty())
1115 with account_session(token) as account:
1116 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1117 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1118 "respond_to_host_request_reminder",
1119 "complete_profile_reminder",
1120 ]
1121 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1122 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1124 # surfer sending a message should not clear the reminder
1125 with requests_session(req_user_token1) as api:
1126 api.SendHostRequestMessage(
1127 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request3_id, text="Any update?")
1128 )
1130 refresh_materialized_views_rapid(empty_pb2.Empty())
1131 with account_session(token) as account:
1132 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1133 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1134 "respond_to_host_request_reminder",
1135 "complete_profile_reminder",
1136 ]
1137 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1138 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1141def test_confirm_host_request_reminder(db, moderator):
1142 host, host_token = generate_user(complete_profile=True)
1143 surfer, surfer_token = generate_user(complete_profile=True)
1145 today_plus_10 = (today() + timedelta(days=10)).isoformat()
1146 today_plus_12 = (today() + timedelta(days=12)).isoformat()
1148 with requests_session(surfer_token) as api:
1149 host_request_id = api.CreateHostRequest(
1150 requests_pb2.CreateHostRequestReq(
1151 host_user_id=host.id,
1152 from_date=today_plus_10,
1153 to_date=today_plus_12,
1154 text=valid_request_text("Please host me"),
1155 )
1156 ).host_request_id
1157 moderator.approve_host_request(host_request_id)
1159 refresh_materialized_views_rapid(empty_pb2.Empty())
1160 with account_session(surfer_token) as account:
1161 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1163 with requests_session(host_token) as api:
1164 api.RespondHostRequest(
1165 requests_pb2.RespondHostRequestReq(
1166 host_request_id=host_request_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
1167 )
1168 )
1170 refresh_materialized_views_rapid(empty_pb2.Empty())
1171 with account_session(surfer_token) as account:
1172 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1173 assert [reminder.WhichOneof("reminder") for reminder in reminders] == ["confirm_host_request_reminder"]
1174 assert reminders[0].confirm_host_request_reminder.host_request_id == host_request_id
1175 assert reminders[0].confirm_host_request_reminder.host_user.user_id == host.id
1177 # after surfer confirms, reminder should clear
1178 with requests_session(surfer_token) as api:
1179 api.RespondHostRequest(
1180 requests_pb2.RespondHostRequestReq(
1181 host_request_id=host_request_id, status=messages_pb2.HOST_REQUEST_STATUS_CONFIRMED
1182 )
1183 )
1185 refresh_materialized_views_rapid(empty_pb2.Empty())
1186 with account_session(surfer_token) as account:
1187 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1190def test_public_trip_offer_reminders(db, moderator):
1191 traveller, traveller_token = generate_user(complete_profile=True)
1192 host1, host1_token = generate_user(complete_profile=True)
1193 host2, host2_token = generate_user(complete_profile=True)
1195 from_date = today() + timedelta(days=5)
1196 to_date = today() + timedelta(days=10)
1197 with session_scope() as session:
1198 trip_id = create_public_trip(session, traveller.id, from_date, to_date)
1200 def offer(host_token: str) -> int:
1201 with requests_session(host_token) as api:
1202 host_request_id: int = api.CreateHostRequest(
1203 requests_pb2.CreateHostRequestReq(
1204 host_user_id=traveller.id,
1205 from_date=from_date.isoformat(),
1206 to_date=to_date.isoformat(),
1207 text=valid_request_text("I'd love to host you"),
1208 public_trip_id=trip_id,
1209 )
1210 ).host_request_id
1211 moderator.approve_host_request(host_request_id)
1212 return host_request_id
1214 def get_reminders(token: str):
1215 refresh_materialized_views_rapid(empty_pb2.Empty())
1216 with account_session(token) as account:
1217 return account.GetReminders(empty_pb2.Empty()).reminders
1219 offer1_id = offer(host1_token)
1220 offer2_id = offer(host2_token)
1222 reminders = get_reminders(traveller_token)
1223 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1224 "respond_to_public_trip_offer_reminder",
1225 "respond_to_public_trip_offer_reminder",
1226 ]
1227 assert reminders[0].respond_to_public_trip_offer_reminder.host_request_id == offer1_id
1228 assert reminders[0].respond_to_public_trip_offer_reminder.host_user.user_id == host1.id
1229 assert reminders[1].respond_to_public_trip_offer_reminder.host_request_id == offer2_id
1230 assert reminders[1].respond_to_public_trip_offer_reminder.host_user.user_id == host2.id
1231 assert list(get_reminders(host1_token)) == []
1233 with requests_session(traveller_token) as api:
1234 api.RespondHostRequest(
1235 requests_pb2.RespondHostRequestReq(
1236 host_request_id=offer1_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
1237 )
1238 )
1240 reminders = get_reminders(traveller_token)
1241 assert [reminder.WhichOneof("reminder") for reminder in reminders] == ["respond_to_public_trip_offer_reminder"]
1242 assert reminders[0].respond_to_public_trip_offer_reminder.host_request_id == offer2_id
1244 reminders = get_reminders(host1_token)
1245 assert [reminder.WhichOneof("reminder") for reminder in reminders] == ["confirm_public_trip_offer_reminder"]
1246 assert reminders[0].confirm_public_trip_offer_reminder.host_request_id == offer1_id
1247 assert reminders[0].confirm_public_trip_offer_reminder.surfer_user.user_id == traveller.id
1249 with requests_session(host1_token) as api:
1250 api.RespondHostRequest(
1251 requests_pb2.RespondHostRequestReq(
1252 host_request_id=offer1_id, status=messages_pb2.HOST_REQUEST_STATUS_CONFIRMED
1253 )
1254 )
1256 assert list(get_reminders(host1_token)) == []
1259def test_my_home_reminder(db):
1260 # can_host with incomplete my home (max_guests not set) → reminder shown
1261 can_host_incomplete, token1 = generate_user(hosting_status=HostingStatus.can_host)
1262 # maybe with incomplete my home → reminder shown
1263 maybe_incomplete, token2 = generate_user(hosting_status=HostingStatus.maybe)
1264 # cant_host → no reminder regardless of my home completion
1265 cant_host, token3 = generate_user(hosting_status=HostingStatus.cant_host)
1266 # can_host with fully completed my home → no reminder
1267 can_host_complete, token4 = generate_user(
1268 hosting_status=HostingStatus.can_host,
1269 max_guests=2,
1270 sleeping_arrangement=SleepingArrangement.private,
1271 # about_place is set by default in make_user
1272 )
1274 with account_session(token1) as account:
1275 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1276 "complete_my_home_reminder",
1277 ]
1279 with account_session(token2) as account:
1280 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1281 "complete_my_home_reminder",
1282 ]
1284 with account_session(token3) as account:
1285 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1287 with account_session(token4) as account:
1288 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1291def test_volunteer_stuff(db):
1292 # taken from couchers/app/backend/resources/badges.json
1293 board_member_id = 8347
1295 # with password
1296 user, token = generate_user(name="Von Tester", username="tester", city="Amsterdam", id=board_member_id)
1298 with account_session(token) as account:
1299 res = account.GetAccountInfo(empty_pb2.Empty())
1300 assert not res.is_volunteer
1302 with pytest.raises(grpc.RpcError) as e:
1303 account.GetMyVolunteerInfo(empty_pb2.Empty())
1304 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1305 assert (
1306 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1307 )
1309 with pytest.raises(grpc.RpcError) as e:
1310 account.UpdateMyVolunteerInfo(account_pb2.UpdateMyVolunteerInfoReq())
1311 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1312 assert (
1313 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1314 )
1316 with session_scope() as session:
1317 session.add(
1318 make_volunteer(
1319 user_id=user.id,
1320 display_name="Great Volunteer",
1321 display_location="The Bitbucket",
1322 role="Lead Tester",
1323 started_volunteering=date(2020, 6, 1),
1324 )
1325 )
1327 with account_session(token) as account:
1328 res = account.GetAccountInfo(empty_pb2.Empty())
1329 assert res.is_volunteer
1331 res = account.GetMyVolunteerInfo(empty_pb2.Empty())
1333 assert res.display_name == "Great Volunteer"
1334 assert res.display_location == "The Bitbucket"
1335 assert res.role == "Lead Tester"
1336 assert res.started_volunteering == "2020-06-01"
1337 assert not res.stopped_volunteering
1338 assert res.show_on_team_page
1339 assert res.link_type == "couchers"
1340 assert res.link_text == "@tester"
1341 assert res.link_url == "http://localhost:3000/user/tester"
1343 res = account.UpdateMyVolunteerInfo(
1344 account_pb2.UpdateMyVolunteerInfoReq(
1345 display_name=wrappers_pb2.StringValue(value=""),
1346 link_type=wrappers_pb2.StringValue(value="website"),
1347 link_text=wrappers_pb2.StringValue(value="testervontester.com.invalid"),
1348 link_url=wrappers_pb2.StringValue(value="https://www.testervontester.com.invalid/"),
1349 )
1350 )
1352 assert res.display_name == ""
1353 assert res.display_location == "The Bitbucket"
1354 assert res.role == "Lead Tester"
1355 assert res.started_volunteering == "2020-06-01"
1356 assert not res.stopped_volunteering
1357 assert res.show_on_team_page
1358 assert res.link_type == "website"
1359 assert res.link_text == "testervontester.com.invalid"
1360 assert res.link_url == "https://www.testervontester.com.invalid/"
1361 res = account.UpdateMyVolunteerInfo(
1362 account_pb2.UpdateMyVolunteerInfoReq(
1363 display_name=wrappers_pb2.StringValue(value=""),
1364 link_type=wrappers_pb2.StringValue(value="linkedin"),
1365 link_text=wrappers_pb2.StringValue(value="tester-vontester"),
1366 )
1367 )
1368 assert res.display_name == ""
1369 assert res.display_location == "The Bitbucket"
1370 assert res.role == "Lead Tester"
1371 assert res.started_volunteering == "2020-06-01"
1372 assert not res.stopped_volunteering
1373 assert res.show_on_team_page
1374 assert res.link_type == "linkedin"
1375 assert res.link_text == "tester-vontester"
1376 assert res.link_url == "https://www.linkedin.com/in/tester-vontester/"
1378 res = account.UpdateMyVolunteerInfo(
1379 account_pb2.UpdateMyVolunteerInfoReq(
1380 display_name=wrappers_pb2.StringValue(value="Tester"),
1381 display_location=wrappers_pb2.StringValue(value=""),
1382 link_type=wrappers_pb2.StringValue(value="email"),
1383 link_text=wrappers_pb2.StringValue(value="tester@vontester.com.invalid"),
1384 )
1385 )
1386 assert res.display_name == "Tester"
1387 assert res.display_location == ""
1388 assert res.role == "Lead Tester"
1389 assert res.started_volunteering == "2020-06-01"
1390 assert not res.stopped_volunteering
1391 assert res.show_on_team_page
1392 assert res.link_type == "email"
1393 assert res.link_text == "tester@vontester.com.invalid"
1394 assert res.link_url == "mailto:tester@vontester.com.invalid"
1396 refresh_materialized_views_rapid(empty_pb2.Empty())
1398 with public_session() as public:
1399 res = public.GetVolunteers(empty_pb2.Empty())
1400 assert len(res.current_volunteers) == 1
1401 v = res.current_volunteers[0]
1402 assert v.name == "Tester"
1403 assert v.username == "tester"
1404 assert v.is_board_member
1405 assert v.role == "Lead Tester"
1406 assert v.location == "Amsterdam"
1407 assert v.img.startswith("http://localhost:5001/img/thumbnail/")
1408 assert v.link_type == "email"
1409 assert v.link_text == "tester@vontester.com.invalid"
1410 assert v.link_url == "mailto:tester@vontester.com.invalid"