Coverage for app/backend/src/tests/test_public.py: 100%

1import json 

2from datetime import UTC, date, datetime 

3from math import sqrt 

4from unittest.mock import patch 

5 

6import grpc 

7import pytest 

8from google.protobuf import empty_pb2 

9 

10from couchers.db import session_scope 

11from couchers.jobs.enqueue import queue_job 

12from couchers.jobs.handlers import update_randomized_locations 

13from couchers.materialized_views import refresh_materialized_views_rapid 

14from couchers.models import ( 

15 Invoice, 

16 InvoiceType, 

17 ModerationObjectType, 

18 ModerationState, 

19 ModerationVisibility, 

20 ProfilePublicVisibility, 

21 Reference, 

22 ReferenceType, 

23) 

24from couchers.proto import api_pb2, public_pb2 

25from couchers.servicers.public import _get_donation_stats, _get_public_users, _get_signup_page_info, _get_volunteers 

26from tests.fixtures.db import generate_user, make_volunteer 

27from tests.fixtures.misc import process_jobs 

28from tests.fixtures.sessions import public_session 

29 

30 

31@pytest.fixture(autouse=True) 

32def _(testconfig): 

33 pass 

34 

35 

36def test_GetPublicMapLayer(db): 

37 user1, _ = generate_user() 

38 user2, _ = generate_user(username="user2", public_visibility=ProfilePublicVisibility.nothing) 

39 user3, _ = generate_user() 

40 user4, _ = generate_user(username="user4", public_visibility=ProfilePublicVisibility.limited) 

41 user5, _ = generate_user() 

42 

43 # these are hardcoded in test_fixtures 

44 test_user_coordinates = [-73.9740, 40.7108] 

45 

46 with session_scope() as session: 

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

48 

49 process_jobs() 

50 

51 with public_session() as public: 

52 http_body = public.GetPublicUsers(empty_pb2.Empty()) 

53 assert http_body.content_type == "application/json" 

54 data = json.loads(http_body.data) 

55 # Sort to ensure a deterministic order 

56 data["features"].sort(key=lambda f: f["geometry"]["coordinates"][0]) 

57 assert data == { 

58 "type": "FeatureCollection", 

59 "features": [ 

60 { 

61 "type": "Feature", 

62 "geometry": {"type": "Point", "coordinates": [-74.042643848, 40.706241098]}, 

63 "properties": {"username": None}, 

64 }, 

65 { 

66 "type": "Feature", 

67 "geometry": {"type": "Point", "coordinates": [-73.974, 40.7108]}, 

68 "properties": {"username": "user4"}, 

69 }, 

70 { 

71 "type": "Feature", 

72 "geometry": {"type": "Point", "coordinates": [-73.955417734, 40.691831306]}, 

73 "properties": {"username": None}, 

74 }, 

75 { 

76 "type": "Feature", 

77 "geometry": {"type": "Point", "coordinates": [-73.928380198, 40.729706144]}, 

78 "properties": {"username": None}, 

79 }, 

80 ], 

81 } 

82 

83 for user in data["features"]: 

84 coords = user["geometry"]["coordinates"] 

85 if user["properties"]["username"]: 

86 assert coords == test_user_coordinates 

87 else: 

88 xdiff = coords[0] - test_user_coordinates[0] 

89 ydiff = coords[1] - test_user_coordinates[1] 

90 dist = sqrt(xdiff**2 + ydiff**2) 

91 assert dist > 0.02 and dist < 0.1 

92 

93 

94def test_GetDonationStats_empty(db, feature_flags): 

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

96 _get_donation_stats.cache_clear() 

97 

98 feature_flags.set("donation_goal_usd", 2500) 

99 feature_flags.set("donation_offset_usd", 700) 

100 with public_session() as public: 

101 res = public.GetDonationStats(empty_pb2.Empty()) 

102 assert res.total_donated_ytd == 0 

103 assert res.goal == 2500 

104 

105 

106def test_GetDonationStats_with_donations(db, feature_flags): 

107 """Test GetDonationStats sums on_platform donations correctly""" 

108 _get_donation_stats.cache_clear() 

