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

1import json 

2from datetime import UTC, 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 ProfilePublicVisibility, 

18 Reference, 

19 ReferenceType, 

20) 

21from couchers.proto import api_pb2, public_pb2 

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

23from tests.fixtures.db import generate_user, make_volunteer 

24from tests.fixtures.misc import process_jobs 

25from tests.fixtures.sessions import public_session 

26 

27 

28@pytest.fixture(autouse=True) 

29def _(testconfig): 

30 pass 

31 

32 

33def test_GetPublicMapLayer(db): 

34 user1, _ = generate_user() 

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

36 user3, _ = generate_user() 

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

38 user5, _ = generate_user() 

39 

40 # these are hardcoded in test_fixtures 

41 test_user_coordinates = [-73.9740, 40.7108] 

42 

43 with session_scope() as session: 

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

45 

46 process_jobs() 

47 

48 with public_session() as public: 

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

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

51 data = json.loads(http_body.data) 

52 # Sort to ensure a deterministic order 

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

54 assert data == { 

55 "type": "FeatureCollection", 

56 "features": [ 

57 { 

58 "type": "Feature", 

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

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

61 }, 

62 { 

63 "type": "Feature", 

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

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

66 }, 

67 { 

68 "type": "Feature", 

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

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

71 }, 

72 { 

73 "type": "Feature", 

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

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

76 }, 

77 ], 

78 } 

79 

80 for user in data["features"]: 

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

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

83 assert coords == test_user_coordinates 

84 else: 

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

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

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

88 assert dist > 0.02 and dist < 0.1 

89 

90 

91def test_GetDonationStats_empty(db): 

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

93 _get_donation_stats.cache_clear() 

94 

95 with ( 

96 patch("couchers.servicers.public.DONATION_GOAL_USD", 2500), 

97 patch("couchers.servicers.public.DONATION_OFFSET_USD", 700), 

98 ): 

99 with public_session() as public: 

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

101 assert res.total_donated_ytd == 0 

102 assert res.goal == 2500 

103 

104 

105def test_GetDonationStats_with_donations(db): 

106 """Test GetDonationStats sums on_platform donations correctly""" 

107 _get_donation_stats.cache_clear() 

108 user, _ = generate_user() 

109 

110 with session_scope() as session: 

111 # Add some on_platform donations (should be counted) 

112 session.add( 

113 Invoice( 

114 user_id=user.id, 

115 amount=100, 

116 stripe_payment_intent_id="pi_test_1", 

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

118 invoice_type=InvoiceType.on_platform, 

119 ) 

120 ) 

121 session.add( 

122 Invoice( 

123 user_id=user.id, 

124 amount=250, 

125 stripe_payment_intent_id="pi_test_2", 

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

127 invoice_type=InvoiceType.on_platform, 

128 ) 

129 ) 

130 session.add( 

131 Invoice( 

132 user_id=user.id, 

133 amount=500, 

134 stripe_payment_intent_id="pi_test_3", 

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

136 invoice_type=InvoiceType.on_platform, 

137 ) 

138 ) 

139 

140 with ( 

141 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

142 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

143 ): 

144 with public_session() as public: 

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

146 assert res.total_donated_ytd == 850 

147 assert res.goal == 5000 

148 

149 

150def test_GetDonationStats_excludes_merch(db): 

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

152 _get_donation_stats.cache_clear() 

153 user, _ = generate_user() 

154 

155 with session_scope() as session: 

156 # Add on_platform donation (should be counted) 

157 session.add( 

158 Invoice( 

159 user_id=user.id, 

160 amount=200, 

161 stripe_payment_intent_id="pi_test_donation", 

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

163 invoice_type=InvoiceType.on_platform, 

164 ) 

165 ) 

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

167 session.add( 

168 Invoice( 

169 user_id=user.id, 

170 amount=50, 

171 stripe_payment_intent_id="pi_test_merch", 

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

173 invoice_type=InvoiceType.external_shop, 

174 ) 

175 ) 

176 

177 with ( 

178 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

179 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

180 ): 

181 with public_session() as public: 

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

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

184 assert res.total_donated_ytd == 200 

185 assert res.goal == 5000 

186 

187 

188def test_GetDonationStats_excludes_previous_years(db): 

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

190 _get_donation_stats.cache_clear() 

191 user, _ = generate_user() 

192 

193 with session_scope() as session: 

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

195 session.add( 

196 Invoice( 

197 user_id=user.id, 

198 amount=300, 

199 stripe_payment_intent_id="pi_test_this_year", 

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

201 invoice_type=InvoiceType.on_platform, 

202 ) 

203 ) 

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

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

