Coverage for app/backend/src/tests/test

1import json

2from datetime import UTC, date, datetime

3from math import sqrt

4from unittest.mock import patch

6import grpc

7import pytest

8from google.protobuf import empty_pb2

9from sqlalchemy import select

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_volunteer

30from tests.fixtures.misc import process_jobs

31from tests.fixtures.sessions import public_session

34@pytest.fixture(autouse=True)

35def _(testconfig):

36 pass

39def test_GetPublicMapLayer(db):

40 user1, _ = generate_user()

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

42 user3, _ = generate_user()

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

44 user5, _ = generate_user()

46 # these are hardcoded in test_fixtures

47 test_user_coordinates = [-73.9740, 40.7108]

49 with session_scope() as session:

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

52 process_jobs()

54 with public_session() as public:

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

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

57 data = json.loads(http_body.data)

58 # Sort to ensure a deterministic order

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

60 assert data == {

61 "type": "FeatureCollection",

62 "features": [

63 {

64 "type": "Feature",

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

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

67 },

68 {

69 "type": "Feature",

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

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

72 },

73 {

74 "type": "Feature",

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

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

77 },

78 {

79 "type": "Feature",

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

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

82 },

83 ],

84 }

86 for user in data["features"]:

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

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

89 assert coords == test_user_coordinates

90 else:

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

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

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

94 assert dist > 0.02 and dist < 0.1

97def test_GetPublicMapLayer_excludes_shadowed(db):

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

100 _get_public_users.cache_clear()

101

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

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

104

105 with session_scope() as session:

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

107

108 with public_session() as public:

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

110

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

112

113

114def test_GetDonationStats_empty(db, feature_flags):

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

116 _get_donation_stats.cache_clear()

117

118 feature_flags.set("donation_goal_usd", 2500)

119 feature_flags.set("donation_offset_usd", 700)

120 with public_session() as public:

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

122 assert res.total_donated_ytd == 0

123 assert res.goal == 2500

124

125

126def test_GetDonationStats_with_donations(db, feature_flags):

127 """Test GetDonationStats sums on_platform donations correctly"""

128 _get_donation_stats.cache_clear()

129 user, _ = generate_user()

130

131 with session_scope() as session:

132 # Add some on_platform donations (should be counted)

133 session.add(

134 Invoice(

135 user_id=user.id,

136 amount=100,

137 stripe_payment_intent_id="pi_test_1",

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

139 invoice_type=InvoiceType.on_platform,

140 )

141 )

142 session.add(

143 Invoice(

144 user_id=user.id,

145 amount=250,

146 stripe_payment_intent_id="pi_test_2",

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

148 invoice_type=InvoiceType.on_platform,

149 )

150 )

151 session.add(

152 Invoice(

153 user_id=user.id,

154 amount=500,

155 stripe_payment_intent_id="pi_test_3",

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

157 invoice_type=InvoiceType.on_platform,

158 )

159 )

160

161 feature_flags.set("donation_goal_usd", 5000)

162 feature_flags.set("donation_offset_usd", 0)

163 with public_session() as public:

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

165 assert res.total_donated_ytd == 850

166 assert res.goal == 5000

167

168

169def test_GetDonationStats_excludes_merch(db, feature_flags):

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

171 _get_donation_stats.cache_clear()

172 user, _ = generate_user()

173

174 with session_scope() as session:

175 # Add on_platform donation (should be counted)

176 session.add(

177 Invoice(

178 user_id=user.id,

179 amount=200,

180 stripe_payment_intent_id="pi_test_donation",

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

182 invoice_type=InvoiceType.on_platform,

183 )

184 )

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

186 session.add(

187 Invoice(

188 user_id=user.id,

189 amount=50,

190 stripe_payment_intent_id="pi_test_merch",

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

192 invoice_type=InvoiceType.external_shop,

193 )

194 )

195

196 feature_flags.set("donation_goal_usd", 5000)

197 feature_flags.set("donation_offset_usd", 0)

198 with public_session() as public:

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

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

201 assert res.total_donated_ytd == 200

202 assert res.goal == 5000

203

204

205def test_GetDonationStats_excludes_previous_years(db, feature_flags):

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

207 _get_donation_stats.cache_clear()

208 user, _ = generate_user()

209

210 with session_scope() as session:

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

212 session.add(

213 Invoice(

214 user_id=user.id,

215 amount=300,

216 stripe_payment_intent_id="pi_test_this_year",

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

218 invoice_type=InvoiceType.on_platform,

219 )

220 )

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

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