109 user, _ = generate_user() 

110 

111 with session_scope() as session: 

112 # Add some on_platform donations (should be counted) 

113 session.add( 

114 Invoice( 

115 user_id=user.id, 

116 amount=100, 

117 stripe_payment_intent_id="pi_test_1", 

118 stripe_receipt_url="https://example.com/receipt/1", 

119 invoice_type=InvoiceType.on_platform, 

120 ) 

121 ) 

122 session.add( 

123 Invoice( 

124 user_id=user.id, 

125 amount=250, 

126 stripe_payment_intent_id="pi_test_2", 

127 stripe_receipt_url="https://example.com/receipt/2", 

128 invoice_type=InvoiceType.on_platform, 

129 ) 

130 ) 

131 session.add( 

132 Invoice( 

133 user_id=user.id, 

134 amount=500, 

135 stripe_payment_intent_id="pi_test_3", 

136 stripe_receipt_url="https://example.com/receipt/3", 

137 invoice_type=InvoiceType.on_platform, 

138 ) 

139 ) 

140 

141 feature_flags.set("donation_goal_usd", 5000) 

142 feature_flags.set("donation_offset_usd", 0) 

143 with public_session() as public: 

144 res = public.GetDonationStats(empty_pb2.Empty()) 

145 assert res.total_donated_ytd == 850 

146 assert res.goal == 5000 

147 

148 

149def test_GetDonationStats_excludes_merch(db, feature_flags): 

150 """Test GetDonationStats excludes external_shop (merch) invoices""" 

151 _get_donation_stats.cache_clear() 

152 user, _ = generate_user() 

153 

154 with session_scope() as session: 

155 # Add on_platform donation (should be counted) 

156 session.add( 

157 Invoice( 

158 user_id=user.id, 

159 amount=200, 

160 stripe_payment_intent_id="pi_test_donation", 

161 stripe_receipt_url="https://example.com/receipt/donation", 

162 invoice_type=InvoiceType.on_platform, 

163 ) 

164 ) 

165 # Add external_shop/merch purchase (should NOT be counted) 

166 session.add( 

167 Invoice( 

168 user_id=user.id, 

169 amount=50, 

170 stripe_payment_intent_id="pi_test_merch", 

171 stripe_receipt_url="https://example.com/receipt/merch", 

172 invoice_type=InvoiceType.external_shop, 

173 ) 

174 ) 

175 

176 feature_flags.set("donation_goal_usd", 5000) 

177 feature_flags.set("donation_offset_usd", 0) 

178 with public_session() as public: 

179 res = public.GetDonationStats(empty_pb2.Empty()) 

180 # Should only count the on_platform donation, not the merch 

181 assert res.total_donated_ytd == 200 

182 assert res.goal == 5000 

183 

184 

185def test_GetDonationStats_excludes_previous_years(db, feature_flags): 

186 """Test GetDonationStats only counts current year donations""" 

187 _get_donation_stats.cache_clear() 

188 user, _ = generate_user() 

189 

190 with session_scope() as session: 

191 # Add donation from this year (should be counted) 

192 session.add( 

193 Invoice( 

194 user_id=user.id, 

195 amount=300, 

196 stripe_payment_intent_id="pi_test_this_year", 

197 stripe_receipt_url="https://example.com/receipt/this_year", 

198 invoice_type=InvoiceType.on_platform, 

199 ) 

200 ) 

201 # Add donation from last year (should NOT be counted) 

202 last_year = datetime(datetime.now(UTC).year - 1, 6, 15, tzinfo=UTC) 

203 invoice = Invoice( 

204 user_id=user.id, 

205 amount=1000, 

206 stripe_payment_intent_id="pi_test_last_year", 

207 stripe_receipt_url="https://example.com/receipt/last_year", 

208 invoice_type=InvoiceType.on_platform, 

209 ) 

210 session.add(invoice) 

211 session.flush() 

212 # Manually set the created date to last year 

213 invoice.created = last_year 

214 

215 feature_flags.set("donation_goal_usd", 5000) 

216 feature_flags.set("donation_offset_usd", 0) 

217 with public_session() as public: 