206 invoice = Invoice( 

207 user_id=user.id, 

208 amount=1000, 

209 stripe_payment_intent_id="pi_test_last_year", 

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

211 invoice_type=InvoiceType.on_platform, 

212 ) 

213 session.add(invoice) 

214 session.flush() 

215 # Manually set the created date to last year 

216 invoice.created = last_year 

217 

218 with ( 

219 patch("couchers.servicers.public.DONATION_GOAL_USD", 5000), 

220 patch("couchers.servicers.public.DONATION_OFFSET_USD", 0), 

221 ): 

222 with public_session() as public: 

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

224 # Should only count this year's donation 

225 assert res.total_donated_ytd == 300 

226 assert res.goal == 5000 

227 

228 

229def test_GetVolunteers_mixed_current_and_past(db): 

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

231 

232 _get_volunteers.cache_clear() 

233 

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

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

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

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

238 

239 with session_scope() as session: 

240 session.add( 

241 make_volunteer( 

242 user_id=current1.id, 

243 role="Current Role 1", 

244 started_volunteering=datetime(2023, 1, 1).date(), 

245 ) 

246 ) 

247 session.add( 

248 make_volunteer( 

249 user_id=current2.id, 

250 role="Current Role 2", 

251 started_volunteering=datetime(2024, 1, 1).date(), 

252 ) 

253 ) 

254 session.add( 

255 make_volunteer( 

256 user_id=past1.id, 

257 role="Past Role 1", 

258 started_volunteering=datetime(2020, 1, 1).date(), 

259 stopped_volunteering=datetime(2022, 6, 1).date(), 

260 ) 

261 ) 

262 session.add( 

263 make_volunteer( 

264 user_id=past2.id, 

265 role="Past Role 2", 

266 started_volunteering=datetime(2021, 1, 1).date(), 

267 stopped_volunteering=datetime(2023, 12, 31).date(), 

268 ) 

269 ) 

270 

271 refresh_materialized_views_rapid(empty_pb2.Empty()) 

272 

273 with public_session() as public: 

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

275 assert len(res.current_volunteers) == 2 

276 assert len(res.past_volunteers) == 2 

277 

278 # Past volunteers are sorted by stopped_volunteering descending 

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

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

281 

282 

283def test_GetVolunteers_custom_sort_key(db): 

284 """Test GetVolunteers respects custom sort_key""" 

285 

286 _get_volunteers.cache_clear() 

287 

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

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

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

291 

292 with session_scope() as session: 

293 # user2 should be first (lowest sort_key) 

294 session.add( 

295 make_volunteer( 

296 user_id=user2.id, 

297 role="Role 2", 

298 started_volunteering=datetime(2023, 3, 1).date(), 

299 sort_key=1.0, 

300 ) 

301 ) 

302 # user3 should be second 

303 session.add( 

304 make_volunteer( 

305 user_id=user3.id, 

306 role="Role 3", 

307 started_volunteering=datetime(2023, 1, 1).date(), 

308 sort_key=2.0, 

309 ) 

310 ) 

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

312 session.add( 

313 make_volunteer( 

314 user_id=user1.id, 

315 role="Role 1", 

316 started_volunteering=datetime(2023, 2, 1).date(), 

317 ) 

318 ) 

319 

320 refresh_materialized_views_rapid(empty_pb2.Empty()) 

321 

322 with public_session() as public: 

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

324 assert len(res.current_volunteers) == 3 

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

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

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

328 

329 

330def test_GetVolunteers_excludes_hidden(db): 

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

332 

333 _get_volunteers.cache_clear() 

334 

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

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

337 

338 with session_scope() as session: 

339 session.add( 

340 make_volunteer( 

341 user_id=user1.id, 

342 role="Visible Role", 

343 started_volunteering=datetime(2023, 1, 1).date(), 

344 ) 

345 ) 

346 session.add( 

347 make_volunteer( 

348 user_id=user2.id, 

349 role="Hidden Role", 

350 started_volunteering=datetime(2023, 1, 1).date(), 

351 show_on_team_page=False, 

352 ) 

353 ) 

354 

355 refresh_materialized_views_rapid(empty_pb2.Empty()) 

356 

357 with public_session() as public: 

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

359 assert len(res.current_volunteers) == 1 

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

361 

362 

363def test_GetVolunteers_link_types(db): 

364 """Test GetVolunteers handles different link types""" 

365 

366 _get_volunteers.cache_clear() 

367 

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

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

370 

371 with session_scope() as session: 

372 # Volunteer with default couchers link 

