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

1import json 

2from datetime import UTC, date, datetime, timedelta 

3from math import sqrt 

4from unittest.mock import patch 

5 

6import grpc 

7import pytest 

8from google.protobuf import empty_pb2 

9from sqlalchemy import func, select, update 

10 

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 

33 

34 

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() 

41 

42 # these are hardcoded in test_fixtures 

43 test_user_coordinates = [-73.9740, 40.7108] 

44 

45 with session_scope() as session: 

46 queue_job(session, job=update_randomized_locations, payload=empty_pb2.Empty()) 

47 

48 process_jobs() 

49 

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 } 

81 

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 

91 

92 

93def test_GetPublicMapLayer_excludes_shadowed(db): 

94 """Test GetPublicUsers excludes shadowed users from the public map""" 

95 

96 _get_public_users.cache_clear() 

97 

98 generate_user(username="visible", public_visibility=ProfilePublicVisibility.limited) 

99 shadowed_user, _ = generate_user(username="shadowed", public_visibility=ProfilePublicVisibility.limited) 

100 

101 with session_scope() as session: 

102 session.execute(select(User).where(User.id == shadowed_user.id)).scalar_one().shadowed_at = now() 

103 

104 with public_session() as public: 

105 data = json.loads(public.GetPublicUsers(empty_pb2.Empty()).data) 

106 

107 assert {feature["properties"]["username"] for feature in data["features"]} == {"visible"} 

108 

109 

110def test_GetDonationStats_empty(db, feature_flags): 

111 """Test GetDonationStats with no donations returns zero and goal""" 

112 _get_donation_stats.cache_clear() 

113 

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 

120 

121 

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() 

126 

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 ) 

156 

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 

163 

164 

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() 

169 

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 ) 

191 

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 

199 

200 

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() 

205 

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 

230 

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 

238 

239 

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() 

244 

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 ) 

255 

256 feature_flags.set("donation_goal_usd", 12000) 

257 feature_flags.set("donation_offset_usd", 300) 

258 

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 

263 

264 _get_donation_stats.cache_clear() 

265 

266 

267def test_GetVolunteers_mixed_current_and_past(db): 

268 """Test GetVolunteers with both current and past volunteers""" 

269 

270 _get_volunteers.cache_clear() 

271 

272 current1, _ = generate_user(username="current1") 

273 current2, _ = generate_user(username="current2") 

274 past1, _ = generate_user(username="past1") 

275 past2, _ = generate_user(username="past2") 

276 

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 ) 

308 

309 refresh_materialized_views_rapid(empty_pb2.Empty()) 

310 

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 

315 

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" 

319 

320 

321def test_GetVolunteers_custom_sort_key(db): 

322 """Test GetVolunteers respects custom sort_key""" 

323 

324 _get_volunteers.cache_clear() 

325 

326 user1, _ = generate_user(username="user1") 

327 user2, _ = generate_user(username="user2") 

328 user3, _ = generate_user(username="user3") 

329 

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 ) 

357 

358 refresh_materialized_views_rapid(empty_pb2.Empty()) 

359 

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" 

366 

367 

368def test_GetVolunteers_excludes_hidden(db): 

369 """Test GetVolunteers excludes volunteers with show_on_team_page=False""" 

370 

371 _get_volunteers.cache_clear() 

372 

373 user1, _ = generate_user(username="visible") 

374 user2, _ = generate_user(username="hidden") 

375 

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 ) 

392 

393 refresh_materialized_views_rapid(empty_pb2.Empty()) 

394 

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" 

399 

400 

401def test_GetVolunteers_link_types(db): 

402 """Test GetVolunteers handles different link types""" 

403 

404 _get_volunteers.cache_clear() 

405 

406 user_default, _ = generate_user(username="default_link") 

407 user_custom, _ = generate_user(username="custom_link") 

408 

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 ) 

429 

430 refresh_materialized_views_rapid(empty_pb2.Empty()) 

431 

432 with public_session() as public: 

433 res = public.GetVolunteers(empty_pb2.Empty()) 

434 assert len(res.current_volunteers) == 2 

435 

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 

441 

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" 

447 

448 

449def test_GetVolunteers_board_member_flag(db): 

450 """Test GetVolunteers correctly identifies board members""" 

451 

452 _get_volunteers.cache_clear() 

453 

454 board_member, _ = generate_user(username="board_member") 

455 regular_volunteer, _ = generate_user(username="regular") 

456 

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 ) 

472 

473 refresh_materialized_views_rapid(empty_pb2.Empty()) 

474 

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 

480 

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 

483 

484 regular_vol = next(v for v in res.current_volunteers if v.username == "regular") 

485 assert regular_vol.is_board_member is False 

486 

487 

488def test_GetSignupPageInfo(db): 

489 """Test GetSignupPageInfo returns a correct user count and last signup info""" 

490 

491 _get_signup_page_info.cache_clear() 

492 

493 user1, _ = generate_user(username="user1") 

494 user2, _ = generate_user(username="user2") 

495 user3, _ = generate_user(username="user3") 

496 

497 refresh_materialized_views_rapid(empty_pb2.Empty()) 

498 

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 

505 

506 

507def test_GetSignupPageInfo_excludes_invisible_users(db): 

508 """Test GetSignupPageInfo excludes deleted/banned users from count""" 

509 _get_signup_page_info.cache_clear() 

510 

511 visible_user, _ = generate_user(username="visible") 

512 deleted_user, _ = generate_user(username="deleted", delete_user=True) 

513 

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 

518 

519 

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 

526 

527 

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) 

531 

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 

536 

537 

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) 

541 

542 with session_scope() as session: 

543 session.execute(select(User).where(User.id == shadowed_user.id)).scalar_one().shadowed_at = now() 

544 

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 

549 

550 

551def test_GetPublicUser_limited_visibility(db): 

552 """Test GetPublicUser returns limited_user for user with limited visibility""" 

553 

554 user, _ = generate_user( 

555 username="limited_user", 

556 name="Limited User", 

557 public_visibility=ProfilePublicVisibility.limited, 

558 ) 

559 

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 

582 

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 

593 

594 

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 ) 

602 

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 

610 

611 

612def test_GetPublicUser_full_visibility(db): 

613 """Test GetPublicUser returns full_user for user with full visibility""" 

614 _get_public_users.cache_clear() 

615 

616 user, _ = generate_user( 

617 username="full_user", 

618 name="Full User", 

619 public_visibility=ProfilePublicVisibility.full, 

620 ) 

621 

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 

630 

631 

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() 

640 

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())) 

655 

656 with public_session() as public: 

657 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username)) 

658 assert res.limited_user.num_references == 3 

659 

660 

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) 

666 

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 

670 

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 

674 

675 

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() 

680 

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())) 

684 

685 with public_session() as public: 

686 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user=user.username)) 

687 assert res.limited_user.num_references == 0 

688 

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