218 res = public.GetDonationStats(empty_pb2.Empty()) 

219 # Should only count this year's donation 

220 assert res.total_donated_ytd == 300 

221 assert res.goal == 5000 

222 

223 

224def test_GetDonationStats_uses_flags(db, feature_flags): 

225 """Goal and offset come from the donation_goal_usd / donation_offset_usd flags when configured""" 

226 _get_donation_stats.cache_clear() 

227 user, _ = generate_user() 

228 

229 with session_scope() as session: 

230 session.add( 

231 Invoice( 

232 user_id=user.id, 

233 amount=1000, 

234 stripe_payment_intent_id="pi_test_flag", 

235 stripe_receipt_url="https://example.com/receipt/flag", 

236 invoice_type=InvoiceType.on_platform, 

237 ) 

238 ) 

239 

240 feature_flags.set("donation_goal_usd", 12000) 

241 feature_flags.set("donation_offset_usd", 300) 

242 

243 with public_session() as public: 

244 res = public.GetDonationStats(empty_pb2.Empty()) 

245 assert res.goal == 12000 

246 assert res.total_donated_ytd == 700 # 1000 donated minus the 300 offset 

247 

248 _get_donation_stats.cache_clear() 

249 

250 

251def test_GetVolunteers_mixed_current_and_past(db): 

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

253 

254 _get_volunteers.cache_clear() 

255 

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

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

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

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

260 

261 with session_scope() as session: 

262 session.add( 

263 make_volunteer( 

264 user_id=current1.id, 

265 role="Current Role 1", 

266 started_volunteering=date(2023, 1, 1), 

267 ) 

268 ) 

269 session.add( 

270 make_volunteer( 

271 user_id=current2.id, 

272 role="Current Role 2", 

273 started_volunteering=date(2024, 1, 1), 

274 ) 

275 ) 

276 session.add( 

277 make_volunteer( 

278 user_id=past1.id, 

279 role="Past Role 1", 

280 started_volunteering=date(2020, 1, 1), 

281 stopped_volunteering=date(2022, 6, 1), 

282 ) 

283 ) 

284 session.add( 

285 make_volunteer( 

286 user_id=past2.id, 

287 role="Past Role 2", 

288 started_volunteering=date(2021, 1, 1), 

289 stopped_volunteering=date(2023, 12, 31), 

290 ) 

291 ) 

292 

293 refresh_materialized_views_rapid(empty_pb2.Empty()) 

294 

295 with public_session() as public: 

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

297 assert len(res.current_volunteers) == 2 

298 assert len(res.past_volunteers) == 2 

299 

300 # Past volunteers are sorted by stopped_volunteering descending 

301 assert res.past_volunteers[0].username == "past2" 

302 assert res.past_volunteers[1].username == "past1" 

303 

304 

305def test_GetVolunteers_custom_sort_key(db): 

306 """Test GetVolunteers respects custom sort_key""" 

307 

308 _get_volunteers.cache_clear() 

309 

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

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

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

313 

314 with session_scope() as session: 

315 # user2 should be first (lowest sort_key) 

316 session.add( 

317 make_volunteer( 

318 user_id=user2.id, 

319 role="Role 2", 

320 started_volunteering=date(2023, 3, 1), 

321 sort_key=1.0, 

322 ) 

323 ) 

324 # user3 should be second 

325 session.add( 

326 make_volunteer( 

327 user_id=user3.id, 

328 role="Role 3", 

329 started_volunteering=date(2023, 1, 1), 

330 sort_key=2.0, 

331 ) 

332 ) 

333 # user1 should be last (no sort_key, falls back to started_volunteering) 

334 session.add( 

335 make_volunteer( 

336 user_id=user1.id, 

337 role="Role 1", 

338 started_volunteering=date(2023, 2, 1), 

339 ) 

340 ) 

341 

342 refresh_materialized_views_rapid(empty_pb2.Empty()) 

343 

344 with public_session() as public: 

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

346 assert len(res.current_volunteers) == 3 

347 assert res.current_volunteers[0].username == "user2" 

348 assert res.current_volunteers[1].username == "user3" 

