Coverage for app/backend/src/tests/test_public.py: 100%
313 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import json
2from datetime import UTC, date, datetime, timedelta
3from math import sqrt
4from unittest.mock import patch
6import grpc
7import pytest
8from google.protobuf import empty_pb2
9from sqlalchemy import func, select, update
11from couchers.db import session_scope
12from couchers.jobs.enqueue import queue_job
13from couchers.jobs.handlers import update_randomized_locations
14from couchers.materialized_views import refresh_materialized_views_rapid
15from couchers.models import (
16 Invoice,
17 InvoiceType,
18 ModerationObjectType,
19 ModerationState,
20 ModerationVisibility,
21 ProfilePublicVisibility,
22 Reference,
23 ReferenceType,
24 User,
25)
26from couchers.proto import api_pb2, public_pb2
27from couchers.servicers.public import _get_donation_stats, _get_public_users, _get_signup_page_info, _get_volunteers
28from couchers.utils import now
29from tests.fixtures.db import generate_user, make_friends, make_volunteer
30from tests.fixtures.misc import process_jobs
31from tests.fixtures.sessions import public_session
32from tests.test_references import create_friend_reference, create_host_reference
35def test_GetPublicMapLayer(db):
36 user1, _ = generate_user()
37 user2, _ = generate_user(username="user2", public_visibility=ProfilePublicVisibility.nothing)
38 user3, _ = generate_user()
39 user4, _ = generate_user(username="user4", public_visibility=ProfilePublicVisibility.limited)
40 user5, _ = generate_user()
42 # these are hardcoded in test_fixtures
43 test_user_coordinates = [-73.9740, 40.7108]
45 with session_scope() as session:
46 queue_job(session, job=update_randomized_locations, payload=empty_pb2.Empty())
48 process_jobs()
50 with public_session() as public:
51 http_body = public.GetPublicUsers(empty_pb2.Empty())
52 assert http_body.content_type == "application/json"
53 data = json.loads(http_body.data)
54 # Sort to ensure a deterministic order
55 data["features"].sort(key=lambda f: f["geometry"]["coordinates"][0])
56 assert data == {
57 "type": "FeatureCollection",
58 "features": [
59 {
60 "type": "Feature",
61 "geometry": {"type": "Point", "coordinates": [-74.042643848, 40.706241098]},
62 "properties": {"username": None},
63 },
64 {
65 "type": "Feature",
66 "geometry": {"type": "Point", "coordinates": [-73.974, 40.7108]},
67 "properties": {"username": "user4"},
68 },
69 {
70 "type": "Feature",
71 "geometry": {"type": "Point", "coordinates": [-73.955417734, 40.691831306]},
72 "properties": {"username": None},
73 },
74 {
75 "type": "Feature",
76 "geometry": {"type": "Point", "coordinates": [-73.928380198, 40.729706144]},
77 "properties": {"username": None},
78 },
79 ],
80 }
82 for user in data["features"]:
83 coords = user["geometry"]["coordinates"]
84 if user["properties"]["username"]:
85 assert coords == test_user_coordinates
86 else:
87 xdiff = coords[0] - test_user_coordinates[0]
88 ydiff = coords[1] - test_user_coordinates[1]
89 dist = sqrt(xdiff**2 + ydiff**2)
90 assert dist > 0.02 and dist < 0.1
93def test_GetPublicMapLayer_excludes_shadowed(db):
94 """Test GetPublicUsers excludes shadowed users from the public map"""
96 _get_public_users.cache_clear()
98 generate_user(username="visible", public_visibility=ProfilePublicVisibility.limited)
99 shadowed_user, _ = generate_user(username="shadowed", public_visibility=ProfilePublicVisibility.limited)
101 with session_scope() as session:
102 session.execute(select(User).where(User.id == shadowed_user.id)).scalar_one().shadowed_at = now()
104 with public_session() as public:
105 data = json.loads(public.GetPublicUsers(empty_pb2.Empty()).data)
107 assert {feature["properties"]["username"] for feature in data["features"]} == {"visible"}
110def test_GetDonationStats_empty(db, feature_flags):
111 """Test GetDonationStats with no donations returns zero and goal"""
112 _get_donation_stats.cache_clear()
114 feature_flags.set("donation_goal_usd", 2500)
115 feature_flags.set("donation_offset_usd", 700)
116 with public_session() as public:
117 res = public.GetDonationStats(empty_pb2.Empty())
118 assert res.total_donated_ytd == 0
119 assert res.goal == 2500
122def test_GetDonationStats_with_donations(db, feature_flags):
123 """Test GetDonationStats sums on_platform donations correctly"""
124 _get_donation_stats.cache_clear()
125 user, _ = generate_user()
127 with session_scope() as session:
128 # Add some on_platform donations (should be counted)
129 session.add(
130 Invoice(
131 user_id=user.id,
132 amount=100,
133 stripe_payment_intent_id="pi_test_1",
134 stripe_receipt_url="https://example.com/receipt/1",
135 invoice_type=InvoiceType.on_platform,
136 )
137 )
138 session.add(
139 Invoice(
140 user_id=user.id,
141 amount=250,
142 stripe_payment_intent_id="pi_test_2",
143 stripe_receipt_url="https://example.com/receipt/2",
144 invoice_type=InvoiceType.on_platform,
145 )
146 )
147 session.add(
148 Invoice(
149 user_id=user.id,
150 amount=500,
151 stripe_payment_intent_id="pi_test_3",
152 stripe_receipt_url="https://example.com/receipt/3",
153 invoice_type=InvoiceType.on_platform,
154 )
155 )
157 feature_flags.set("donation_goal_usd", 5000)
158 feature_flags.set("donation_offset_usd", 0)
159 with public_session() as public:
160 res = public.GetDonationStats(empty_pb2.Empty())
161 assert res.total_donated_ytd == 850
162 assert res.goal == 5000
165def test_GetDonationStats_excludes_merch(db, feature_flags):
166 """Test GetDonationStats excludes external_shop (merch) invoices"""
167 _get_donation_stats.cache_clear()
168 user, _ = generate_user()
170 with session_scope() as session:
171 # Add on_platform donation (should be counted)
172 session.add(
173 Invoice(
174 user_id=user.id,
175 amount=200,
176 stripe_payment_intent_id="pi_test_donation",
177 stripe_receipt_url="https://example.com/receipt/donation",
178 invoice_type=InvoiceType.on_platform,
179 )
180 )
181 # Add external_shop/merch purchase (should NOT be counted)
182 session.add(
183 Invoice(
184 user_id=user.id,
185 amount=50,
186 stripe_payment_intent_id="pi_test_merch",
187 stripe_receipt_url="https://example.com/receipt/merch",
188 invoice_type=InvoiceType.external_shop,
189 )
190 )
192 feature_flags.set("donation_goal_usd", 5000)
193 feature_flags.set("donation_offset_usd", 0)
194 with public_session() as public:
195 res = public.GetDonationStats(empty_pb2.Empty())
196 # Should only count the on_platform donation, not the merch
197 assert res.total_donated_ytd == 200
198 assert res.goal == 5000
201def test_GetDonationStats_excludes_previous_years(db, feature_flags):
202 """Test GetDonationStats only counts current year donations"""
203 _get_donation_stats.cache_clear()
204 user, _ = generate_user()
206 with session_scope() as session:
207 # Add donation from this year (should be counted)
208 session.add(
209 Invoice(
210 user_id=user.id,
211 amount=300,
212 stripe_payment_intent_id="pi_test_this_year",
213 stripe_receipt_url="https://example.com/receipt/this_year",
214 invoice_type=InvoiceType.on_platform,
215 )
216 )
217 # Add donation from last year (should NOT be counted)
218 last_year = datetime(datetime.now(UTC).year - 1, 6, 15, tzinfo=UTC)
219 invoice = Invoice(
220 user_id=user.id,
221 amount=1000,
222 stripe_payment_intent_id="pi_test_last_year",
223 stripe_receipt_url="https://example.com/receipt/last_year",
224 invoice_type=InvoiceType.on_platform,
225 )
226 session.add(invoice)
227 session.flush()
228 # Manually set the created date to last year
229 invoice.created = last_year
231 feature_flags.set("donation_goal_usd", 5000)
232 feature_flags.set("donation_offset_usd", 0)
233 with public_session() as public:
234 res = public.GetDonationStats(empty_pb2.Empty())
235 # Should only count this year's donation
236 assert res.total_donated_ytd == 300
237 assert res.goal == 5000
240def test_GetDonationStats_uses_flags(db, feature_flags):
241 """Goal and offset come from the donation_goal_usd / donation_offset_usd flags when configured"""
242 _get_donation_stats.cache_clear()
243 user, _ = generate_user()
245 with session_scope() as session:
246 session.add(
247 Invoice(
248 user_id=user.id,
249 amount=1000,
250 stripe_payment_intent_id="pi_test_flag",
251 stripe_receipt_url="https://example.com/receipt/flag",
252 invoice_type=InvoiceType.on_platform,
253 )
254 )
256 feature_flags.set("donation_goal_usd", 12000)
257 feature_flags.set("donation_offset_usd", 300)
259 with public_session() as public:
260 res = public.GetDonationStats(empty_pb2.Empty())
261 assert res.goal == 12000
262 assert res.total_donated_ytd == 700 # 1000 donated minus the 300 offset
264 _get_donation_stats.cache_clear()
267def test_GetVolunteers_mixed_current_and_past(db):
268 """Test GetVolunteers with both current and past volunteers"""
270 _get_volunteers.cache_clear()
272 current1, _ = generate_user(username="current1")
273 current2, _ = generate_user(username="current2")
274 past1, _ = generate_user(username="past1")
275 past2, _ = generate_user(username="past2")
277 with session_scope() as session:
278 session.add(
279 make_volunteer(
280 user_id=current1.id,
281 role="Current Role 1",
282 started_volunteering=date(2023, 1, 1),
283 )
284 )
285 session.add(
286 make_volunteer(
287 user_id=current2.id,
288 role="Current Role 2",
289 started_volunteering=date(2024, 1, 1),
290 )
291 )
292 session.add(
293 make_volunteer(
294 user_id=past1.id,
295 role="Past Role 1",
296 started_volunteering=date(2020, 1, 1),
297 stopped_volunteering=date(2022, 6, 1),
298 )
299 )
300 session.add(
301 make_volunteer(
302 user_id=past2.id,
303 role="Past Role 2",
304 started_volunteering=date(2021, 1, 1),
305 stopped_volunteering=date(2023, 12, 31),
306 )
307 )
309 refresh_materialized_views_rapid(empty_pb2.Empty())
311 with public_session() as public:
312 res = public.GetVolunteers(empty_pb2.Empty())
313 assert len(res.current_volunteers) == 2
314 assert len(res.past_volunteers) == 2
316 # Past volunteers are sorted by stopped_volunteering descending
317 assert res.past_volunteers[0].username == "past2"
318 assert res.past_volunteers[1].username == "past1"
321def test_GetVolunteers_custom_sort_key(db):
322 """Test GetVolunteers respects custom sort_key"""
324 _get_volunteers.cache_clear()
326 user1, _ = generate_user(username="user1")
327 user2, _ = generate_user(username="user2")
328 user3, _ = generate_user(username="user3")
330 with session_scope() as session:
331 # user2 should be first (lowest sort_key)
332 session.add(
333 make_volunteer(
334 user_id=user2.id,
335 role="Role 2",
336 started_volunteering=date(2023, 3, 1),
337 sort_key=1.0,
338 )
339 )
340 # user3 should be second
341 session.add(
342 make_volunteer(
343 user_id=user3.id,
344 role="Role 3",
345 started_volunteering=date(2023, 1, 1),
346 sort_key=2.0,
347 )
348 )
349 # user1 should be last (no sort_key, falls back to started_volunteering)
350 session.add(
351 make_volunteer(
352 user_id=user1.id,
353 role="Role 1",
354 started_volunteering=date(2023, 2, 1),
355 )
356 )
358 refresh_materialized_views_rapid(empty_pb2.Empty())
360 with public_session() as public:
361 res = public.GetVolunteers(empty_pb2.Empty())
362 assert len(res.current_volunteers) == 3
363 assert res.current_volunteers[0].username == "user2"
364 assert res.current_volunteers[1].username == "user3"
365 assert res.current_volunteers[2].username == "user1"
368def test_GetVolunteers_excludes_hidden(db):
369 """Test GetVolunteers excludes volunteers with show_on_team_page=False"""
371 _get_volunteers.cache_clear()
373 user1, _ = generate_user(username="visible")
374 user2, _ = generate_user(username="hidden")
376 with session_scope() as session:
377 session.add(
378 make_volunteer(
379 user_id=user1.id,
380 role="Visible Role",
381 started_volunteering=date(2023, 1, 1),
382 )
383 )
384 session.add(
385 make_volunteer(
386 user_id=user2.id,
387 role="Hidden Role",
388 started_volunteering=date(2023, 1, 1),
389 show_on_team_page=False,
390 )
391 )
393 refresh_materialized_views_rapid(empty_pb2.Empty())
395 with public_session() as public:
396 res = public.GetVolunteers(empty_pb2.Empty())
397 assert len(res.current_volunteers) == 1
398 assert res.current_volunteers[0].username == "visible"
401def test_GetVolunteers_link_types(db):
402 """Test GetVolunteers handles different link types"""
404 _get_volunteers.cache_clear()
406 user_default, _ = generate_user(username="default_link")
407 user_custom, _ = generate_user(username="custom_link")
409 with session_scope() as session:
410 # Volunteer with default couchers link
411 session.add(
412 make_volunteer(
413 user_id=user_default.id,
414 role="Default Link",
415 started_volunteering=date(2023, 1, 1),
416 )
417 )
418 # Volunteer with custom link
419 session.add(
420 make_volunteer(
421 user_id=user_custom.id,
422 role="Custom Link",
423 started_volunteering=date(2023, 1, 1),
424 link_type="email",
425 link_text="contact@example.com",
426 link_url="mailto:contact@example.com",
427 )
428 )
430 refresh_materialized_views_rapid(empty_pb2.Empty())
432 with public_session() as public:
433 res = public.GetVolunteers(empty_pb2.Empty())
434 assert len(res.current_volunteers) == 2
436 # Check default link
437 default_vol = next(v for v in res.current_volunteers if v.username == "default_link")
438 assert default_vol.link_type == "couchers"
439 assert default_vol.link_text == "@default_link"
440 assert "default_link" in default_vol.link_url
442 # Check custom link
443 custom_vol = next(v for v in res.current_volunteers if v.username == "custom_link")
444 assert custom_vol.link_type == "email"
445 assert custom_vol.link_text == "contact@example.com"
446 assert custom_vol.link_url == "mailto:contact@example.com"
449def test_GetVolunteers_board_member_flag(db):
450 """Test GetVolunteers correctly identifies board members"""
452 _get_volunteers.cache_clear()
454 board_member, _ = generate_user(username="board_member")
455 regular_volunteer, _ = generate_user(username="regular")
457 with session_scope() as session:
458 session.add(
459 make_volunteer(
460 user_id=board_member.id,
461 role="Board Member Role",
462 started_volunteering=date(2023, 1, 1),
463 )
464 )
465 session.add(
466 make_volunteer(
467 user_id=regular_volunteer.id,
468 role="Regular Role",
469 started_volunteering=date(2023, 1, 1),
470 )
471 )
473 refresh_materialized_views_rapid(empty_pb2.Empty())
475 # Mock the static badge dict to include board_member
476 with patch("couchers.servicers.public.get_static_badge_dict", return_value={"board_member": [board_member.id]}):
477 with public_session() as public:
478 res = public.GetVolunteers(empty_pb2.Empty())
479 assert len(res.current_volunteers) == 2
481 board_vol = next(v for v in res.current_volunteers if v.username == "board_member")
482 assert board_vol.is_board_member is True
484 regular_vol = next(v for v in res.current_volunteers if v.username == "regular")
485 assert regular_vol.is_board_member is False
488def test_GetSignupPageInfo(db):
489 """Test GetSignupPageInfo returns a correct user count and last signup info"""
491 _get_signup_page_info.cache_clear()
493 user1, _ = generate_user(username="user1")
494 user2, _ = generate_user(username="user2")
495 user3, _ = generate_user(username="user3")
497 refresh_materialized_views_rapid(empty_pb2.Empty())
499 with public_session() as public:
500 res = public.GetSignupPageInfo(empty_pb2.Empty())
501 # user3 should be the last signup (highest id)
502 assert res.user_count >= 3
503 assert res.last_location # Should have some location
504 assert res.last_signup # Should have a timestamp
507def test_GetSignupPageInfo_excludes_invisible_users(db):
508 """Test GetSignupPageInfo excludes deleted/banned users from count"""
509 _get_signup_page_info.cache_clear()
511 visible_user, _ = generate_user(username="visible")
512 deleted_user, _ = generate_user(username="deleted", delete_user=True)
514 with public_session() as public:
515 res = public.GetSignupPageInfo(empty_pb2.Empty())
516 # Deleted user should not be counted or be the last signup
517 assert res.user_count >= 1
520def test_GetPublicUser_not_found(db):
521 """Test GetPublicUser returns NOT_FOUND for nonexistent user"""
522 with public_session() as public:
523 with pytest.raises(grpc.RpcError) as exc:
524 public.GetPublicUser(public_pb2.GetPublicUserReq(user="nonexistent_user"))
525 assert exc.value.code() == grpc.StatusCode.NOT_FOUND
528def test_GetPublicUser_invisible_user(db):
529 """Test GetPublicUser returns NOT_FOUND for deleted/banned user"""
530 deleted_user, _ = generate_user(username="deleted", delete_user=True)
532 with public_session() as public:
533 with pytest.raises(grpc.RpcError) as exc:
534 public.GetPublicUser(public_pb2.GetPublicUserReq(user="deleted"))
535 assert exc.value.code() == grpc.StatusCode.NOT_FOUND
538def test_GetPublicUser_shadowed_user(db):
539 """Test GetPublicUser returns NOT_FOUND for a shadowed user"""
540 shadowed_user, _ = generate_user(username="shadowed", public_visibility=ProfilePublicVisibility.full)
542 with session_scope() as session:
543 session.execute(select(User).where(User.id == shadowed_user.id)).scalar_one().shadowed_at = now()
545 with public_session() as public:
546 with pytest.raises(grpc.RpcError) as exc:
547 public.GetPublicUser(public_pb2.GetPublicUserReq(user="shadowed"))
548 assert exc.value.code() == grpc.StatusCode.NOT_FOUND
551def test_GetPublicUser_limited_visibility(db):
552 """Test GetPublicUser returns limited_user for user with limited visibility"""
554 user, _ = generate_user(
555 username="limited_user",
556 name="Limited User",
557 public_visibility=ProfilePublicVisibility.limited,
558 )
560 # Add a reference to test reference counting
561 referrer, _ = generate_user(username="referrer")
562 with session_scope() as session:
563 moderation_state = ModerationState(
564 object_type=ModerationObjectType.reference,
565 object_id=0,
566 visibility=ModerationVisibility.visible,
567 )
568 session.add(moderation_state)
569 session.flush()
570 reference = Reference(
571 from_user_id=referrer.id,
572 to_user_id=user.id,
573 reference_type=ReferenceType.friend,
574 text="Great host!",
575 rating=0.8,
576 was_appropriate=True,
577 moderation_state_id=moderation_state.id,
578 )
579 session.add(reference)
580 session.flush()
581 moderation_state.object_id = reference.id
583 with public_session() as public:
584 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="limited_user"))
585 assert res.HasField("limited_user")
586 assert res.limited_user.username == "limited_user"
587 assert res.limited_user.name == "Limited User"
588 assert res.limited_user.city == "Testing city"
589 assert res.limited_user.hometown == "Test hometown"
590 assert res.limited_user.num_references == 1
591 assert res.limited_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST
592 assert len(res.limited_user.badges) == 0
595def test_GetPublicUser_most_visibility(db):
596 """Test GetPublicUser returns most_user for user with most visibility"""
597 user, _ = generate_user(
598 username="most_user",
599 name="Most User",
600 public_visibility=ProfilePublicVisibility.most,
601 )
603 with public_session() as public:
604 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="most_user"))
605 assert res.HasField("most_user")
606 assert res.most_user.username == "most_user"
607 assert res.most_user.name == "Most User"
608 assert res.most_user.city == "Testing city"
609 assert res.most_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST
612def test_GetPublicUser_full_visibility(db):
613 """Test GetPublicUser returns full_user for user with full visibility"""
614 _get_public_users.cache_clear()
616 user, _ = generate_user(
617 username="full_user",
618 name="Full User",
619 public_visibility=ProfilePublicVisibility.full,
620 )
622 with public_session() as public:
623 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="full_user"))
624 assert res.HasField("full_user")
625 assert res.full_user.username == "full_user"
626 assert res.full_user.name == "Full User"
627 assert res.full_user.city == "Testing city"
628 # Full user should have all the fields from the complete user profile
629 assert res.full_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST
632def test_GetPublicUser_num_references_visibility_rules(db):
633 """The public reference count follows the same visibility rules as the reference list"""
634 user, _ = generate_user(public_visibility=ProfilePublicVisibility.limited)
635 friend_referrer, _ = generate_user()
636 recent_host, _ = generate_user()
637 deleted_referrer1, _ = generate_user()
638 deleted_referrer2, _ = generate_user()
639 shadowed_referrer, _ = generate_user()
641 with session_scope() as session:
642 # counted: a normal friend reference
643 create_friend_reference(session, friend_referrer.id, user.id, timedelta(days=15))
644 # not counted: a recent stay where the reciprocal reference hasn't been written yet
645 create_host_reference(session, recent_host.id, user.id, timedelta(days=3), surfing=False)
646 # counted: references from deleted users remain visible (two of them, so that miscounting
647 # deleted authors can't coincidentally cancel out against the hidden recent stay above)
648 create_friend_reference(session, deleted_referrer1.id, user.id, timedelta(days=16))
649 create_host_reference(session, deleted_referrer2.id, user.id, timedelta(days=30), surfing=False)
650 # not counted: references from shadowed users are hidden from others
651 create_friend_reference(session, shadowed_referrer.id, user.id, timedelta(days=17))
652 for deleted in (deleted_referrer1, deleted_referrer2):
653 session.execute(update(User).where(User.username == deleted.username).values(deleted_at=func.now()))
654 session.execute(update(User).where(User.username == shadowed_referrer.username).values(shadowed_at=func.now()))
656 with public_session() as public:
657 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username))
658 assert res.limited_user.num_references == 3
661def test_GetPublicUser_full_visibility_uses_callers_context(db):
662 """A logged-in caller sees the profile as it looks to them, not the anonymous view"""
663 user, _ = generate_user(public_visibility=ProfilePublicVisibility.full)
664 viewer, viewer_token = generate_user()
665 make_friends(user, viewer)
667 with public_session() as public:
668 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username))
669 assert res.full_user.friends == api_pb2.User.FriendshipStatus.NA
671 with public_session(viewer_token) as public:
672 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username))
673 assert res.full_user.friends == api_pb2.User.FriendshipStatus.FRIENDS
676def test_GetPublicUser_num_references_uses_callers_context(db):
677 """A shadowed author still sees their own reference counted, so the shadow ban isn't leaked"""
678 user, _ = generate_user(public_visibility=ProfilePublicVisibility.limited)
679 shadowed_referrer, shadowed_token = generate_user()
681 with session_scope() as session:
682 create_friend_reference(session, shadowed_referrer.id, user.id, timedelta(days=15))
683 session.execute(update(User).where(User.username == shadowed_referrer.username).values(shadowed_at=func.now()))
685 with public_session() as public:
686 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username))
687 assert res.limited_user.num_references == 0
689 with public_session(shadowed_token) as public:
690 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username))
691 assert res.limited_user.num_references == 1