Coverage for app/backend/src/tests/test_account.py: 100%
742 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-21 21:52 +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, conversations_pb2, requests_pb2
26from couchers.utils import now, today
27from tests.fixtures.db import 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.test_requests import valid_request_text
39@pytest.fixture(autouse=True)
40def _(testconfig):
41 pass
44def test_GetAccountInfo(db, fast_passwords):
45 # with password
46 user1, token1 = generate_user(hashed_password=hash_password(random_hex()), email="user@couchers.invalid")
48 with account_session(token1) as account:
49 res = account.GetAccountInfo(empty_pb2.Empty())
50 assert res.email == "user@couchers.invalid"
51 assert res.username == user1.username
52 assert not res.has_strong_verification
53 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED
54 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED
55 assert not res.is_superuser
56 assert res.ui_language_preference == ""
57 assert not res.is_volunteer
60def test_donation_banner_no_drive(db):
61 """Test that the banner is not shown when no drive is configured (flag unset)"""
62 # User has donated, but there's no drive, so the banner should not show
63 user, token = generate_user()
65 with account_session(token) as account:
66 res = account.GetAccountInfo(empty_pb2.Empty())
67 assert not res.should_show_donation_banner
70def test_donation_banner_never_donated(db, feature_flags):
71 """Test that banner is shown when user has never donated and drive is active"""
72 # Explicitly set last_donated=None since generate_user defaults to now()
73 user, token = generate_user(last_donated=None)
75 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
76 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
77 with account_session(token) as account:
78 res = account.GetAccountInfo(empty_pb2.Empty())
79 assert res.should_show_donation_banner
82def test_donation_banner_donated_before_drive(db, feature_flags):
83 """Test that banner is shown when user donated before drive start"""
84 user, token = generate_user()
86 # Set donation before drive start
87 with session_scope() as session:
88 last_donated = datetime(2025, 10, 15, tzinfo=UTC) # Before Nov 1
89 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
91 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
92 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
93 with account_session(token) as account:
94 res = account.GetAccountInfo(empty_pb2.Empty())
95 assert res.should_show_donation_banner
98def test_donation_banner_donated_after_drive(db, feature_flags):
99 """Test that banner is not shown when user donated after drive start"""
100 user, token = generate_user()
102 # Set donation after drive start
103 with session_scope() as session:
104 last_donated = datetime(2025, 11, 15, tzinfo=UTC) # After Nov 1
105 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
107 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
108 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
109 with account_session(token) as account:
110 res = account.GetAccountInfo(empty_pb2.Empty())
111 assert not res.should_show_donation_banner
114def test_donation_banner_donated_exactly_at_drive_start(db, feature_flags):
115 """Test that banner is not shown when user donated exactly at drive start time"""
116 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
118 user, token = generate_user()
120 # Set donation exactly at drive start
121 with session_scope() as session:
122 session.execute(update(User).where(User.id == user.id).values(last_donated=drive_start))
124 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
125 with account_session(token) as account:
126 res = account.GetAccountInfo(empty_pb2.Empty())
127 assert not res.should_show_donation_banner
130def test_GetAccountInfo_regression(db):
131 # there was a bug in evaluating `has_completed_profile` on the backend (in python)
132 # when about_me is None but the user has a key, it was failing because len(about_me) doesn't work on None
133 user, token = generate_user(about_me=None, complete_profile=False)
135 # add an avatar photo to the user's profile gallery
136 with session_scope() as session:
137 key = random_hex(32)
138 filename = random_hex(32) + ".jpg"
139 session.add(
140 Upload(
141 key=key,
142 filename=filename,
143 creator_user_id=user.id,
144 )
145 )
146 session.flush()
147 assert user.profile_gallery_id is not None
148 session.add(
149 PhotoGalleryItem(
150 gallery_id=user.profile_gallery_id,
151 upload_key=key,
152 position=0,
153 )
154 )
156 with account_session(token) as account:
157 res = account.GetAccountInfo(empty_pb2.Empty())
160def test_ChangePasswordV2_normal(db, fast_passwords, email_collector: EmailCollector, push_collector: PushCollector):
161 # user has old password and is changing to new password
162 old_password = random_hex()
163 new_password = random_hex()
164 user, token = generate_user(hashed_password=hash_password(old_password))
166 with account_session(token) as account:
167 account.ChangePasswordV2(
168 account_pb2.ChangePasswordV2Req(
169 old_password=old_password,
170 new_password=new_password,
171 )
172 )
174 email = email_collector.pop_for_recipient(user.email, last=True)
175 assert email.subject == "[TEST] Your password was changed"
177 push = push_collector.pop_for_user(user.id, last=True)
178 assert push.content.title == "Password changed"
179 assert push.content.body == "Your password was changed."
181 with session_scope() as session:
182 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
183 assert updated_user.hashed_password == hash_password(new_password)
186def test_ChangePasswordV2_regression(db, fast_passwords):
187 # send_password_changed_email wasn't working
188 # user has old password and is changing to new password
189 old_password = random_hex()
190 new_password = random_hex()
191 user, token = generate_user(hashed_password=hash_password(old_password))
193 with account_session(token) as account:
194 account.ChangePasswordV2(
195 account_pb2.ChangePasswordV2Req(
196 old_password=old_password,
197 new_password=new_password,
198 )
199 )
201 with session_scope() as session:
202 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
203 assert updated_user.hashed_password == hash_password(new_password)
206def test_ChangePasswordV2_normal_short_password(db, fast_passwords):
207 # user has old password and is changing to new password, but used short password
208 old_password = random_hex()
209 new_password = random_hex(length=1)
210 user, token = generate_user(hashed_password=hash_password(old_password))
212 with account_session(token) as account:
213 with pytest.raises(grpc.RpcError) as e:
214 account.ChangePasswordV2(
215 account_pb2.ChangePasswordV2Req(
216 old_password=old_password,
217 new_password=new_password,
218 )
219 )
220 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
221 assert e.value.details() == "The password must be 8 or more characters long."
223 with session_scope() as session:
224 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
225 assert updated_user.hashed_password == hash_password(old_password)
228def test_ChangePasswordV2_normal_long_password(db, fast_passwords):
229 # user has old password and is changing to new password, but used short password
230 old_password = random_hex()
231 new_password = random_hex(length=1000)
232 user, token = generate_user(hashed_password=hash_password(old_password))
234 with account_session(token) as account:
235 with pytest.raises(grpc.RpcError) as e:
236 account.ChangePasswordV2(
237 account_pb2.ChangePasswordV2Req(
238 old_password=old_password,
239 new_password=new_password,
240 )
241 )
242 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
243 assert e.value.details() == "The password must be less than 256 characters."
245 with session_scope() as session:
246 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
247 assert updated_user.hashed_password == hash_password(old_password)
250def test_ChangePasswordV2_normal_insecure_password(db, fast_passwords):
251 # user has old password and is changing to new password, but used insecure password
252 old_password = random_hex()
253 new_password = "12345678"
254 user, token = generate_user(hashed_password=hash_password(old_password))
256 with account_session(token) as account:
257 with pytest.raises(grpc.RpcError) as e:
258 account.ChangePasswordV2(
259 account_pb2.ChangePasswordV2Req(
260 old_password=old_password,
261 new_password=new_password,
262 )
263 )
264 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
265 assert e.value.details() == "The password is insecure. Please use one that is not easily guessable."
267 with session_scope() as session:
268 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
269 assert updated_user.hashed_password == hash_password(old_password)
272def test_ChangePasswordV2_normal_wrong_password(db, fast_passwords):
273 # user has old password and is changing to new password, but used wrong old password
274 old_password = random_hex()
275 new_password = random_hex()
276 user, token = generate_user(hashed_password=hash_password(old_password))
278 with account_session(token) as account:
279 with pytest.raises(grpc.RpcError) as e:
280 account.ChangePasswordV2(
281 account_pb2.ChangePasswordV2Req(
282 old_password="Wrong password",
283 new_password=new_password,
284 )
285 )
286 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
287 assert e.value.details() == "Wrong password."
289 with session_scope() as session:
290 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
291 assert updated_user.hashed_password == hash_password(old_password)
294def test_ChangePasswordV2_normal_no_passwords(db, fast_passwords):
295 # user has old password and called with empty body
296 old_password = random_hex()
297 user, token = generate_user(hashed_password=hash_password(old_password))
299 with account_session(token) as account:
300 with pytest.raises(grpc.RpcError) as e:
301 account.ChangePasswordV2(account_pb2.ChangePasswordV2Req(old_password=old_password))
302 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
303 assert e.value.details() == "The password must be 8 or more characters long."
305 with session_scope() as session:
306 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
307 assert updated_user.hashed_password == hash_password(old_password)
310def test_ChangeEmailV2_wrong_password(db, fast_passwords):
311 password = random_hex()
312 new_email = f"{random_hex()}@couchers.org.invalid"
313 user, token = generate_user(hashed_password=hash_password(password))
315 with account_session(token) as account:
316 with pytest.raises(grpc.RpcError) as e:
317 account.ChangeEmailV2(
318 account_pb2.ChangeEmailV2Req(
319 password="Wrong password",
320 new_email=new_email,
321 )
322 )
323 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
324 assert e.value.details() == "Wrong password."
326 with session_scope() as session:
327 assert (
328 session.execute(
329 select(func.count())
330 .select_from(User)
331 .where(User.new_email_token_created <= func.now())
332 .where(User.new_email_token_expiry >= func.now())
333 )
334 ).scalar_one() == 0
337def test_ChangeEmailV2_wrong_email(db, fast_passwords):
338 password = random_hex()
339 new_email = f"{random_hex()}@couchers.org.invalid"
340 user, token = generate_user(hashed_password=hash_password(password))
342 with account_session(token) as account:
343 with pytest.raises(grpc.RpcError) as e:
344 account.ChangeEmailV2(
345 account_pb2.ChangeEmailV2Req(
346 password="Wrong password",
347 new_email=new_email,
348 )
349 )
350 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
351 assert e.value.details() == "Wrong password."
353 with session_scope() as session:
354 assert (
355 session.execute(
356 select(func.count())
357 .select_from(User)
358 .where(User.new_email_token_created <= func.now())
359 .where(User.new_email_token_expiry >= func.now())
360 )
361 ).scalar_one() == 0
364def test_ChangeEmailV2_invalid_email(db, fast_passwords):
365 password = random_hex()
366 user, token = generate_user(hashed_password=hash_password(password))
368 with account_session(token) as account:
369 with pytest.raises(grpc.RpcError) as e:
370 account.ChangeEmailV2(
371 account_pb2.ChangeEmailV2Req(
372 password=password,
373 new_email="not a real email",
374 )
375 )
376 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
377 assert e.value.details() == "Invalid email."
379 with session_scope() as session:
380 assert (
381 session.execute(
382 select(func.count())
383 .select_from(User)
384 .where(User.new_email_token_created <= func.now())
385 .where(User.new_email_token_expiry >= func.now())
386 )
387 ).scalar_one() == 0
390def test_ChangeEmailV2_email_in_use(db, fast_passwords):
391 password = random_hex()
392 user, token = generate_user(hashed_password=hash_password(password))
393 user2, token2 = generate_user(hashed_password=hash_password(password))
395 with account_session(token) as account:
396 with pytest.raises(grpc.RpcError) as e:
397 account.ChangeEmailV2(
398 account_pb2.ChangeEmailV2Req(
399 password=password,
400 new_email=user2.email,
401 )
402 )
403 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
404 assert e.value.details() == "Invalid email."
406 with session_scope() as session:
407 assert (
408 session.execute(
409 select(func.count())
410 .select_from(User)
411 .where(User.new_email_token_created <= func.now())
412 .where(User.new_email_token_expiry >= func.now())
413 )
414 ).scalar_one() == 0
417def test_ChangeEmailV2_no_change(db, fast_passwords):
418 password = random_hex()
419 user, token = generate_user(hashed_password=hash_password(password))
421 with account_session(token) as account:
422 with pytest.raises(grpc.RpcError) as e:
423 account.ChangeEmailV2(
424 account_pb2.ChangeEmailV2Req(
425 password=password,
426 new_email=user.email,
427 )
428 )
429 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
430 assert e.value.details() == "Invalid email."
432 with session_scope() as session:
433 assert (
434 session.execute(
435 select(func.count())
436 .select_from(User)
437 .where(User.new_email_token_created <= func.now())
438 .where(User.new_email_token_expiry >= func.now())
439 )
440 ).scalar_one() == 0
443def test_ChangeEmailV2_wrong_token(db, fast_passwords):
444 password = random_hex()
445 new_email = f"{random_hex()}@couchers.org.invalid"
446 user, token = generate_user(hashed_password=hash_password(password))
448 with account_session(token) as account:
449 account.ChangeEmailV2(
450 account_pb2.ChangeEmailV2Req(
451 password=password,
452 new_email=new_email,
453 )
454 )
456 with auth_api_session() as (auth_api, metadata_interceptor):
457 with pytest.raises(grpc.RpcError) as e:
458 res = auth_api.ConfirmChangeEmailV2(
459 auth_pb2.ConfirmChangeEmailV2Req(
460 change_email_token="wrongtoken",
461 )
462 )
463 assert e.value.code() == grpc.StatusCode.NOT_FOUND
464 assert e.value.details() == "Invalid token."
466 with session_scope() as session:
467 user_updated = session.execute(select(User).where(User.id == user.id)).scalar_one()
468 assert user_updated.email == user.email
471def test_ChangeEmailV2_tokens_two_hour_window(db):
472 def two_hours_one_minute_in_future():
473 return now() + timedelta(hours=2, minutes=1)
475 def one_minute_ago():
476 return now() - timedelta(minutes=1)
478 password = random_hex()
479 new_email = f"{random_hex()}@couchers.org.invalid"
480 user, token = generate_user(hashed_password=hash_password(password))
482 with account_session(token) as account:
483 account.ChangeEmailV2(
484 account_pb2.ChangeEmailV2Req(
485 password=password,
486 new_email=new_email,
487 )
488 )
490 with session_scope() as session:
491 new_email_token = session.execute(select(User.new_email_token).where(User.id == user.id)).scalar_one()
493 with patch("couchers.servicers.auth.now", one_minute_ago):
494 with auth_api_session() as (auth_api, metadata_interceptor):
495 with pytest.raises(grpc.RpcError) as e:
496 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
497 assert e.value.code() == grpc.StatusCode.NOT_FOUND
498 assert e.value.details() == "Invalid token."
500 with pytest.raises(grpc.RpcError) as e:
501 auth_api.ConfirmChangeEmailV2(
502 auth_pb2.ConfirmChangeEmailV2Req(
503 change_email_token=new_email_token,
504 )
505 )
506 assert e.value.code() == grpc.StatusCode.NOT_FOUND
507 assert e.value.details() == "Invalid token."
509 with patch("couchers.servicers.auth.now", two_hours_one_minute_in_future):
510 with auth_api_session() as (auth_api, metadata_interceptor):
511 with pytest.raises(grpc.RpcError) as e:
512 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
513 assert e.value.code() == grpc.StatusCode.NOT_FOUND
514 assert e.value.details() == "Invalid token."
516 with pytest.raises(grpc.RpcError) as e:
517 auth_api.ConfirmChangeEmailV2(
518 auth_pb2.ConfirmChangeEmailV2Req(
519 change_email_token=new_email_token,
520 )
521 )
522 assert e.value.code() == grpc.StatusCode.NOT_FOUND
523 assert e.value.details() == "Invalid token."
526def test_ChangeEmailV2(db, fast_passwords, push_collector: PushCollector):
527 password = random_hex()
528 new_email = f"{random_hex()}@couchers.org.invalid"
529 user, token = generate_user(hashed_password=hash_password(password))
530 user_id = user.id
532 with account_session(token) as account:
533 account.ChangeEmailV2(
534 account_pb2.ChangeEmailV2Req(
535 password=password,
536 new_email=new_email,
537 )
538 )
540 with session_scope() as session:
541 user_updated = session.execute(select(User).where(User.id == user_id)).scalar_one()
542 assert user_updated.email == user.email
543 assert user_updated.new_email == new_email
544 assert user_updated.new_email_token is not None
545 assert user_updated.new_email_token_created
546 assert user_updated.new_email_token_created <= now()
547 assert user_updated.new_email_token_expiry
548 assert user_updated.new_email_token_expiry >= now()
550 token = user_updated.new_email_token
552 process_jobs()
553 push = push_collector.pop_for_user(user_id, last=True)
554 assert push.content.title == "Email change requested"
555 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
557 with auth_api_session() as (auth_api, metadata_interceptor):
558 auth_api.ConfirmChangeEmailV2(
559 auth_pb2.ConfirmChangeEmailV2Req(
560 change_email_token=token,
561 )
562 )
564 with session_scope() as session:
565 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
566 assert user.email == new_email
567 assert user.new_email is None
568 assert user.new_email_token is None
569 assert user.new_email_token_created is None
570 assert user.new_email_token_expiry is None
572 process_jobs()
573 push = push_collector.pop_for_user(user_id, last=True)
574 assert push.content.title == "Email verified"
575 assert push.content.body == "Your new email address has been verified."
578def test_ChangeEmailV2_sends_proper_emails(db, fast_passwords, push_collector: PushCollector):
579 password = random_hex()
580 new_email = f"{random_hex()}@couchers.org.invalid"
581 user, token = generate_user(hashed_password=hash_password(password))
583 with account_session(token) as account:
584 account.ChangeEmailV2(
585 account_pb2.ChangeEmailV2Req(
586 password=password,
587 new_email=new_email,
588 )
589 )
591 process_jobs()
593 with session_scope() as session:
594 jobs = session.execute(select(BackgroundJob).where(BackgroundJob.job_type == "send_email")).scalars().all()
595 assert len(jobs) == 2
596 uq_str1 = b"Email address change initiated"
597 uq_str2 = b"You requested that your email be changed from"
598 assert (uq_str1 in jobs[0].payload and uq_str2 in jobs[1].payload) or (
599 uq_str2 in jobs[0].payload and uq_str1 in jobs[1].payload
600 )
602 push = push_collector.pop_for_user(user.id, last=True)
603 assert push.content.title == "Email change requested"
604 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
607def test_ChangeLanguagePreference(db, fast_passwords):
608 # user changes from default to ISO 639-1 language code
609 new_lang = "zh"
610 user, token = generate_user()
612 with real_account_session(token) as account:
613 res = account.GetAccountInfo(empty_pb2.Empty())
614 assert res.ui_language_preference == ""
616 # call will have info about the request
617 res, call = account.ChangeLanguagePreference.with_call(
618 account_pb2.ChangeLanguagePreferenceReq(ui_language_preference=new_lang)
619 )
621 # cookies are sent via initial metadata, so we check for it there
622 # the value of "set-cookie" will be the full cookie string, pull the key value from the string
623 cookie_values = [v.split(";")[0] for k, v in call.initial_metadata() if k == "set-cookie"]
624 assert any(val == "NEXT_LOCALE=zh" for val in cookie_values), (
625 f"Didn't find the right cookie, got {call.initial_metadata()}"
626 )
628 # the changed language preference should also be sent to the backend
629 res = account.GetAccountInfo(empty_pb2.Empty())
630 assert res.ui_language_preference == "zh"
633def test_contributor_form(db):
634 user, token = generate_user()
636 with account_session(token) as account:
637 res = account.GetContributorFormInfo(empty_pb2.Empty())
638 assert not res.filled_contributor_form
640 account.FillContributorForm(account_pb2.FillContributorFormReq(contributor_form=auth_pb2.ContributorForm()))
642 res = account.GetContributorFormInfo(empty_pb2.Empty())
643 assert res.filled_contributor_form
646def test_DeleteAccount_start(db, email_collector: EmailCollector):
647 user, token = generate_user()
649 with account_session(token) as account:
650 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None))
651 email = email_collector.pop_for_recipient(user.email, last=True)
652 assert email.subject == "[TEST] Confirm your account deletion"
654 with session_scope() as session:
655 deletion_token: AccountDeletionToken = session.execute(
656 select(AccountDeletionToken).where(AccountDeletionToken.user_id == user.id)
657 ).scalar_one()
659 assert deletion_token.is_valid
660 assert session.execute(select(User).where(User.id == user.id)).scalar_one().deleted_at is None
663def test_DeleteAccount_message_storage(db):
664 user, token = generate_user()
666 with account_session(token) as account:
667 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None)) # not stored
668 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="")) # not stored
669 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="Reason"))
670 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="0192#(&!&#)*@//)(8"))
671 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="\n\n\t")) # not stored
672 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="1337"))
674 with session_scope() as session:
675 assert session.execute(select(func.count()).select_from(AccountDeletionReason)).scalar_one() == 3
678def test_full_delete_account_with_recovery(db, email_collector: EmailCollector, push_collector: PushCollector):
679 user, token = generate_user()
680 user_id = user.id
682 with account_session(token) as account:
683 with pytest.raises(grpc.RpcError) as err:
684 account.DeleteAccount(account_pb2.DeleteAccountReq())
685 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
686 assert err.value.details() == "Please confirm your account deletion."
688 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
690 email = email_collector.pop_for_recipient(user.email, last=True)
691 assert email.subject == "[TEST] Confirm your account deletion"
692 assert email.recipient == user.email
693 assert "account deletion" in email.subject.lower()
694 unique_string = "You requested that we delete your Couchers.org account."
695 assert unique_string in email.plain
696 assert unique_string in email.html
697 assert "support@couchers.org" in email.plain
698 assert "support@couchers.org" in email.html
700 push = push_collector.pop_for_user(user_id, last=True)
701 assert push.content.title == "Account deletion requested"
702 assert push.content.body == "Use the link we emailed you to confirm."
704 with session_scope() as session:
705 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
706 delete_token = token_o.token
708 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
709 assert token_o.user == user_
710 assert user_.deleted_at is None
711 assert not user_.undelete_token
712 assert not user_.undelete_until
714 assert delete_token in email.plain
715 assert delete_token in email.html
716 delete_url = f"http://localhost:3000/delete-account?token={delete_token}"
717 assert delete_url in email.plain
718 assert delete_url in email.html
720 with auth_api_session() as (auth_api, metadata_interceptor):
721 auth_api.ConfirmDeleteAccount(
722 auth_pb2.ConfirmDeleteAccountReq(
723 token=delete_token,
724 )
725 )
727 email = email_collector.pop_for_recipient(user.email, last=True)
728 assert email.recipient == user.email
729 assert "account has been deleted" in email.subject.lower()
730 unique_string = "You have successfully deleted your Couchers.org account."
731 assert unique_string in email.plain
732 assert unique_string in email.html
733 assert "7 days" in email.plain
734 assert "7 days" in email.html
735 assert "support@couchers.org" in email.plain
736 assert "support@couchers.org" in email.html
738 push = push_collector.pop_for_user(user_id, last=True)
739 assert push.content.title == "Account deleted"
740 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
742 with session_scope() as session:
743 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
745 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
746 assert user_.deleted_at is not None
747 assert user_.undelete_token
748 assert user_.undelete_until
749 assert user_.undelete_until > now()
751 undelete_token = user_.undelete_token
753 undelete_url = f"http://localhost:3000/recover-account?token={undelete_token}"
754 assert undelete_url in email.plain
755 assert undelete_url in email.html
757 with auth_api_session() as (auth_api, metadata_interceptor):
758 auth_api.RecoverAccount(
759 auth_pb2.RecoverAccountReq(
760 token=undelete_token,
761 )
762 )
764 email = email_collector.pop_for_recipient(user.email, last=True)
765 assert email.recipient == user.email
766 assert "account has been recovered" in email.subject.lower()
767 unique_string = "Your Couchers.org account has been successfully recovered."
768 assert unique_string in email.plain
769 assert unique_string in email.html
770 assert "support@couchers.org" in email.plain
771 assert "support@couchers.org" in email.html
773 push = push_collector.pop_for_user(user_id, last=True)
774 assert push.content.title == "Account restored"
775 assert push.content.body == "Welcome back!"
777 with session_scope() as session:
778 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
780 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
781 assert user.deleted_at is None
782 assert not user.undelete_token
783 assert not user.undelete_until
786def test_multiple_delete_tokens(db):
787 """
788 Make sure deletion tokens are deleted on delete
789 """
790 user, token = generate_user()
792 with account_session(token) as account:
793 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
794 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
795 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
797 with session_scope() as session:
798 assert session.execute(select(func.count()).select_from(AccountDeletionToken)).scalar_one() == 3
799 token = session.execute(select(AccountDeletionToken.token).limit(1)).scalar_one()
801 with auth_api_session() as (auth_api, metadata_interceptor):
802 auth_api.ConfirmDeleteAccount(
803 auth_pb2.ConfirmDeleteAccountReq(
804 token=token,
805 )
806 )
808 with session_scope() as session:
809 assert not session.execute(select(AccountDeletionToken.token)).scalar_one_or_none()
812def test_ListActiveSessions_pagination(db, fast_passwords):
813 password = random_hex()
814 user, token = generate_user(hashed_password=hash_password(password))
816 with auth_api_session() as (auth_api, metadata_interceptor):
817 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
818 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
819 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
820 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
822 with real_account_session(token) as account:
823 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_size=3))
824 assert len(res.active_sessions) == 3
825 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_token=res.next_page_token, page_size=3))
826 assert len(res.active_sessions) == 2
827 assert not res.next_page_token
830def test_ListActiveSessions_details(db, fast_passwords):
831 password = random_hex()
832 user, token = generate_user(hashed_password=hash_password(password))
834 ips_user_agents = [
835 (
836 "108.123.33.162",
837 "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",
838 ),
839 (
840 "8.245.212.28",
841 "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",
842 ),
843 (
844 "95.254.140.156",
845 "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:130.0) Gecko/20100101 Firefox/130.0",
846 ),
847 ]
849 for ip, user_agent in ips_user_agents:
850 options = (("grpc.primary_user_agent", user_agent),)
851 with auth_api_session(grpc_channel_options=options) as (auth_api, metadata_interceptor):
852 auth_api.Authenticate(
853 auth_pb2.AuthReq(user=user.username, password=password), metadata=(("x-couchers-real-ip", ip),)
854 )
856 def dummy_geoip(ip_address):
857 return {
858 "108.123.33.162": "Chicago, United States",
859 "8.245.212.28": "Sydney, Australia",
860 }.get(ip_address)
862 with real_account_session(token) as account:
863 with patch("couchers.servicers.account.geoip_approximate_location", dummy_geoip):
864 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
865 print(res)
866 assert len(res.active_sessions) == 4
868 # this one currently making the API call
869 assert res.active_sessions[0].operating_system == "Other"
870 assert res.active_sessions[0].browser == "Other"
871 assert res.active_sessions[0].device == "Other"
872 assert res.active_sessions[0].approximate_location == "Unknown"
873 assert res.active_sessions[0].is_current_session
875 assert res.active_sessions[1].operating_system == "Ubuntu"
876 assert res.active_sessions[1].browser == "Firefox"
877 assert res.active_sessions[1].device == "Other"
878 assert res.active_sessions[1].approximate_location == "Unknown"
879 assert not res.active_sessions[1].is_current_session
881 assert res.active_sessions[2].operating_system == "Android"
882 assert res.active_sessions[2].browser == "Samsung Internet"
883 assert res.active_sessions[2].device == "K"
884 assert res.active_sessions[2].approximate_location == "Sydney, Australia"
885 assert not res.active_sessions[2].is_current_session
887 assert res.active_sessions[3].operating_system == "iOS"
888 assert res.active_sessions[3].browser == "Mobile Safari"
889 assert res.active_sessions[3].device == "iPhone"
890 assert res.active_sessions[3].approximate_location == "Chicago, United States"
891 assert not res.active_sessions[3].is_current_session
894def test_LogOutSession(db, fast_passwords):
895 password = random_hex()
896 user, token = generate_user(hashed_password=hash_password(password))
898 with auth_api_session() as (auth_api, metadata_interceptor):
899 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
900 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
901 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
902 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
904 with real_account_session(token) as account:
905 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
906 assert len(res.active_sessions) == 5
907 account.LogOutSession(account_pb2.LogOutSessionReq(created=res.active_sessions[3].created))
909 res2 = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
910 assert len(res2.active_sessions) == 4
912 # ignore the first session as it changes
913 assert res.active_sessions[1:3] + res.active_sessions[4:] == res2.active_sessions[1:]
916def test_LogOutOtherSessions(db, fast_passwords):
917 password = random_hex()
918 user, token = generate_user(hashed_password=hash_password(password))
920 with auth_api_session() as (auth_api, metadata_interceptor):
921 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
922 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
923 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
924 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
926 with real_account_session(token) as account:
927 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
928 assert len(res.active_sessions) == 5
929 with pytest.raises(grpc.RpcError) as e:
930 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=False))
931 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
932 assert e.value.details() == "Please confirm you want to log out of other sessions."
934 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=True))
935 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
936 assert len(res.active_sessions) == 1
939def test_CreateInviteCode(db):
940 user, token = generate_user()
942 with account_session(token) as account:
943 res = account.CreateInviteCode(account_pb2.CreateInviteCodeReq())
944 code = res.code
945 assert len(code) == 8
947 with session_scope() as session:
948 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
949 assert invite.creator_user_id == user.id
950 assert invite.disabled is None
951 assert res.url == urls.invite_code_link(code=res.code)
954def test_DisableInviteCode(db):
955 user, token = generate_user()
957 with account_session(token) as account:
958 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
959 account.DisableInviteCode(account_pb2.DisableInviteCodeReq(code=code))
961 with session_scope() as session:
962 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
963 assert invite.disabled is not None
966def test_ListInviteCodes(db):
967 user, token = generate_user()
968 another_user, _ = generate_user()
970 with account_session(token) as account:
971 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
973 # simulate another_user having signed up with this invite code
974 with session_scope() as session:
975 session.execute(update(User).where(User.id == another_user.id).values(invite_code_id=code))
977 with account_session(token) as account:
978 res = account.ListInviteCodes(empty_pb2.Empty())
979 assert len(res.invite_codes) == 1
980 assert res.invite_codes[0].code == code
981 assert res.invite_codes[0].uses == 1
982 assert res.invite_codes[0].url == urls.invite_code_link(code=code)
985def test_reminders(db, moderator):
986 # reference writing reminders tested in test_AvailableWriteReferences_and_ListPendingReferencesToWrite
987 # we use LiteUser, so remember to refresh materialized views
988 user, token = generate_user(complete_profile=False)
989 complete_user, complete_token = generate_user(complete_profile=True)
990 req_user1, req_user_token1 = generate_user(complete_profile=True)
991 req_user2, req_user_token2 = generate_user(complete_profile=True)
993 refresh_materialized_views_rapid(empty_pb2.Empty())
994 with account_session(complete_token) as account:
995 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
996 with account_session(token) as account:
997 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
998 "complete_profile_reminder",
999 ]
1001 today_plus_2 = (today() + timedelta(days=2)).isoformat()
1002 today_plus_3 = (today() + timedelta(days=3)).isoformat()
1003 with requests_session(req_user_token1) as api:
1004 host_request1_id = api.CreateHostRequest(
1005 requests_pb2.CreateHostRequestReq(
1006 host_user_id=user.id,
1007 from_date=today_plus_2,
1008 to_date=today_plus_3,
1009 text=valid_request_text("Test request 1"),
1010 )
1011 ).host_request_id
1012 moderator.approve_host_request(host_request1_id)
1014 with account_session(token) as account:
1015 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1016 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1017 "respond_to_host_request_reminder",
1018 "complete_profile_reminder",
1019 ]
1020 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1021 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1023 with requests_session(req_user_token2) as api:
1024 host_request2_id = api.CreateHostRequest(
1025 requests_pb2.CreateHostRequestReq(
1026 host_user_id=user.id,
1027 from_date=today_plus_2,
1028 to_date=today_plus_3,
1029 text=valid_request_text("Test request 2"),
1030 )
1031 ).host_request_id
1032 moderator.approve_host_request(host_request2_id)
1034 refresh_materialized_views_rapid(empty_pb2.Empty())
1035 with account_session(token) as account:
1036 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1037 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1038 "respond_to_host_request_reminder",
1039 "respond_to_host_request_reminder",
1040 "complete_profile_reminder",
1041 ]
1042 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1043 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1044 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1045 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1047 with requests_session(req_user_token1) as api:
1048 host_request3_id = api.CreateHostRequest(
1049 requests_pb2.CreateHostRequestReq(
1050 host_user_id=user.id,
1051 from_date=today_plus_2,
1052 to_date=today_plus_3,
1053 text=valid_request_text("Test request 3"),
1054 )
1055 ).host_request_id
1056 moderator.approve_host_request(host_request3_id)
1058 refresh_materialized_views_rapid(empty_pb2.Empty())
1059 with account_session(token) as account:
1060 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1061 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1062 "respond_to_host_request_reminder",
1063 "respond_to_host_request_reminder",
1064 "respond_to_host_request_reminder",
1065 "complete_profile_reminder",
1066 ]
1067 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1068 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1069 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1070 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1071 assert reminders[2].respond_to_host_request_reminder.host_request_id == host_request3_id
1072 assert reminders[2].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1074 # accept req
1075 with requests_session(token) as api:
1076 api.RespondHostRequest(
1077 requests_pb2.RespondHostRequestReq(
1078 host_request_id=host_request1_id, status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED
1079 )
1080 )
1082 refresh_materialized_views_rapid(empty_pb2.Empty())
1083 with account_session(token) as account:
1084 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1085 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1086 "respond_to_host_request_reminder",
1087 "respond_to_host_request_reminder",
1088 "complete_profile_reminder",
1089 ]
1090 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request2_id
1091 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1092 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request3_id
1093 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1095 # host replies to req2 with a message: reminder should clear even though it's still pending
1096 with requests_session(token) as api:
1097 api.SendHostRequestMessage(
1098 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request2_id, text="Let me think about it")
1099 )
1101 refresh_materialized_views_rapid(empty_pb2.Empty())
1102 with account_session(token) as account:
1103 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1104 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1105 "respond_to_host_request_reminder",
1106 "complete_profile_reminder",
1107 ]
1108 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1109 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1111 # surfer sending a message should not clear the reminder
1112 with requests_session(req_user_token1) as api:
1113 api.SendHostRequestMessage(
1114 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request3_id, text="Any update?")
1115 )
1117 refresh_materialized_views_rapid(empty_pb2.Empty())
1118 with account_session(token) as account:
1119 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1120 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1121 "respond_to_host_request_reminder",
1122 "complete_profile_reminder",
1123 ]
1124 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1125 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1128def test_confirm_host_request_reminder(db, moderator):
1129 host, host_token = generate_user(complete_profile=True)
1130 surfer, surfer_token = generate_user(complete_profile=True)
1132 today_plus_10 = (today() + timedelta(days=10)).isoformat()
1133 today_plus_12 = (today() + timedelta(days=12)).isoformat()
1135 with requests_session(surfer_token) as api:
1136 host_request_id = api.CreateHostRequest(
1137 requests_pb2.CreateHostRequestReq(
1138 host_user_id=host.id,
1139 from_date=today_plus_10,
1140 to_date=today_plus_12,
1141 text=valid_request_text("Please host me"),
1142 )
1143 ).host_request_id
1144 moderator.approve_host_request(host_request_id)
1146 refresh_materialized_views_rapid(empty_pb2.Empty())
1147 with account_session(surfer_token) as account:
1148 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1150 with requests_session(host_token) as api:
1151 api.RespondHostRequest(
1152 requests_pb2.RespondHostRequestReq(
1153 host_request_id=host_request_id, status=conversations_pb2.HOST_REQUEST_STATUS_ACCEPTED
1154 )
1155 )
1157 refresh_materialized_views_rapid(empty_pb2.Empty())
1158 with account_session(surfer_token) as account:
1159 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1160 assert [reminder.WhichOneof("reminder") for reminder in reminders] == ["confirm_host_request_reminder"]
1161 assert reminders[0].confirm_host_request_reminder.host_request_id == host_request_id
1162 assert reminders[0].confirm_host_request_reminder.host_user.user_id == host.id
1164 # after surfer confirms, reminder should clear
1165 with requests_session(surfer_token) as api:
1166 api.RespondHostRequest(
1167 requests_pb2.RespondHostRequestReq(
1168 host_request_id=host_request_id, status=conversations_pb2.HOST_REQUEST_STATUS_CONFIRMED
1169 )
1170 )
1172 refresh_materialized_views_rapid(empty_pb2.Empty())
1173 with account_session(surfer_token) as account:
1174 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1177def test_my_home_reminder(db):
1178 # can_host with incomplete my home (max_guests not set) → reminder shown
1179 can_host_incomplete, token1 = generate_user(hosting_status=HostingStatus.can_host)
1180 # maybe with incomplete my home → reminder shown
1181 maybe_incomplete, token2 = generate_user(hosting_status=HostingStatus.maybe)
1182 # cant_host → no reminder regardless of my home completion
1183 cant_host, token3 = generate_user(hosting_status=HostingStatus.cant_host)
1184 # can_host with fully completed my home → no reminder
1185 can_host_complete, token4 = generate_user(
1186 hosting_status=HostingStatus.can_host,
1187 max_guests=2,
1188 sleeping_arrangement=SleepingArrangement.private,
1189 # about_place is set by default in make_user
1190 )
1192 with account_session(token1) as account:
1193 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1194 "complete_my_home_reminder",
1195 ]
1197 with account_session(token2) as account:
1198 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1199 "complete_my_home_reminder",
1200 ]
1202 with account_session(token3) as account:
1203 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1205 with account_session(token4) as account:
1206 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1209def test_volunteer_stuff(db):
1210 # taken from couchers/app/backend/resources/badges.json
1211 board_member_id = 8347
1213 # with password
1214 user, token = generate_user(name="Von Tester", username="tester", city="Amsterdam", id=board_member_id)
1216 with account_session(token) as account:
1217 res = account.GetAccountInfo(empty_pb2.Empty())
1218 assert not res.is_volunteer
1220 with pytest.raises(grpc.RpcError) as e:
1221 account.GetMyVolunteerInfo(empty_pb2.Empty())
1222 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1223 assert (
1224 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1225 )
1227 with pytest.raises(grpc.RpcError) as e:
1228 account.UpdateMyVolunteerInfo(account_pb2.UpdateMyVolunteerInfoReq())
1229 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1230 assert (
1231 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1232 )
1234 with session_scope() as session:
1235 session.add(
1236 make_volunteer(
1237 user_id=user.id,
1238 display_name="Great Volunteer",
1239 display_location="The Bitbucket",
1240 role="Lead Tester",
1241 started_volunteering=date(2020, 6, 1),
1242 )
1243 )
1245 with account_session(token) as account:
1246 res = account.GetAccountInfo(empty_pb2.Empty())
1247 assert res.is_volunteer
1249 res = account.GetMyVolunteerInfo(empty_pb2.Empty())
1251 assert res.display_name == "Great Volunteer"
1252 assert res.display_location == "The Bitbucket"
1253 assert res.role == "Lead Tester"
1254 assert res.started_volunteering == "2020-06-01"
1255 assert not res.stopped_volunteering
1256 assert res.show_on_team_page
1257 assert res.link_type == "couchers"
1258 assert res.link_text == "@tester"
1259 assert res.link_url == "http://localhost:3000/user/tester"
1261 res = account.UpdateMyVolunteerInfo(
1262 account_pb2.UpdateMyVolunteerInfoReq(
1263 display_name=wrappers_pb2.StringValue(value=""),
1264 link_type=wrappers_pb2.StringValue(value="website"),
1265 link_text=wrappers_pb2.StringValue(value="testervontester.com.invalid"),
1266 link_url=wrappers_pb2.StringValue(value="https://www.testervontester.com.invalid/"),
1267 )
1268 )
1270 assert res.display_name == ""
1271 assert res.display_location == "The Bitbucket"
1272 assert res.role == "Lead Tester"
1273 assert res.started_volunteering == "2020-06-01"
1274 assert not res.stopped_volunteering
1275 assert res.show_on_team_page
1276 assert res.link_type == "website"
1277 assert res.link_text == "testervontester.com.invalid"
1278 assert res.link_url == "https://www.testervontester.com.invalid/"
1279 res = account.UpdateMyVolunteerInfo(
1280 account_pb2.UpdateMyVolunteerInfoReq(
1281 display_name=wrappers_pb2.StringValue(value=""),
1282 link_type=wrappers_pb2.StringValue(value="linkedin"),
1283 link_text=wrappers_pb2.StringValue(value="tester-vontester"),
1284 )
1285 )
1286 assert res.display_name == ""
1287 assert res.display_location == "The Bitbucket"
1288 assert res.role == "Lead Tester"
1289 assert res.started_volunteering == "2020-06-01"
1290 assert not res.stopped_volunteering
1291 assert res.show_on_team_page
1292 assert res.link_type == "linkedin"
1293 assert res.link_text == "tester-vontester"
1294 assert res.link_url == "https://www.linkedin.com/in/tester-vontester/"
1296 res = account.UpdateMyVolunteerInfo(
1297 account_pb2.UpdateMyVolunteerInfoReq(
1298 display_name=wrappers_pb2.StringValue(value="Tester"),
1299 display_location=wrappers_pb2.StringValue(value=""),
1300 link_type=wrappers_pb2.StringValue(value="email"),
1301 link_text=wrappers_pb2.StringValue(value="tester@vontester.com.invalid"),
1302 )
1303 )
1304 assert res.display_name == "Tester"
1305 assert res.display_location == ""
1306 assert res.role == "Lead Tester"
1307 assert res.started_volunteering == "2020-06-01"
1308 assert not res.stopped_volunteering
1309 assert res.show_on_team_page
1310 assert res.link_type == "email"
1311 assert res.link_text == "tester@vontester.com.invalid"
1312 assert res.link_url == "mailto:tester@vontester.com.invalid"
1314 refresh_materialized_views_rapid(empty_pb2.Empty())
1316 with public_session() as public:
1317 res = public.GetVolunteers(empty_pb2.Empty())
1318 assert len(res.current_volunteers) == 1
1319 v = res.current_volunteers[0]
1320 assert v.name == "Tester"
1321 assert v.username == "tester"
1322 assert v.is_board_member
1323 assert v.role == "Lead Tester"
1324 assert v.location == "Amsterdam"
1325 assert v.img.startswith("http://localhost:5001/img/thumbnail/")
1326 assert v.link_type == "email"
1327 assert v.link_text == "tester@vontester.com.invalid"
1328 assert v.link_url == "mailto:tester@vontester.com.invalid"