349 assert res.current_volunteers[2].username == "user1" 

350 

351 

352def test_GetVolunteers_excludes_hidden(db): 

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

354 

355 _get_volunteers.cache_clear() 

356 

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

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

359 

360 with session_scope() as session: 

361 session.add( 

362 make_volunteer( 

363 user_id=user1.id, 

364 role="Visible Role", 

365 started_volunteering=date(2023, 1, 1), 

366 ) 

367 ) 

368 session.add( 

369 make_volunteer( 

370 user_id=user2.id, 

371 role="Hidden Role", 

372 started_volunteering=date(2023, 1, 1), 

373 show_on_team_page=False, 

374 ) 

375 ) 

376 

377 refresh_materialized_views_rapid(empty_pb2.Empty()) 

378 

379 with public_session() as public: 

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

381 assert len(res.current_volunteers) == 1 

382 assert res.current_volunteers[0].username == "visible" 

383 

384 

385def test_GetVolunteers_link_types(db): 

386 """Test GetVolunteers handles different link types""" 

387 

388 _get_volunteers.cache_clear() 

389 

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

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

392 

393 with session_scope() as session: 

394 # Volunteer with default couchers link 

395 session.add( 

396 make_volunteer( 

397 user_id=user_default.id, 

398 role="Default Link", 

399 started_volunteering=date(2023, 1, 1), 

400 ) 

401 ) 

402 # Volunteer with custom link 

403 session.add( 

404 make_volunteer( 

405 user_id=user_custom.id, 

406 role="Custom Link", 

407 started_volunteering=date(2023, 1, 1), 

408 link_type="email", 

409 link_text="contact@example.com", 

410 link_url="mailto:contact@example.com", 

411 ) 

412 ) 

413 

414 refresh_materialized_views_rapid(empty_pb2.Empty()) 

415 

416 with public_session() as public: 

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

418 assert len(res.current_volunteers) == 2 

419 

420 # Check default link 

421 default_vol = next(v for v in res.current_volunteers if v.username == "default_link") 

422 assert default_vol.link_type == "couchers" 

423 assert default_vol.link_text == "@default_link" 

424 assert "default_link" in default_vol.link_url 

425 

426 # Check custom link 

427 custom_vol = next(v for v in res.current_volunteers if v.username == "custom_link") 

428 assert custom_vol.link_type == "email" 

429 assert custom_vol.link_text == "contact@example.com" 

430 assert custom_vol.link_url == "mailto:contact@example.com" 

431 

432 

433def test_GetVolunteers_board_member_flag(db): 

434 """Test GetVolunteers correctly identifies board members""" 

435 

436 _get_volunteers.cache_clear() 

437 

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

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

440 

441 with session_scope() as session: 

442 session.add( 

443 make_volunteer( 

444 user_id=board_member.id, 

445 role="Board Member Role", 

446 started_volunteering=date(2023, 1, 1), 

447 ) 

448 ) 

449 session.add( 

450 make_volunteer( 

451 user_id=regular_volunteer.id, 

452 role="Regular Role", 

453 started_volunteering=date(2023, 1, 1), 

454 ) 

455 ) 

456 

457 refresh_materialized_views_rapid(empty_pb2.Empty()) 

458 

459 # Mock the static badge dict to include board_member 

460 with patch("couchers.servicers.public.get_static_badge_dict", return_value={"board_member": [board_member.id]}): 

461 with public_session() as public: 

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

463 assert len(res.current_volunteers) == 2 

464 

465 board_vol = next(v for v in res.current_volunteers if v.username == "board_member") 

466 assert board_vol.is_board_member is True 

467 

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

469 assert regular_vol.is_board_member is False 

470 

471 

472def test_GetSignupPageInfo(db): 

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

474 

475 _get_signup_page_info.cache_clear() 

476 

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

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

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

480 

481 refresh_materialized_views_rapid(empty_pb2.Empty()) 

482 

483 with public_session() as public: 

484 res = public.GetSignupPageInfo(empty_pb2.Empty()) 

485 # user3 should be the last signup (highest id) 

486 assert res.user_count >= 3 

487 assert res.last_location # Should have some location 