373 session.add( 

374 make_volunteer( 

375 user_id=user_default.id, 

376 role="Default Link", 

377 started_volunteering=datetime(2023, 1, 1).date(), 

378 ) 

379 ) 

380 # Volunteer with custom link 

381 session.add( 

382 make_volunteer( 

383 user_id=user_custom.id, 

384 role="Custom Link", 

385 started_volunteering=datetime(2023, 1, 1).date(), 

386 link_type="email", 

387 link_text="contact@example.com", 

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

389 ) 

390 ) 

391 

392 refresh_materialized_views_rapid(empty_pb2.Empty()) 

393 

394 with public_session() as public: 

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

396 assert len(res.current_volunteers) == 2 

397 

398 # Check default link 

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

400 assert default_vol.link_type == "couchers" 

401 assert default_vol.link_text == "@default_link" 

402 assert "default_link" in default_vol.link_url 

403 

404 # Check custom link 

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

406 assert custom_vol.link_type == "email" 

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

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

409 

410 

411def test_GetVolunteers_board_member_flag(db): 

412 """Test GetVolunteers correctly identifies board members""" 

413 

414 _get_volunteers.cache_clear() 

415 

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

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

418 

419 with session_scope() as session: 

420 session.add( 

421 make_volunteer( 

422 user_id=board_member.id, 

423 role="Board Member Role", 

424 started_volunteering=datetime(2023, 1, 1).date(), 

425 ) 

426 ) 

427 session.add( 

428 make_volunteer( 

429 user_id=regular_volunteer.id, 

430 role="Regular Role", 

431 started_volunteering=datetime(2023, 1, 1).date(), 

432 ) 

433 ) 

434 

435 refresh_materialized_views_rapid(empty_pb2.Empty()) 

436 

437 # Mock the static badge dict to include board_member 

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

439 with public_session() as public: 

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

441 assert len(res.current_volunteers) == 2 

442 

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

444 assert board_vol.is_board_member is True 

445 

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

447 assert regular_vol.is_board_member is False 

448 

449 

450def test_GetSignupPageInfo(db): 

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

452 

453 _get_signup_page_info.cache_clear() 

454 

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

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

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

458 

459 refresh_materialized_views_rapid(empty_pb2.Empty()) 

460 

461 with public_session() as public: 

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

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

464 assert res.user_count >= 3 

465 assert res.last_location # Should have some location 

466 assert res.last_signup # Should have a timestamp 

467 

468 

469def test_GetSignupPageInfo_excludes_invisible_users(db): 

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

471 _get_signup_page_info.cache_clear() 

472 

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

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

475 

476 with public_session() as public: 

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

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

479 assert res.user_count >= 1 

480 

481 

482def test_GetPublicUser_not_found(db): 

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

484 with public_session() as public: 

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

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

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

488 

489 

490def test_GetPublicUser_invisible_user(db): 

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

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

493 

494 with public_session() as public: 

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

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

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

498 

499 

500def test_GetPublicUser_limited_visibility(db): 

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

502 

503 user, _ = generate_user( 

504 username="limited_user", 

505 name="Limited User", 

506 public_visibility=ProfilePublicVisibility.limited, 

507 ) 

508 

509 # Add a reference to test reference counting 

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

511 with session_scope() as session: 

512 session.add( 

513 Reference( 

514 from_user_id=referrer.id, 

515 to_user_id=user.id, 

516 reference_type=ReferenceType.friend, 

517 text="Great host!", 

518 rating=0.8, 

519 was_appropriate=True, 

520 ) 

521 ) 

522 

523 with public_session() as public: 

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

525 assert res.HasField("limited_user") 

526 assert res.limited_user.username == "limited_user" 

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

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

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

530 assert res.limited_user.num_references == 1 

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

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

533 

534 

535def test_GetPublicUser_most_visibility(db): 

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

537 user, _ = generate_user( 

538 username="most_user", 

539 name="Most User", 

540 public_visibility=ProfilePublicVisibility.most, 

541 ) 

542 

543 with public_session() as public: 

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

545 assert res.HasField("most_user") 

546 assert res.most_user.username == "most_user" 

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

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

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

550 

551 

552def test_GetPublicUser_full_visibility(db): 

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

554 _get_public_users.cache_clear() 

555 

556 user, _ = generate_user( 

557 username="full_user", 

558 name="Full User", 

559 public_visibility=ProfilePublicVisibility.full, 

560 ) 

561 

562 with public_session() as public: 

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

564 assert res.HasField("full_user") 

565 assert res.full_user.username == "full_user" 

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

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

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

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