Coverage for app/backend/src/tests/test_account.py: 100%
740 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-10 12:25 +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_requests import valid_request_text
40@pytest.fixture(autouse=True)
41def _(testconfig):
42 pass
45def test_GetAccountInfo(db, fast_passwords):
46 # with password
47 user1, token1 = generate_user(hashed_password=hash_password(random_hex()), email="user@couchers.invalid")
49 with account_session(token1) as account:
50 res = account.GetAccountInfo(empty_pb2.Empty())
51 assert res.email == "user@couchers.invalid"
52 assert res.username == user1.username
53 assert not res.has_strong_verification
54 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED
55 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED
56 assert not res.is_superuser
57 assert res.ui_language_preference == ""
58 assert not res.is_volunteer
61def test_donation_banner_no_drive(db):
62 """Test that the banner is not shown when no drive is configured (flag unset)"""
63 # User has donated, but there's no drive, so the banner should not show
64 user, token = generate_user()
66 with account_session(token) as account:
67 res = account.GetAccountInfo(empty_pb2.Empty())
68 assert not res.should_show_donation_banner
71def test_donation_banner_never_donated(db, feature_flags):
72 """Test that banner is shown when user has never donated and drive is active"""
73 # Explicitly set last_donated=None since generate_user defaults to now()
74 user, token = generate_user(last_donated=None)
76 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
77 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
78 with account_session(token) as account:
79 res = account.GetAccountInfo(empty_pb2.Empty())
80 assert res.should_show_donation_banner
83def test_donation_banner_donated_before_drive(db, feature_flags):
84 """Test that banner is shown when user donated before drive start"""
85 user, token = generate_user()
87 # Set donation before drive start
88 with session_scope() as session:
89 last_donated = datetime(2025, 10, 15, tzinfo=UTC) # Before Nov 1
90 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
92 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
93 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
94 with account_session(token) as account:
95 res = account.GetAccountInfo(empty_pb2.Empty())
96 assert res.should_show_donation_banner
99def test_donation_banner_donated_after_drive(db, feature_flags):
100 """Test that banner is not shown when user donated after drive start"""
101 user, token = generate_user()
103 # Set donation after drive start
104 with session_scope() as session:
105 last_donated = datetime(2025, 11, 15, tzinfo=UTC) # After Nov 1
106 session.execute(update(User).where(User.id == user.id).values(last_donated=last_donated))
108 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
109 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
110 with account_session(token) as account:
111 res = account.GetAccountInfo(empty_pb2.Empty())
112 assert not res.should_show_donation_banner
115def test_donation_banner_donated_exactly_at_drive_start(db, feature_flags):
116 """Test that banner is not shown when user donated exactly at drive start time"""
117 drive_start = datetime(2025, 11, 1, tzinfo=UTC)
119 user, token = generate_user()
121 # Set donation exactly at drive start
122 with session_scope() as session:
123 session.execute(update(User).where(User.id == user.id).values(last_donated=drive_start))
125 feature_flags.set("donation_drive_start", int(drive_start.timestamp()))
126 with account_session(token) as account:
127 res = account.GetAccountInfo(empty_pb2.Empty())
128 assert not res.should_show_donation_banner
131def test_GetAccountInfo_regression(db):
132 # there was a bug in evaluating `has_completed_profile` on the backend (in python)
133 # when about_me is None but the user has a key, it was failing because len(about_me) doesn't work on None
134 user, token = generate_user(about_me=None, complete_profile=False)
136 # add an avatar photo to the user's profile gallery
137 with session_scope() as session:
138 key = random_hex(32)
139 filename = random_hex(32) + ".jpg"
140 session.add(
141 Upload(
142 key=key,
143 filename=filename,
144 creator_user_id=user.id,
145 )
146 )
147 session.flush()
148 assert user.profile_gallery_id is not None
149 session.add(
150 PhotoGalleryItem(
151 gallery_id=user.profile_gallery_id,
152 upload_key=key,
153 position=0,
154 )
155 )
157 with account_session(token) as account:
158 res = account.GetAccountInfo(empty_pb2.Empty())
161def test_ChangePasswordV2_normal(db, fast_passwords, email_collector: EmailCollector, push_collector: PushCollector):
162 # user has old password and is changing to new password
163 old_password = random_hex()
164 new_password = random_hex()
165 user, token = generate_user(hashed_password=hash_password(old_password))
167 with account_session(token) as account:
168 account.ChangePasswordV2(
169 account_pb2.ChangePasswordV2Req(
170 old_password=old_password,
171 new_password=new_password,
172 )
173 )
175 email = email_collector.pop_for_recipient(user.email, last=True)
176 assert email.subject == "[TEST] Your password was changed"
178 push = push_collector.pop_for_user(user.id, last=True)
179 assert push.content.title == "Password changed"
180 assert push.content.body == "Your password was changed."
182 with session_scope() as session:
183 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
184 assert updated_user.hashed_password == hash_password(new_password)
187def test_ChangePasswordV2_regression(db, fast_passwords):
188 # send_password_changed_email wasn't working
189 # user has old password and is changing to new password
190 old_password = random_hex()
191 new_password = random_hex()
192 user, token = generate_user(hashed_password=hash_password(old_password))
194 with account_session(token) as account:
195 account.ChangePasswordV2(
196 account_pb2.ChangePasswordV2Req(
197 old_password=old_password,
198 new_password=new_password,
199 )
200 )
202 with session_scope() as session:
203 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
204 assert updated_user.hashed_password == hash_password(new_password)
207def test_ChangePasswordV2_normal_short_password(db, fast_passwords):
208 # user has old password and is changing to new password, but used short password
209 old_password = random_hex()
210 new_password = random_hex(length=1)
211 user, token = generate_user(hashed_password=hash_password(old_password))
213 with account_session(token) as account:
214 with pytest.raises(grpc.RpcError) as e:
215 account.ChangePasswordV2(
216 account_pb2.ChangePasswordV2Req(
217 old_password=old_password,
218 new_password=new_password,
219 )
220 )
221 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
222 assert e.value.details() == "The password must be 8 or more characters long."
224 with session_scope() as session:
225 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
226 assert updated_user.hashed_password == hash_password(old_password)
229def test_ChangePasswordV2_normal_long_password(db, fast_passwords):
230 # user has old password and is changing to new password, but used short password
231 old_password = random_hex()
232 new_password = random_hex(length=1000)
233 user, token = generate_user(hashed_password=hash_password(old_password))
235 with account_session(token) as account:
236 with pytest.raises(grpc.RpcError) as e:
237 account.ChangePasswordV2(
238 account_pb2.ChangePasswordV2Req(
239 old_password=old_password,
240 new_password=new_password,
241 )
242 )
243 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
244 assert e.value.details() == "The password must be less than 256 characters."
246 with session_scope() as session:
247 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
248 assert updated_user.hashed_password == hash_password(old_password)
251def test_ChangePasswordV2_normal_insecure_password(db, fast_passwords):
252 # user has old password and is changing to new password, but used insecure password
253 old_password = random_hex()
254 new_password = "12345678"
255 user, token = generate_user(hashed_password=hash_password(old_password))
257 with account_session(token) as account:
258 with pytest.raises(grpc.RpcError) as e:
259 account.ChangePasswordV2(
260 account_pb2.ChangePasswordV2Req(
261 old_password=old_password,
262 new_password=new_password,
263 )
264 )
265 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
266 assert e.value.details() == "The password is insecure. Please use one that is not easily guessable."
268 with session_scope() as session:
269 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
270 assert updated_user.hashed_password == hash_password(old_password)
273def test_ChangePasswordV2_normal_wrong_password(db, fast_passwords):
274 # user has old password and is changing to new password, but used wrong old password
275 old_password = random_hex()
276 new_password = random_hex()
277 user, token = generate_user(hashed_password=hash_password(old_password))
279 with account_session(token) as account:
280 with pytest.raises(grpc.RpcError) as e:
281 account.ChangePasswordV2(
282 account_pb2.ChangePasswordV2Req(
283 old_password="Wrong password",
284 new_password=new_password,
285 )
286 )
287 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
288 assert e.value.details() == "Wrong password."
290 with session_scope() as session:
291 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
292 assert updated_user.hashed_password == hash_password(old_password)
295def test_ChangePasswordV2_normal_no_passwords(db, fast_passwords):
296 # user has old password and called with empty body
297 old_password = random_hex()
298 user, token = generate_user(hashed_password=hash_password(old_password))
300 with account_session(token) as account:
301 with pytest.raises(grpc.RpcError) as e:
302 account.ChangePasswordV2(account_pb2.ChangePasswordV2Req(old_password=old_password))
303 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
304 assert e.value.details() == "The password must be 8 or more characters long."
306 with session_scope() as session:
307 updated_user = session.execute(select(User).where(User.id == user.id)).scalar_one()
308 assert updated_user.hashed_password == hash_password(old_password)
311def test_ChangeEmailV2_wrong_password(db, fast_passwords):
312 password = random_hex()
313 new_email = f"{random_hex()}@couchers.org.invalid"
314 user, token = generate_user(hashed_password=hash_password(password))
316 with account_session(token) as account:
317 with pytest.raises(grpc.RpcError) as e:
318 account.ChangeEmailV2(
319 account_pb2.ChangeEmailV2Req(
320 password="Wrong password",
321 new_email=new_email,
322 )
323 )
324 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
325 assert e.value.details() == "Wrong password."
327 with session_scope() as session:
328 assert (
329 session.execute(
330 select(func.count())
331 .select_from(User)
332 .where(User.new_email_token_created <= func.now())
333 .where(User.new_email_token_expiry >= func.now())
334 )
335 ).scalar_one() == 0
338def test_ChangeEmailV2_wrong_email(db, fast_passwords):
339 password = random_hex()
340 new_email = f"{random_hex()}@couchers.org.invalid"
341 user, token = generate_user(hashed_password=hash_password(password))
343 with account_session(token) as account:
344 with pytest.raises(grpc.RpcError) as e:
345 account.ChangeEmailV2(
346 account_pb2.ChangeEmailV2Req(
347 password="Wrong password",
348 new_email=new_email,
349 )
350 )
351 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
352 assert e.value.details() == "Wrong password."
354 with session_scope() as session:
355 assert (
356 session.execute(
357 select(func.count())
358 .select_from(User)
359 .where(User.new_email_token_created <= func.now())
360 .where(User.new_email_token_expiry >= func.now())
361 )
362 ).scalar_one() == 0
365def test_ChangeEmailV2_invalid_email(db, fast_passwords):
366 password = random_hex()
367 user, token = generate_user(hashed_password=hash_password(password))
369 with account_session(token) as account:
370 with pytest.raises(grpc.RpcError) as e:
371 account.ChangeEmailV2(
372 account_pb2.ChangeEmailV2Req(
373 password=password,
374 new_email="not a real email",
375 )
376 )
377 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
378 assert e.value.details() == "Invalid email."
380 with session_scope() as session:
381 assert (
382 session.execute(
383 select(func.count())
384 .select_from(User)
385 .where(User.new_email_token_created <= func.now())
386 .where(User.new_email_token_expiry >= func.now())
387 )
388 ).scalar_one() == 0
391def test_ChangeEmailV2_email_in_use(db, fast_passwords):
392 password = random_hex()
393 user, token = generate_user(hashed_password=hash_password(password))
394 user2, token2 = generate_user(hashed_password=hash_password(password))
396 with account_session(token) as account:
397 with pytest.raises(grpc.RpcError) as e:
398 account.ChangeEmailV2(
399 account_pb2.ChangeEmailV2Req(
400 password=password,
401 new_email=user2.email,
402 )
403 )
404 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
405 assert e.value.details() == "Invalid email."
407 with session_scope() as session:
408 assert (
409 session.execute(
410 select(func.count())
411 .select_from(User)
412 .where(User.new_email_token_created <= func.now())
413 .where(User.new_email_token_expiry >= func.now())
414 )
415 ).scalar_one() == 0
418def test_ChangeEmailV2_no_change(db, fast_passwords):
419 password = random_hex()
420 user, token = generate_user(hashed_password=hash_password(password))
422 with account_session(token) as account:
423 with pytest.raises(grpc.RpcError) as e:
424 account.ChangeEmailV2(
425 account_pb2.ChangeEmailV2Req(
426 password=password,
427 new_email=user.email,
428 )
429 )
430 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
431 assert e.value.details() == "Invalid email."
433 with session_scope() as session:
434 assert (
435 session.execute(
436 select(func.count())
437 .select_from(User)
438 .where(User.new_email_token_created <= func.now())
439 .where(User.new_email_token_expiry >= func.now())
440 )
441 ).scalar_one() == 0
444def test_ChangeEmailV2_wrong_token(db, fast_passwords):
445 password = random_hex()
446 new_email = f"{random_hex()}@couchers.org.invalid"
447 user, token = generate_user(hashed_password=hash_password(password))
449 with account_session(token) as account:
450 account.ChangeEmailV2(
451 account_pb2.ChangeEmailV2Req(
452 password=password,
453 new_email=new_email,
454 )
455 )
457 with auth_api_session() as (auth_api, metadata_interceptor):
458 with pytest.raises(grpc.RpcError) as e:
459 res = auth_api.ConfirmChangeEmailV2(
460 auth_pb2.ConfirmChangeEmailV2Req(
461 change_email_token="wrongtoken",
462 )
463 )
464 assert e.value.code() == grpc.StatusCode.NOT_FOUND
465 assert e.value.details() == "Invalid token."
467 with session_scope() as session:
468 user_updated = session.execute(select(User).where(User.id == user.id)).scalar_one()
469 assert user_updated.email == user.email
472def test_ChangeEmailV2_tokens_two_hour_window(db, fast_passwords, timewarp: Timewarp):
473 password = random_hex()
474 new_email = f"{random_hex()}@couchers.org.invalid"
475 user, token = generate_user(hashed_password=hash_password(password))
477 with account_session(token) as account:
478 account.ChangeEmailV2(
479 account_pb2.ChangeEmailV2Req(
480 password=password,
481 new_email=new_email,
482 )
483 )
485 with session_scope() as session:
486 new_email_token = session.execute(select(User.new_email_token).where(User.id == user.id)).scalar_one()
488 # before the token was issued
489 timewarp.advance(-timedelta(minutes=1))
490 with auth_api_session() as (auth_api, metadata_interceptor):
491 with pytest.raises(grpc.RpcError) as e:
492 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
493 assert e.value.code() == grpc.StatusCode.NOT_FOUND
494 assert e.value.details() == "Invalid token."
496 with pytest.raises(grpc.RpcError) as e:
497 auth_api.ConfirmChangeEmailV2(
498 auth_pb2.ConfirmChangeEmailV2Req(
499 change_email_token=new_email_token,
500 )
501 )
502 assert e.value.code() == grpc.StatusCode.NOT_FOUND
503 assert e.value.details() == "Invalid token."
505 # and a minute after the two hour window closed
506 timewarp.advance(timedelta(hours=2, minutes=1))
507 with auth_api_session() as (auth_api, metadata_interceptor):
508 with pytest.raises(grpc.RpcError) as e:
509 auth_api.ConfirmChangeEmailV2(auth_pb2.ConfirmChangeEmailV2Req())
510 assert e.value.code() == grpc.StatusCode.NOT_FOUND
511 assert e.value.details() == "Invalid token."
513 with pytest.raises(grpc.RpcError) as e:
514 auth_api.ConfirmChangeEmailV2(
515 auth_pb2.ConfirmChangeEmailV2Req(
516 change_email_token=new_email_token,
517 )
518 )
519 assert e.value.code() == grpc.StatusCode.NOT_FOUND
520 assert e.value.details() == "Invalid token."
523def test_ChangeEmailV2(db, fast_passwords, push_collector: PushCollector):
524 password = random_hex()
525 new_email = f"{random_hex()}@couchers.org.invalid"
526 user, token = generate_user(hashed_password=hash_password(password))
527 user_id = user.id
529 with account_session(token) as account:
530 account.ChangeEmailV2(
531 account_pb2.ChangeEmailV2Req(
532 password=password,
533 new_email=new_email,
534 )
535 )
537 with session_scope() as session:
538 user_updated = session.execute(select(User).where(User.id == user_id)).scalar_one()
539 assert user_updated.email == user.email
540 assert user_updated.new_email == new_email
541 assert user_updated.new_email_token is not None
542 assert user_updated.new_email_token_created
543 assert user_updated.new_email_token_created <= now()
544 assert user_updated.new_email_token_expiry
545 assert user_updated.new_email_token_expiry >= now()
547 token = user_updated.new_email_token
549 process_jobs()
550 push = push_collector.pop_for_user(user_id, last=True)
551 assert push.content.title == "Email change requested"
552 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
554 with auth_api_session() as (auth_api, metadata_interceptor):
555 auth_api.ConfirmChangeEmailV2(
556 auth_pb2.ConfirmChangeEmailV2Req(
557 change_email_token=token,
558 )
559 )
561 with session_scope() as session:
562 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
563 assert user.email == new_email
564 assert user.new_email is None
565 assert user.new_email_token is None
566 assert user.new_email_token_created is None
567 assert user.new_email_token_expiry is None
569 process_jobs()
570 push = push_collector.pop_for_user(user_id, last=True)
571 assert push.content.title == "Email verified"
572 assert push.content.body == "Your new email address has been verified."
575def test_ChangeEmailV2_sends_proper_emails(db, fast_passwords, push_collector: PushCollector):
576 password = random_hex()
577 new_email = f"{random_hex()}@couchers.org.invalid"
578 user, token = generate_user(hashed_password=hash_password(password))
580 with account_session(token) as account:
581 account.ChangeEmailV2(
582 account_pb2.ChangeEmailV2Req(
583 password=password,
584 new_email=new_email,
585 )
586 )
588 process_jobs()
590 with session_scope() as session:
591 jobs = session.execute(select(BackgroundJob).where(BackgroundJob.job_type == "send_email")).scalars().all()
592 assert len(jobs) == 2
593 uq_str1 = b"Email address change initiated"
594 uq_str2 = b"You requested that your email be changed from"
595 assert (uq_str1 in jobs[0].payload and uq_str2 in jobs[1].payload) or (
596 uq_str2 in jobs[0].payload and uq_str1 in jobs[1].payload
597 )
599 push = push_collector.pop_for_user(user.id, last=True)
600 assert push.content.title == "Email change requested"
601 assert push.content.body == f"Use the link we sent to {new_email} to confirm your new address."
604def test_ChangeLanguagePreference(db, fast_passwords):
605 # user changes from default to ISO 639-1 language code
606 new_lang = "zh"
607 user, token = generate_user()
609 with real_account_session(token) as account:
610 res = account.GetAccountInfo(empty_pb2.Empty())
611 assert res.ui_language_preference == ""
613 # call will have info about the request
614 res, call = account.ChangeLanguagePreference.with_call(
615 account_pb2.ChangeLanguagePreferenceReq(ui_language_preference=new_lang)
616 )
618 # cookies are sent via initial metadata, so we check for it there
619 # the value of "set-cookie" will be the full cookie string, pull the key value from the string
620 cookie_values = [v.split(";")[0] for k, v in call.initial_metadata() if k == "set-cookie"]
621 assert any(val == "NEXT_LOCALE=zh" for val in cookie_values), (
622 f"Didn't find the right cookie, got {call.initial_metadata()}"
623 )
625 # the changed language preference should also be sent to the backend
626 res = account.GetAccountInfo(empty_pb2.Empty())
627 assert res.ui_language_preference == "zh"
630def test_contributor_form(db):
631 user, token = generate_user()
633 with account_session(token) as account:
634 res = account.GetContributorFormInfo(empty_pb2.Empty())
635 assert not res.filled_contributor_form
637 account.FillContributorForm(account_pb2.FillContributorFormReq(contributor_form=auth_pb2.ContributorForm()))
639 res = account.GetContributorFormInfo(empty_pb2.Empty())
640 assert res.filled_contributor_form
643def test_DeleteAccount_start(db, email_collector: EmailCollector):
644 user, token = generate_user()
646 with account_session(token) as account:
647 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None))
648 email = email_collector.pop_for_recipient(user.email, last=True)
649 assert email.subject == "[TEST] Confirm your account deletion"
651 with session_scope() as session:
652 deletion_token: AccountDeletionToken = session.execute(
653 select(AccountDeletionToken).where(AccountDeletionToken.user_id == user.id)
654 ).scalar_one()
656 assert deletion_token.is_valid
657 assert session.execute(select(User).where(User.id == user.id)).scalar_one().deleted_at is None
660def test_DeleteAccount_message_storage(db):
661 user, token = generate_user()
663 with account_session(token) as account:
664 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason=None)) # not stored
665 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="")) # not stored
666 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="Reason"))
667 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="0192#(&!&#)*@//)(8"))
668 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="\n\n\t")) # not stored
669 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True, reason="1337"))
671 with session_scope() as session:
672 assert session.execute(select(func.count()).select_from(AccountDeletionReason)).scalar_one() == 3
675def test_full_delete_account_with_recovery(db, email_collector: EmailCollector, push_collector: PushCollector):
676 user, token = generate_user()
677 user_id = user.id
679 with account_session(token) as account:
680 with pytest.raises(grpc.RpcError) as err:
681 account.DeleteAccount(account_pb2.DeleteAccountReq())
682 assert err.value.code() == grpc.StatusCode.FAILED_PRECONDITION
683 assert err.value.details() == "Please confirm your account deletion."
685 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
687 email = email_collector.pop_for_recipient(user.email, last=True)
688 assert email.subject == "[TEST] Confirm your account deletion"
689 assert email.recipient == user.email
690 assert "account deletion" in email.subject.lower()
691 unique_string = "You requested that we delete your Couchers.org account."
692 assert unique_string in email.plain
693 assert unique_string in email.html
694 assert "support@couchers.org" in email.plain
695 assert "support@couchers.org" in email.html
697 push = push_collector.pop_for_user(user_id, last=True)
698 assert push.content.title == "Account deletion requested"
699 assert push.content.body == "Use the link we emailed you to confirm."
701 with session_scope() as session:
702 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
703 delete_token = token_o.token
705 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
706 assert token_o.user == user_
707 assert user_.deleted_at is None
708 assert not user_.undelete_token
709 assert not user_.undelete_until
711 assert delete_token in email.plain
712 assert delete_token in email.html
713 delete_url = f"http://localhost:3000/delete-account?token={delete_token}"
714 assert delete_url in email.plain
715 assert delete_url in email.html
717 with auth_api_session() as (auth_api, metadata_interceptor):
718 auth_api.ConfirmDeleteAccount(
719 auth_pb2.ConfirmDeleteAccountReq(
720 token=delete_token,
721 )
722 )
724 email = email_collector.pop_for_recipient(user.email, last=True)
725 assert email.recipient == user.email
726 assert "account has been deleted" in email.subject.lower()
727 unique_string = "You have successfully deleted your Couchers.org account."
728 assert unique_string in email.plain
729 assert unique_string in email.html
730 assert "7 days" in email.plain
731 assert "7 days" in email.html
732 assert "support@couchers.org" in email.plain
733 assert "support@couchers.org" in email.html
735 push = push_collector.pop_for_user(user_id, last=True)
736 assert push.content.title == "Account deleted"
737 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
739 with session_scope() as session:
740 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
742 user_ = session.execute(select(User).where(User.id == user_id)).scalar_one()
743 assert user_.deleted_at is not None
744 assert user_.undelete_token
745 assert user_.undelete_until
746 assert user_.undelete_until > now()
748 undelete_token = user_.undelete_token
750 undelete_url = f"http://localhost:3000/recover-account?token={undelete_token}"
751 assert undelete_url in email.plain
752 assert undelete_url in email.html
754 with auth_api_session() as (auth_api, metadata_interceptor):
755 auth_api.RecoverAccount(
756 auth_pb2.RecoverAccountReq(
757 token=undelete_token,
758 )
759 )
761 email = email_collector.pop_for_recipient(user.email, last=True)
762 assert email.recipient == user.email
763 assert "account has been recovered" in email.subject.lower()
764 unique_string = "Your Couchers.org account has been successfully recovered."
765 assert unique_string in email.plain
766 assert unique_string in email.html
767 assert "support@couchers.org" in email.plain
768 assert "support@couchers.org" in email.html
770 push = push_collector.pop_for_user(user_id, last=True)
771 assert push.content.title == "Account restored"
772 assert push.content.body == "Welcome back!"
774 with session_scope() as session:
775 assert not session.execute(select(AccountDeletionToken)).scalar_one_or_none()
777 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
778 assert user.deleted_at is None
779 assert not user.undelete_token
780 assert not user.undelete_until
783def test_multiple_delete_tokens(db):
784 """
785 Make sure deletion tokens are deleted on delete
786 """
787 user, token = generate_user()
789 with account_session(token) as account:
790 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
791 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
792 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
794 with session_scope() as session:
795 assert session.execute(select(func.count()).select_from(AccountDeletionToken)).scalar_one() == 3
796 token = session.execute(select(AccountDeletionToken.token).limit(1)).scalar_one()
798 with auth_api_session() as (auth_api, metadata_interceptor):
799 auth_api.ConfirmDeleteAccount(
800 auth_pb2.ConfirmDeleteAccountReq(
801 token=token,
802 )
803 )
805 with session_scope() as session:
806 assert not session.execute(select(AccountDeletionToken.token)).scalar_one_or_none()
809def test_ListActiveSessions_pagination(db, fast_passwords):
810 password = random_hex()
811 user, token = generate_user(hashed_password=hash_password(password))
813 with auth_api_session() as (auth_api, metadata_interceptor):
814 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
815 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
816 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
817 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
819 with real_account_session(token) as account:
820 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_size=3))
821 assert len(res.active_sessions) == 3
822 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq(page_token=res.next_page_token, page_size=3))
823 assert len(res.active_sessions) == 2
824 assert not res.next_page_token
827def test_ListActiveSessions_details(db, fast_passwords):
828 password = random_hex()
829 user, token = generate_user(hashed_password=hash_password(password))
831 ips_user_agents = [
832 (
833 "108.123.33.162",
834 "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",
835 ),
836 (
837 "8.245.212.28",
838 "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",
839 ),
840 (
841 "95.254.140.156",
842 "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:130.0) Gecko/20100101 Firefox/130.0",
843 ),
844 ]
846 for ip, user_agent in ips_user_agents:
847 options = (("grpc.primary_user_agent", user_agent),)
848 with auth_api_session(grpc_channel_options=options) as (auth_api, metadata_interceptor):
849 auth_api.Authenticate(
850 auth_pb2.AuthReq(user=user.username, password=password), metadata=(("x-couchers-real-ip", ip),)
851 )
853 def dummy_geoip(ip_address):
854 return {
855 "108.123.33.162": "Chicago, United States",
856 "8.245.212.28": "Sydney, Australia",
857 }.get(ip_address)
859 with real_account_session(token) as account:
860 with patch("couchers.servicers.account.geoip_approximate_location", dummy_geoip):
861 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
862 print(res)
863 assert len(res.active_sessions) == 4
865 # this one currently making the API call
866 assert res.active_sessions[0].operating_system == "Other"
867 assert res.active_sessions[0].browser == "Other"
868 assert res.active_sessions[0].device == "Other"
869 assert res.active_sessions[0].approximate_location == "Unknown"
870 assert res.active_sessions[0].is_current_session
872 assert res.active_sessions[1].operating_system == "Ubuntu"
873 assert res.active_sessions[1].browser == "Firefox"
874 assert res.active_sessions[1].device == "Other"
875 assert res.active_sessions[1].approximate_location == "Unknown"
876 assert not res.active_sessions[1].is_current_session
878 assert res.active_sessions[2].operating_system == "Android"
879 assert res.active_sessions[2].browser == "Samsung Internet"
880 assert res.active_sessions[2].device == "K"
881 assert res.active_sessions[2].approximate_location == "Sydney, Australia"
882 assert not res.active_sessions[2].is_current_session
884 assert res.active_sessions[3].operating_system == "iOS"
885 assert res.active_sessions[3].browser == "Mobile Safari"
886 assert res.active_sessions[3].device == "iPhone"
887 assert res.active_sessions[3].approximate_location == "Chicago, United States"
888 assert not res.active_sessions[3].is_current_session
891def test_LogOutSession(db, fast_passwords):
892 password = random_hex()
893 user, token = generate_user(hashed_password=hash_password(password))
895 with auth_api_session() as (auth_api, metadata_interceptor):
896 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
897 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
898 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
899 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
901 with real_account_session(token) as account:
902 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
903 assert len(res.active_sessions) == 5
904 account.LogOutSession(account_pb2.LogOutSessionReq(created=res.active_sessions[3].created))
906 res2 = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
907 assert len(res2.active_sessions) == 4
909 # ignore the first session as it changes
910 assert res.active_sessions[1:3] + res.active_sessions[4:] == res2.active_sessions[1:]
913def test_LogOutOtherSessions(db, fast_passwords):
914 password = random_hex()
915 user, token = generate_user(hashed_password=hash_password(password))
917 with auth_api_session() as (auth_api, metadata_interceptor):
918 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
919 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
920 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
921 auth_api.Authenticate(auth_pb2.AuthReq(user=user.username, password=password))
923 with real_account_session(token) as account:
924 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
925 assert len(res.active_sessions) == 5
926 with pytest.raises(grpc.RpcError) as e:
927 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=False))
928 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
929 assert e.value.details() == "Please confirm you want to log out of other sessions."
931 account.LogOutOtherSessions(account_pb2.LogOutOtherSessionsReq(confirm=True))
932 res = account.ListActiveSessions(account_pb2.ListActiveSessionsReq())
933 assert len(res.active_sessions) == 1
936def test_CreateInviteCode(db):
937 user, token = generate_user()
939 with account_session(token) as account:
940 res = account.CreateInviteCode(account_pb2.CreateInviteCodeReq())
941 code = res.code
942 assert len(code) == 8
944 with session_scope() as session:
945 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
946 assert invite.creator_user_id == user.id
947 assert invite.disabled is None
948 assert res.url == urls.invite_code_link(code=res.code)
951def test_DisableInviteCode(db):
952 user, token = generate_user()
954 with account_session(token) as account:
955 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
956 account.DisableInviteCode(account_pb2.DisableInviteCodeReq(code=code))
958 with session_scope() as session:
959 invite = session.execute(select(InviteCode).where(InviteCode.id == code)).scalar_one()
960 assert invite.disabled is not None
963def test_ListInviteCodes(db):
964 user, token = generate_user()
965 another_user, _ = generate_user()
967 with account_session(token) as account:
968 code = account.CreateInviteCode(account_pb2.CreateInviteCodeReq()).code
970 # simulate another_user having signed up with this invite code
971 with session_scope() as session:
972 session.execute(update(User).where(User.id == another_user.id).values(invite_code_id=code))
974 with account_session(token) as account:
975 res = account.ListInviteCodes(empty_pb2.Empty())
976 assert len(res.invite_codes) == 1
977 assert res.invite_codes[0].code == code
978 assert res.invite_codes[0].uses == 1
979 assert res.invite_codes[0].url == urls.invite_code_link(code=code)
982def test_reminders(db, moderator):
983 # reference writing reminders tested in test_AvailableWriteReferences_and_ListPendingReferencesToWrite
984 # we use LiteUser, so remember to refresh materialized views
985 user, token = generate_user(complete_profile=False)
986 complete_user, complete_token = generate_user(complete_profile=True)
987 req_user1, req_user_token1 = generate_user(complete_profile=True)
988 req_user2, req_user_token2 = generate_user(complete_profile=True)
990 refresh_materialized_views_rapid(empty_pb2.Empty())
991 with account_session(complete_token) as account:
992 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
993 with account_session(token) as account:
994 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
995 "complete_profile_reminder",
996 ]
998 today_plus_2 = (today() + timedelta(days=2)).isoformat()
999 today_plus_3 = (today() + timedelta(days=3)).isoformat()
1000 with requests_session(req_user_token1) as api:
1001 host_request1_id = api.CreateHostRequest(
1002 requests_pb2.CreateHostRequestReq(
1003 host_user_id=user.id,
1004 from_date=today_plus_2,
1005 to_date=today_plus_3,
1006 text=valid_request_text("Test request 1"),
1007 )
1008 ).host_request_id
1009 moderator.approve_host_request(host_request1_id)
1011 with account_session(token) as account:
1012 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1013 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1014 "respond_to_host_request_reminder",
1015 "complete_profile_reminder",
1016 ]
1017 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1018 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1020 with requests_session(req_user_token2) as api:
1021 host_request2_id = api.CreateHostRequest(
1022 requests_pb2.CreateHostRequestReq(
1023 host_user_id=user.id,
1024 from_date=today_plus_2,
1025 to_date=today_plus_3,
1026 text=valid_request_text("Test request 2"),
1027 )
1028 ).host_request_id
1029 moderator.approve_host_request(host_request2_id)
1031 refresh_materialized_views_rapid(empty_pb2.Empty())
1032 with account_session(token) as account:
1033 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1034 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1035 "respond_to_host_request_reminder",
1036 "respond_to_host_request_reminder",
1037 "complete_profile_reminder",
1038 ]
1039 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1040 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1041 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1042 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1044 backdate_conversations()
1045 with requests_session(req_user_token1) as api:
1046 host_request3_id = api.CreateHostRequest(
1047 requests_pb2.CreateHostRequestReq(
1048 host_user_id=user.id,
1049 from_date=today_plus_2,
1050 to_date=today_plus_3,
1051 text=valid_request_text("Test request 3"),
1052 )
1053 ).host_request_id
1054 moderator.approve_host_request(host_request3_id)
1056 refresh_materialized_views_rapid(empty_pb2.Empty())
1057 with account_session(token) as account:
1058 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1059 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1060 "respond_to_host_request_reminder",
1061 "respond_to_host_request_reminder",
1062 "respond_to_host_request_reminder",
1063 "complete_profile_reminder",
1064 ]
1065 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request1_id
1066 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1067 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request2_id
1068 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1069 assert reminders[2].respond_to_host_request_reminder.host_request_id == host_request3_id
1070 assert reminders[2].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1072 # accept req
1073 with requests_session(token) as api:
1074 api.RespondHostRequest(
1075 requests_pb2.RespondHostRequestReq(
1076 host_request_id=host_request1_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
1077 )
1078 )
1080 refresh_materialized_views_rapid(empty_pb2.Empty())
1081 with account_session(token) as account:
1082 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1083 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1084 "respond_to_host_request_reminder",
1085 "respond_to_host_request_reminder",
1086 "complete_profile_reminder",
1087 ]
1088 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request2_id
1089 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user2.id
1090 assert reminders[1].respond_to_host_request_reminder.host_request_id == host_request3_id
1091 assert reminders[1].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1093 # host replies to req2 with a message: reminder should clear even though it's still pending
1094 with requests_session(token) as api:
1095 api.SendHostRequestMessage(
1096 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request2_id, text="Let me think about it")
1097 )
1099 refresh_materialized_views_rapid(empty_pb2.Empty())
1100 with account_session(token) as account:
1101 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1102 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1103 "respond_to_host_request_reminder",
1104 "complete_profile_reminder",
1105 ]
1106 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1107 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1109 # surfer sending a message should not clear the reminder
1110 with requests_session(req_user_token1) as api:
1111 api.SendHostRequestMessage(
1112 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request3_id, text="Any update?")
1113 )
1115 refresh_materialized_views_rapid(empty_pb2.Empty())
1116 with account_session(token) as account:
1117 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1118 assert [reminder.WhichOneof("reminder") for reminder in reminders] == [
1119 "respond_to_host_request_reminder",
1120 "complete_profile_reminder",
1121 ]
1122 assert reminders[0].respond_to_host_request_reminder.host_request_id == host_request3_id
1123 assert reminders[0].respond_to_host_request_reminder.surfer_user.user_id == req_user1.id
1126def test_confirm_host_request_reminder(db, moderator):
1127 host, host_token = generate_user(complete_profile=True)
1128 surfer, surfer_token = generate_user(complete_profile=True)
1130 today_plus_10 = (today() + timedelta(days=10)).isoformat()
1131 today_plus_12 = (today() + timedelta(days=12)).isoformat()
1133 with requests_session(surfer_token) as api:
1134 host_request_id = api.CreateHostRequest(
1135 requests_pb2.CreateHostRequestReq(
1136 host_user_id=host.id,
1137 from_date=today_plus_10,
1138 to_date=today_plus_12,
1139 text=valid_request_text("Please host me"),
1140 )
1141 ).host_request_id
1142 moderator.approve_host_request(host_request_id)
1144 refresh_materialized_views_rapid(empty_pb2.Empty())
1145 with account_session(surfer_token) as account:
1146 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1148 with requests_session(host_token) as api:
1149 api.RespondHostRequest(
1150 requests_pb2.RespondHostRequestReq(
1151 host_request_id=host_request_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
1152 )
1153 )
1155 refresh_materialized_views_rapid(empty_pb2.Empty())
1156 with account_session(surfer_token) as account:
1157 reminders = account.GetReminders(empty_pb2.Empty()).reminders
1158 assert [reminder.WhichOneof("reminder") for reminder in reminders] == ["confirm_host_request_reminder"]
1159 assert reminders[0].confirm_host_request_reminder.host_request_id == host_request_id
1160 assert reminders[0].confirm_host_request_reminder.host_user.user_id == host.id
1162 # after surfer confirms, reminder should clear
1163 with requests_session(surfer_token) as api:
1164 api.RespondHostRequest(
1165 requests_pb2.RespondHostRequestReq(
1166 host_request_id=host_request_id, status=messages_pb2.HOST_REQUEST_STATUS_CONFIRMED
1167 )
1168 )
1170 refresh_materialized_views_rapid(empty_pb2.Empty())
1171 with account_session(surfer_token) as account:
1172 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1175def test_my_home_reminder(db):
1176 # can_host with incomplete my home (max_guests not set) → reminder shown
1177 can_host_incomplete, token1 = generate_user(hosting_status=HostingStatus.can_host)
1178 # maybe with incomplete my home → reminder shown
1179 maybe_incomplete, token2 = generate_user(hosting_status=HostingStatus.maybe)
1180 # cant_host → no reminder regardless of my home completion
1181 cant_host, token3 = generate_user(hosting_status=HostingStatus.cant_host)
1182 # can_host with fully completed my home → no reminder
1183 can_host_complete, token4 = generate_user(
1184 hosting_status=HostingStatus.can_host,
1185 max_guests=2,
1186 sleeping_arrangement=SleepingArrangement.private,
1187 # about_place is set by default in make_user
1188 )
1190 with account_session(token1) as account:
1191 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1192 "complete_my_home_reminder",
1193 ]
1195 with account_session(token2) as account:
1196 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == [
1197 "complete_my_home_reminder",
1198 ]
1200 with account_session(token3) as account:
1201 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1203 with account_session(token4) as account:
1204 assert [reminder.WhichOneof("reminder") for reminder in account.GetReminders(empty_pb2.Empty()).reminders] == []
1207def test_volunteer_stuff(db):
1208 # taken from couchers/app/backend/resources/badges.json
1209 board_member_id = 8347
1211 # with password
1212 user, token = generate_user(name="Von Tester", username="tester", city="Amsterdam", id=board_member_id)
1214 with account_session(token) as account:
1215 res = account.GetAccountInfo(empty_pb2.Empty())
1216 assert not res.is_volunteer
1218 with pytest.raises(grpc.RpcError) as e:
1219 account.GetMyVolunteerInfo(empty_pb2.Empty())
1220 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1221 assert (
1222 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1223 )
1225 with pytest.raises(grpc.RpcError) as e:
1226 account.UpdateMyVolunteerInfo(account_pb2.UpdateMyVolunteerInfoReq())
1227 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1228 assert (
1229 e.value.details() == "You are currently not registered as a volunteer, if this is wrong, please contact us."
1230 )
1232 with session_scope() as session:
1233 session.add(
1234 make_volunteer(
1235 user_id=user.id,
1236 display_name="Great Volunteer",
1237 display_location="The Bitbucket",
1238 role="Lead Tester",
1239 started_volunteering=date(2020, 6, 1),
1240 )
1241 )
1243 with account_session(token) as account:
1244 res = account.GetAccountInfo(empty_pb2.Empty())
1245 assert res.is_volunteer
1247 res = account.GetMyVolunteerInfo(empty_pb2.Empty())
1249 assert res.display_name == "Great Volunteer"
1250 assert res.display_location == "The Bitbucket"
1251 assert res.role == "Lead Tester"
1252 assert res.started_volunteering == "2020-06-01"
1253 assert not res.stopped_volunteering
1254 assert res.show_on_team_page
1255 assert res.link_type == "couchers"
1256 assert res.link_text == "@tester"
1257 assert res.link_url == "http://localhost:3000/user/tester"
1259 res = account.UpdateMyVolunteerInfo(
1260 account_pb2.UpdateMyVolunteerInfoReq(
1261 display_name=wrappers_pb2.StringValue(value=""),
1262 link_type=wrappers_pb2.StringValue(value="website"),
1263 link_text=wrappers_pb2.StringValue(value="testervontester.com.invalid"),
1264 link_url=wrappers_pb2.StringValue(value="https://www.testervontester.com.invalid/"),
1265 )
1266 )
1268 assert res.display_name == ""
1269 assert res.display_location == "The Bitbucket"
1270 assert res.role == "Lead Tester"
1271 assert res.started_volunteering == "2020-06-01"
1272 assert not res.stopped_volunteering
1273 assert res.show_on_team_page
1274 assert res.link_type == "website"
1275 assert res.link_text == "testervontester.com.invalid"
1276 assert res.link_url == "https://www.testervontester.com.invalid/"
1277 res = account.UpdateMyVolunteerInfo(
1278 account_pb2.UpdateMyVolunteerInfoReq(
1279 display_name=wrappers_pb2.StringValue(value=""),
1280 link_type=wrappers_pb2.StringValue(value="linkedin"),
1281 link_text=wrappers_pb2.StringValue(value="tester-vontester"),
1282 )
1283 )
1284 assert res.display_name == ""
1285 assert res.display_location == "The Bitbucket"
1286 assert res.role == "Lead Tester"
1287 assert res.started_volunteering == "2020-06-01"
1288 assert not res.stopped_volunteering
1289 assert res.show_on_team_page
1290 assert res.link_type == "linkedin"
1291 assert res.link_text == "tester-vontester"
1292 assert res.link_url == "https://www.linkedin.com/in/tester-vontester/"
1294 res = account.UpdateMyVolunteerInfo(
1295 account_pb2.UpdateMyVolunteerInfoReq(
1296 display_name=wrappers_pb2.StringValue(value="Tester"),
1297 display_location=wrappers_pb2.StringValue(value=""),
1298 link_type=wrappers_pb2.StringValue(value="email"),
1299 link_text=wrappers_pb2.StringValue(value="tester@vontester.com.invalid"),
1300 )
1301 )
1302 assert res.display_name == "Tester"
1303 assert res.display_location == ""
1304 assert res.role == "Lead Tester"
1305 assert res.started_volunteering == "2020-06-01"
1306 assert not res.stopped_volunteering
1307 assert res.show_on_team_page
1308 assert res.link_type == "email"
1309 assert res.link_text == "tester@vontester.com.invalid"
1310 assert res.link_url == "mailto:tester@vontester.com.invalid"
1312 refresh_materialized_views_rapid(empty_pb2.Empty())
1314 with public_session() as public:
1315 res = public.GetVolunteers(empty_pb2.Empty())
1316 assert len(res.current_volunteers) == 1
1317 v = res.current_volunteers[0]
1318 assert v.name == "Tester"
1319 assert v.username == "tester"
1320 assert v.is_board_member
1321 assert v.role == "Lead Tester"
1322 assert v.location == "Amsterdam"
1323 assert v.img.startswith("http://localhost:5001/img/thumbnail/")
1324 assert v.link_type == "email"
1325 assert v.link_text == "tester@vontester.com.invalid"
1326 assert v.link_url == "mailto:tester@vontester.com.invalid"