488 assert res.last_signup # Should have a timestamp 

489 

490 

491def test_GetSignupPageInfo_excludes_invisible_users(db): 

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

493 _get_signup_page_info.cache_clear() 

494 

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

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

497 

498 with public_session() as public: 

499 res = public.GetSignupPageInfo(empty_pb2.Empty()) 

500 # Deleted user should not be counted or be the last signup 

501 assert res.user_count >= 1 

502 

503 

504def test_GetPublicUser_not_found(db): 

505 """Test GetPublicUser returns NOT_FOUND for nonexistent user""" 

506 with public_session() as public: 

507 with pytest.raises(grpc.RpcError) as exc: 

508 public.GetPublicUser(public_pb2.GetPublicUserReq(user="nonexistent_user")) 

509 assert exc.value.code() == grpc.StatusCode.NOT_FOUND 

510 

511 

512def test_GetPublicUser_invisible_user(db): 

513 """Test GetPublicUser returns NOT_FOUND for deleted/banned user""" 

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

515 

516 with public_session() as public: 

517 with pytest.raises(grpc.RpcError) as exc: 

518 public.GetPublicUser(public_pb2.GetPublicUserReq(user="deleted")) 

519 assert exc.value.code() == grpc.StatusCode.NOT_FOUND 

520 

521 

522def test_GetPublicUser_limited_visibility(db): 

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

524 

525 user, _ = generate_user( 

526 username="limited_user", 

527 name="Limited User", 

528 public_visibility=ProfilePublicVisibility.limited, 

529 ) 

530 

531 # Add a reference to test reference counting 

532 referrer, _ = generate_user(username="referrer") 

533 with session_scope() as session: 

534 moderation_state = ModerationState( 

535 object_type=ModerationObjectType.reference, 

536 object_id=0, 

537 visibility=ModerationVisibility.visible, 

538 ) 

539 session.add(moderation_state) 

540 session.flush() 

541 reference = Reference( 

542 from_user_id=referrer.id, 

543 to_user_id=user.id, 

544 reference_type=ReferenceType.friend, 

545 text="Great host!", 

546 rating=0.8, 

547 was_appropriate=True, 

548 moderation_state_id=moderation_state.id, 

549 ) 

550 session.add(reference) 

551 session.flush() 

552 moderation_state.object_id = reference.id 

553 

554 with public_session() as public: 

555 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="limited_user")) 

556 assert res.HasField("limited_user") 

557 assert res.limited_user.username == "limited_user" 

558 assert res.limited_user.name == "Limited User" 

559 assert res.limited_user.city == "Testing city" 

560 assert res.limited_user.hometown == "Test hometown" 

561 assert res.limited_user.num_references == 1 

562 assert res.limited_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST 

563 assert len(res.limited_user.badges) == 0 

564 

565 

566def test_GetPublicUser_most_visibility(db): 

567 """Test GetPublicUser returns most_user for user with most visibility""" 

568 user, _ = generate_user( 

569 username="most_user", 

570 name="Most User", 

571 public_visibility=ProfilePublicVisibility.most, 

572 ) 

573 

574 with public_session() as public: 

575 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="most_user")) 

576 assert res.HasField("most_user") 

577 assert res.most_user.username == "most_user" 

578 assert res.most_user.name == "Most User" 

579 assert res.most_user.city == "Testing city" 

580 assert res.most_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST 

581 

582 

583def test_GetPublicUser_full_visibility(db): 

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

585 _get_public_users.cache_clear() 

586 

587 user, _ = generate_user( 

588 username="full_user", 

589 name="Full User", 

590 public_visibility=ProfilePublicVisibility.full, 

591 ) 

592 

593 with public_session() as public: 

594 res = public.GetPublicUser(public_pb2.GetPublicUserReq(user="full_user")) 

595 assert res.HasField("full_user") 

596 assert res.full_user.username == "full_user" 

597 assert res.full_user.name == "Full User" 

598 assert res.full_user.city == "Testing city" 

599 # Full user should have all the fields from the complete user profile 

600 assert res.full_user.hosting_status == api_pb2.HOSTING_STATUS_CANT_HOST