223 invoice = Invoice(

224 user_id=user.id,

225 amount=1000,

226 stripe_payment_intent_id="pi_test_last_year",

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

228 invoice_type=InvoiceType.on_platform,

229 )

230 session.add(invoice)

231 session.flush()

232 # Manually set the created date to last year

233 invoice.created = last_year

234

235 feature_flags.set("donation_goal_usd", 5000)

236 feature_flags.set("donation_offset_usd", 0)

237 with public_session() as public:

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

239 # Should only count this year's donation

240 assert res.total_donated_ytd == 300

241 assert res.goal == 5000

242

243

244def test_GetDonationStats_uses_flags(db, feature_flags):

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

246 _get_donation_stats.cache_clear()

247 user, _ = generate_user()

248

249 with session_scope() as session:

250 session.add(

251 Invoice(

252 user_id=user.id,

253 amount=1000,

254 stripe_payment_intent_id="pi_test_flag",

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

256 invoice_type=InvoiceType.on_platform,

257 )

258 )

259

260 feature_flags.set("donation_goal_usd", 12000)

261 feature_flags.set("donation_offset_usd", 300)

262

263 with public_session() as public:

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

265 assert res.goal == 12000

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

267

268 _get_donation_stats.cache_clear()

269

270

271def test_GetVolunteers_mixed_current_and_past(db):

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

273

274 _get_volunteers.cache_clear()

275

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

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

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

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

280

281 with session_scope() as session:

282 session.add(

283 make_volunteer(

284 user_id=current1.id,

285 role="Current Role 1",

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

287 )

288 )

289 session.add(

290 make_volunteer(

291 user_id=current2.id,

292 role="Current Role 2",

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

294 )

295 )

296 session.add(

297 make_volunteer(

298 user_id=past1.id,

299 role="Past Role 1",

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

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

302 )

303 )

304 session.add(

305 make_volunteer(

306 user_id=past2.id,

307 role="Past Role 2",

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

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

310 )

311 )

312

313 refresh_materialized_views_rapid(empty_pb2.Empty())

314

315 with public_session() as public:

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

317 assert len(res.current_volunteers) == 2

318 assert len(res.past_volunteers) == 2

319

320 # Past volunteers are sorted by stopped_volunteering descending

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

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

323

324

325def test_GetVolunteers_custom_sort_key(db):

326 """Test GetVolunteers respects custom sort_key"""

327

328 _get_volunteers.cache_clear()

329

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

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

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

333

334 with session_scope() as session:

335 # user2 should be first (lowest sort_key)

336 session.add(

337 make_volunteer(

338 user_id=user2.id,

339 role="Role 2",

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

341 sort_key=1.0,

342 )

343 )

344 # user3 should be second

345 session.add(

346 make_volunteer(

347 user_id=user3.id,

348 role="Role 3",

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

350 sort_key=2.0,

351 )

352 )

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

354 session.add(

355 make_volunteer(

356 user_id=user1.id,

357 role="Role 1",

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

359 )

360 )

361

362 refresh_materialized_views_rapid(empty_pb2.Empty())

363

364 with public_session() as public:

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

366 assert len(res.current_volunteers) == 3

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

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

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

370

371

372def test_GetVolunteers_excludes_hidden(db):

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

374

375 _get_volunteers.cache_clear()

376

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

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

379

380 with session_scope() as session:

381 session.add(

382 make_volunteer(

383 user_id=user1.id,

384 role="Visible Role",

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

386 )

387 )

388 session.add(

389 make_volunteer(

390 user_id=user2.id,

391 role="Hidden Role",

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

393 show_on_team_page=False,

394 )

395 )

396

397 refresh_materialized_views_rapid(empty_pb2.Empty())

398

399 with public_session() as public:

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

401 assert len(res.current_volunteers) == 1

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

403

404

405def test_GetVolunteers_link_types(db):

406 """Test GetVolunteers handles different link types"""

407

408 _get_volunteers.cache_clear()

409

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

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

412

413 with session_scope() as session:

414 # Volunteer with default couchers link

415 session.add(

416 make_volunteer(

417 user_id=user_default.id,

418 role="Default Link",

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

420 )

421 )

422 # Volunteer with custom link

423 session.add(

424 make_volunteer(

425 user_id=user_custom.id,

426 role="Custom Link",

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

428 link_type="email",

429 link_text="contact@example.com",

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

431 )

432 )

