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