433

434 refresh_materialized_views_rapid(empty_pb2.Empty())

435

436 with public_session() as public:

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

438 assert len(res.current_volunteers) == 2

439

440 # Check default link

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

442 assert default_vol.link_type == "couchers"

443 assert default_vol.link_text == "@default_link"

444 assert "default_link" in default_vol.link_url

445

446 # Check custom link

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

448 assert custom_vol.link_type == "email"

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

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

451

452

453def test_GetVolunteers_board_member_flag(db):

454 """Test GetVolunteers correctly identifies board members"""

455

456 _get_volunteers.cache_clear()

457

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

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

460

461 with session_scope() as session:

462 session.add(

463 make_volunteer(

464 user_id=board_member.id,

465 role="Board Member Role",

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

467 )

468 )

469 session.add(

470 make_volunteer(

471 user_id=regular_volunteer.id,

472 role="Regular Role",

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

474 )

475 )

476

477 refresh_materialized_views_rapid(empty_pb2.Empty())

478

479 # Mock the static badge dict to include board_member

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

481 with public_session() as public:

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

483 assert len(res.current_volunteers) == 2

484

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

486 assert board_vol.is_board_member is True

487

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

489 assert regular_vol.is_board_member is False

490

491

492def test_GetSignupPageInfo(db):

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

494

495 _get_signup_page_info.cache_clear()

496

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

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

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

500

501 refresh_materialized_views_rapid(empty_pb2.Empty())

502

503 with public_session() as public:

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

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

506 assert res.user_count >= 3

507 assert res.last_location # Should have some location

508 assert res.last_signup # Should have a timestamp

509

510

511def test_GetSignupPageInfo_excludes_invisible_users(db):

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

513 _get_signup_page_info.cache_clear()

514

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

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

517

518 with public_session() as public:

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

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

521 assert res.user_count >= 1

522

523

524def test_GetPublicUser_not_found(db):

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

526 with public_session() as public:

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

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

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

530

531

532def test_GetPublicUser_invisible_user(db):

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

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

535

536 with public_session() as public:

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

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

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

540

541

542def test_GetPublicUser_shadowed_user(db):

543 """Test GetPublicUser returns NOT_FOUND for a shadowed user"""

544 shadowed_user, _ = generate_user(username="shadowed", public_visibility=ProfilePublicVisibility.full)

545

546 with session_scope() as session:

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

548

549 with public_session() as public:

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

551 public.GetPublicUser(public_pb2.GetPublicUserReq(user="shadowed"))

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

553

554

555def test_GetPublicUser_limited_visibility(db):

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

557

558 user, _ = generate_user(

559 username="limited_user",

560 name="Limited User",

561 public_visibility=ProfilePublicVisibility.limited,

562 )

563

564 # Add a reference to test reference counting

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

566 with session_scope() as session:

567 moderation_state = ModerationState(

568 object_type=ModerationObjectType.reference,

569 object_id=0,

570 visibility=ModerationVisibility.visible,

571 )

572 session.add(moderation_state)

573 session.flush()

574 reference = Reference(

575 from_user_id=referrer.id,

576 to_user_id=user.id,

577 reference_type=ReferenceType.friend,

578 text="Great host!",

579 rating=0.8,

580 was_appropriate=True,

581 moderation_state_id=moderation_state.id,

582 )

583 session.add(reference)

584 session.flush()

585 moderation_state.object_id = reference.id

586

587 with public_session() as public:

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

589 assert res.HasField("limited_user")

590 assert res.limited_user.username == "limited_user"

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

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

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

594 assert res.limited_user.num_references == 1

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

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

597

598

599def test_GetPublicUser_most_visibility(db):

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

601 user, _ = generate_user(

602 username="most_user",

603 name="Most User",

604 public_visibility=ProfilePublicVisibility.most,

605 )

606

607 with public_session() as public:

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

609 assert res.HasField("most_user")

610 assert res.most_user.username == "most_user"

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

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

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

614

615

616def test_GetPublicUser_full_visibility(db):

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

618 _get_public_users.cache_clear()

619

620 user, _ = generate_user(

621 username="full_user",

622 name="Full User",

623 public_visibility=ProfilePublicVisibility.full,

624 )

625

626 with public_session() as public:

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

628 assert res.HasField("full_user")

629 assert res.full_user.username == "full_user"

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

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

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

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

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

274 statements