Coverage for app/backend/src/tests/test

1from datetime import timedelta

2from typing import Any

4import grpc

5import pytest

6from google.protobuf import empty_pb2, wrappers_pb2

7from sqlalchemy import select

9from couchers.db import session_scope

10from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid

11from couchers.models import EventOccurrence, HostingStatus, LanguageAbility, LanguageFluency, MeetupStatus

12from couchers.proto import api_pb2, communities_pb2, events_pb2, search_pb2

13from couchers.utils import Timestamp_from_datetime, create_coordinate, millis_from_dt, now

14from tests.fixtures.db import generate_user

15from tests.fixtures.misc import Moderator

16from tests.fixtures.sessions import communities_session, events_session, search_session

17from tests.test_communities import create_community, testing_communities # noqa

18from tests.test_references import create_friend_reference

21@pytest.fixture(autouse=True)

22def _(testconfig):

23 pass

26def test_Search(testing_communities):

27 user, token = generate_user()

28 with search_session(token) as api:

29 res = api.Search(

30 search_pb2.SearchReq(

31 query="Country 1, Region 1",

32 include_users=True,

33 include_communities=True,

34 include_groups=True,

35 include_places=True,

36 include_guides=True,

37 )

38 )

39 res = api.Search(

40 search_pb2.SearchReq(

41 query="Country 1, Region 1, Attraction",

42 title_only=True,

43 include_users=True,

44 include_communities=True,

45 include_groups=True,

46 include_places=True,

47 include_guides=True,

48 )

49 )

52def test_UserSearch(testing_communities):

53 """Test that UserSearch returns all users if no filter is set."""

54 user, token = generate_user()

56 refresh_materialized_views_rapid(empty_pb2.Empty())

57 refresh_materialized_views(empty_pb2.Empty())

59 with search_session(token) as api:

60 res = api.UserSearch(search_pb2.UserSearchReq())

61 assert len(res.results) > 0

62 assert res.total_items == len(res.results)

63 res = api.UserSearchV2(search_pb2.UserSearchReq())

64 assert len(res.results) > 0

65 assert res.total_items == len(res.results)

68def test_regression_search_in_area(db):

69 """

70 Makes sure search_in_area works.

72 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.

73 """

75 # outside

76 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)

77 # outside

78 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)

79 # inside

80 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)

81 # inside

82 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)

83 # outside

84 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)

86 refresh_materialized_views_rapid(empty_pb2.Empty())

87 refresh_materialized_views(empty_pb2.Empty())

89 with search_session(token5) as api:

90 res = api.UserSearch(

91 search_pb2.UserSearchReq(

92 search_in_area=search_pb2.Area(

93 lat=0,

94 lng=0,

95 radius=100000,

96 )

97 )

98 )

99 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]

100

101 res = api.UserSearchV2(

102 search_pb2.UserSearchReq(

103 search_in_area=search_pb2.Area(

104 lat=0,

105 lng=0,

106 radius=100000,

107 )

108 )

109 )

110 assert [result.user_id for result in res.results] == [user3.id, user4.id]

111

112

113def test_user_search_in_rectangle(db):

114 """

115 Makes sure search_in_rectangle works as expected.

116 """

117

118 # outside

119 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)

120 # outside

121 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)

122 # inside

123 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)

124 # inside

125 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)

126 # outside (not fully inside)

127 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)

128 # outside

129 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)

130 # outside

131 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)

132

133 refresh_materialized_views_rapid(empty_pb2.Empty())

134 refresh_materialized_views(empty_pb2.Empty())

135

136 with search_session(token5) as api:

137 res = api.UserSearch(

138 search_pb2.UserSearchReq(

139 search_in_rectangle=search_pb2.RectArea(

140 lat_min=0,

141 lat_max=2,

142 lng_min=0,

143 lng_max=1,

144 )

145 )

146 )

147 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]

148

149 res = api.UserSearchV2(

150 search_pb2.UserSearchReq(

151 search_in_rectangle=search_pb2.RectArea(

152 lat_min=0,

153 lat_max=2,

154 lng_min=0,

155 lng_max=1,

156 )

157 )

158 )

159 assert [result.user_id for result in res.results] == [user3.id, user4.id]

160

161

162def test_user_filter_complete_profile(db):

163 """

164 Make sure the completed profile flag returns only completed user profile

165 """

166 user_complete_profile, token6 = generate_user(complete_profile=True)

167

168 user_incomplete_profile, token7 = generate_user(complete_profile=False)

169

170 refresh_materialized_views_rapid(empty_pb2.Empty())

171 refresh_materialized_views(empty_pb2.Empty())

172

173 with search_session(token7) as api:

174 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))

175 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]

176

177 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))

178 assert user_incomplete_profile.id in [result.user_id for result in res.results]

179

180 with search_session(token6) as api:

181 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))

182 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]

183

184 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))

185 assert [result.user_id for result in res.results] == [user_complete_profile.id]

186

187

188def test_user_filter_meetup_status(db):

189 """

190 Make sure the completed profile flag returns only completed user profile

191 """

192 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)

193

194 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)

195

196 refresh_materialized_views_rapid(empty_pb2.Empty())

197 refresh_materialized_views(empty_pb2.Empty())

198

199 with search_session(token8) as api:

200 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))

201 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]

202

203 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))

204 assert user_wants_to_meetup.id in [result.user_id for result in res.results]

205

206 with search_session(token9) as api:

207 res = api.UserSearch(

208 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])

209 )

210 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]

211

212 res = api.UserSearchV2(

213 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])

214 )

215 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]

216

217

218def test_user_filter_language(db):

219 """

220 Test filtering users by language ability.

221 """

222 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)

223 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)

224 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)

225

226 with session_scope() as session:

227 session.add(

228 LanguageAbility(

229 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner

230 ),

231 )

232 session.add(

233 LanguageAbility(

234 user_id=user_with_japanese_conversational.id,

235 language_code="jpn",

236 fluency=LanguageFluency.fluent,

237 )

238 )

239 session.add(

240 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)

241 )

242

243 refresh_materialized_views_rapid(empty_pb2.Empty())

244 refresh_materialized_views(empty_pb2.Empty())

245

246 with search_session(token11) as api:

247 res = api.UserSearch(

248 search_pb2.UserSearchReq(

249 language_ability_filter=[

250 api_pb2.LanguageAbility(

251 code="deu",

252 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

253 )

254 ]

255 )

256 )

257 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]

258

259 res = api.UserSearchV2(

260 search_pb2.UserSearchReq(

261 language_ability_filter=[

262 api_pb2.LanguageAbility(

263 code="deu",

264 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

265 )

266 ]

267 )

268 )

269 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]

270

271 res = api.UserSearch(

272 search_pb2.UserSearchReq(

273 language_ability_filter=[

274 api_pb2.LanguageAbility(

275 code="jpn",

276 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

277 )

278 ]

279 )

280 )

281 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]

282

283 res = api.UserSearchV2(

284 search_pb2.UserSearchReq(

285 language_ability_filter=[

286 api_pb2.LanguageAbility(

287 code="jpn",

288 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

289 )

290 ]

291 )

292 )

293 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]

294

295

296def test_user_filter_strong_verification(db):

297 user1, token1 = generate_user()

298 user2, _ = generate_user(strong_verification=True)

299 user3, _ = generate_user()

300 user4, _ = generate_user(strong_verification=True)

301 user5, _ = generate_user(strong_verification=True)

302

303 refresh_materialized_views_rapid(empty_pb2.Empty())

304 refresh_materialized_views(empty_pb2.Empty())

305

306 with search_session(token1) as api:

307 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))

308 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]

309

310 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))

311 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]

312

313 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))

314 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]

315

316 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))

317 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]

318

319

320def test_regression_search_only_with_references(db):

321 user1, token1 = generate_user()

322 user2, _ = generate_user()

323 user3, _ = generate_user()

324 user4, _ = generate_user(delete_user=True)

325

326 refresh_materialized_views_rapid(empty_pb2.Empty())

327 refresh_materialized_views(empty_pb2.Empty())

328

329 with session_scope() as session:

330 # user 2 has references

331 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))

332 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))

333 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))

334

335 # user 3 only has reference from a deleted user

336 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))

337

338 with search_session(token1) as api:

339 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))

340 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]

341

342 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))

343 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]

344

345 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))

346 assert [result.user.user_id for result in res.results] == [user2.id]

347

348 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))

349 assert [result.user_id for result in res.results] == [user2.id]

350

351

352def test_user_search_exactly_user_ids(db):

353 """

354 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.

355 """

356 # Create users with different properties

357 user1, token1 = generate_user()

358 user2, _ = generate_user(strong_verification=True)

359 user3, _ = generate_user(complete_profile=True)

360 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)

361 user5, _ = generate_user(delete_user=True) # Deleted user

362

363 refresh_materialized_views_rapid(empty_pb2.Empty())

364 refresh_materialized_views(empty_pb2.Empty())

365

366 with search_session(token1) as api:

367 # Test that exactly_user_ids returns only the specified users

368 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))

369 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

370

371 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))

372 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

373

374 # Test that exactly_user_ids ignores other filters

375 res = api.UserSearch(

376 search_pb2.UserSearchReq(

377 exactly_user_ids=[user2.id, user3.id, user4.id],

378 only_with_strong_verification=True, # This would normally filter out user3 and user4

379 )

380 )

381 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

382

383 res = api.UserSearchV2(

384 search_pb2.UserSearchReq(

385 exactly_user_ids=[user2.id, user3.id, user4.id],

386 only_with_strong_verification=True, # This would normally filter out user3 and user4

387 )

388 )

389 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

390

391 # Test with non-existent user IDs (should be ignored)

392 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))

393 assert [result.user.user_id for result in res.results] == [user1.id]

394

395 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))

396 assert [result.user_id for result in res.results] == [user1.id]

397

398 # Test with deleted user ID (should be ignored due to visibility filter)

399 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))

400 assert [result.user.user_id for result in res.results] == [user1.id]

401

402 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))

403 assert [result.user_id for result in res.results] == [user1.id]

404

405

406@pytest.fixture

407def sample_event_data() -> dict[str, Any]:

408 """Dummy data for creating events."""

409 start_time = now() + timedelta(hours=2)

410 end_time = start_time + timedelta(hours=3)

411 return {

412 "title": "Dummy Title",

413 "content": "Dummy content.",

414 "photo_key": None,

415 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),

416 "start_time": Timestamp_from_datetime(start_time),

417 "end_time": Timestamp_from_datetime(end_time),

418 "timezone": "UTC",

419 }

420

421

422@pytest.fixture

423def create_event(sample_event_data):

424 """Factory for creating events."""

425

426 def _create_event(event_api, **kwargs) -> EventOccurrence:

427 """Create an event with default values, unless overridden by kwargs."""

428 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore

429

430 return _create_event

431

432

433@pytest.fixture

434def sample_community(db) -> int:

435 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""

436 user, _ = generate_user()

437 with session_scope() as session:

438 return create_community(session, -50, 50, "Community", [user], [], None).id

439

440

441def test_EventSearch_no_filters(testing_communities):

442 """Test that EventSearch returns all events if no filter is set."""

443 user, token = generate_user()

444 with search_session(token) as api:

445 res = api.EventSearch(search_pb2.EventSearchReq())

446 assert len(res.events) > 0

447

448

449def test_event_search_by_query(sample_community, create_event):

450 """Test that EventSearch finds events by title (and content if query_title_only=False)."""

451 user, token = generate_user()

452

453 with events_session(token) as api:

454 event1 = create_event(api, title="Lorem Ipsum")

455 event2 = create_event(api, content="Lorem Ipsum")

456 create_event(api)

457

458 with search_session(token) as api:

459 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))

460 assert len(res.events) == 2

461 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}

462

463 res = api.EventSearch(

464 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)

465 )

466 assert len(res.events) == 1

467 assert res.events[0].event_id == event1.event_id

468

469

470def test_event_search_by_time(sample_community, create_event):

471 """Test that EventSearch filters with the given time range."""

472 user, token = generate_user()

473

474 with events_session(token) as api:

475 event1 = create_event(

476 api,

477 start_time=Timestamp_from_datetime(now() + timedelta(hours=1)),

478 end_time=Timestamp_from_datetime(now() + timedelta(hours=2)),

479 )

480 event2 = create_event(

481 api,

482 start_time=Timestamp_from_datetime(now() + timedelta(hours=4)),

483 end_time=Timestamp_from_datetime(now() + timedelta(hours=5)),

484 )

485 event3 = create_event(

486 api,

487 start_time=Timestamp_from_datetime(now() + timedelta(hours=7)),

488 end_time=Timestamp_from_datetime(now() + timedelta(hours=8)),

489 )

490

491 with search_session(token) as api:

492 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))

493 assert len(res.events) == 2

494 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}

495

496 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))

497 assert len(res.events) == 2

498 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}

499

500 res = api.EventSearch(

501 search_pb2.EventSearchReq(

502 before=Timestamp_from_datetime(now() + timedelta(hours=6)),

503 after=Timestamp_from_datetime(now() + timedelta(hours=3)),

504 )

505 )

506 assert len(res.events) == 1

507 assert res.events[0].event_id == event2.event_id

508

509

510def test_event_search_by_circle(sample_community, create_event):

511 """Test that EventSearch only returns events within the given circle."""

512 user, token = generate_user()

513

514 with events_session(token) as api:

515 inside_pts = [(0.1, 0.01), (0.01, 0.1)]

516 for i, (lat, lng) in enumerate(inside_pts):

517 create_event(

518 api,

519 title=f"Inside area {i}",

520 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),

521 )

522

523 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]

524 for i, (lat, lng) in enumerate(outside_pts):

525 create_event(

526 api,

527 title=f"Outside area {i}",

528 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),

529 )

530

531 with search_session(token) as api:

532 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))

533 assert len(res.events) == len(inside_pts)

534 assert all(event.title.startswith("Inside area") for event in res.events)

535

536

537def test_event_search_by_rectangle(sample_community, create_event):

538 """Test that EventSearch only returns events within the given rectangular area."""

539 user, token = generate_user()

540

541 with events_session(token) as api:

542 inside_pts = [(0.1, 0.2), (1.2, 0.2)]

543 for i, (lat, lng) in enumerate(inside_pts):

544 create_event(

545 api,

546 title=f"Inside area {i}",

547 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),

548 )

549

550 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]

551 for i, (lat, lng) in enumerate(outside_pts):

552 create_event(

553 api,

554 title=f"Outside area {i}",

555 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),

556 )

557

558 with search_session(token) as api:

559 res = api.EventSearch(

560 search_pb2.EventSearchReq(

561 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)

562 )

563 )

564 assert len(res.events) == len(inside_pts)

565 assert all(event.title.startswith("Inside area") for event in res.events)

566

567

568def test_event_search_pagination(sample_community, create_event):

569 """Test that EventSearch paginates correctly.

570

571 Check that

572 - <page_size> events are returned, if available

573 - sort order is applied (default: past=False)

574 - the next page token is correct

575 """

576 user, token = generate_user()

577

578 anchor_time = now()

579 with events_session(token) as api:

580 for i in range(5):

581 create_event(

582 api,

583 title=f"Event {i + 1}",

584 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),

585 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),

586 )

587

588 with search_session(token) as api:

589 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))

590 assert len(res.events) == 4

591 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]

592 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=5, minutes=30)))

593

594 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))

595 assert len(res.events) == 1

596 assert res.events[0].title == "Event 5"

597 assert res.next_page_token == ""

598

599 res = api.EventSearch(

600 search_pb2.EventSearchReq(

601 past=True, page_size=2, page_token=str(millis_from_dt(anchor_time + timedelta(hours=4, minutes=30)))

602 )

603 )

604 assert len(res.events) == 2

605 assert [event.title for event in res.events] == ["Event 4", "Event 3"]

606 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=2, minutes=30)))

607

608 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))

609 assert len(res.events) == 2

610 assert [event.title for event in res.events] == ["Event 2", "Event 1"]

611 assert res.next_page_token == ""

612

613

614def test_event_search_pagination_with_page_number(sample_community, create_event):

615 """Test that EventSearch paginates correctly with page number.

616

617 Check that

618 - <page_size> events are returned, if available

619 - sort order is applied (default: past=False)

620 - <page_number> is respected

621 - <total_items> is correct

622 """

623 user, token = generate_user()

624

625 anchor_time = now()

626 with events_session(token) as api:

627 for i in range(5):

628 create_event(

629 api,

630 title=f"Event {i + 1}",

631 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),

632 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),

633 )

634

635 with search_session(token) as api:

636 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))

637 assert len(res.events) == 2

638 assert [event.title for event in res.events] == ["Event 1", "Event 2"]

639 assert res.total_items == 5

640

641 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))

642 assert len(res.events) == 2

643 assert [event.title for event in res.events] == ["Event 3", "Event 4"]

644 assert res.total_items == 5

645

646 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))

647 assert len(res.events) == 1

648 assert [event.title for event in res.events] == ["Event 5"]

649 assert res.total_items == 5

650

651 # Verify no more pages

652 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))

653 assert not res.events

654 assert res.total_items == 5

655

656

657def test_event_search_filter_subscription_attendance_organizing_my_communities(

658 sample_community, create_event, moderator: Moderator

659):

660 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default

661 returns all events.

662 """

663 _, token = generate_user()

664 other_user, other_token = generate_user()

665

666 with communities_session(token) as api:

667 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))

668

669 with session_scope() as session:

670 create_community(session, 55, 60, "Other community", [other_user], [], None)

671

672 with events_session(other_token) as api:

673 e_subscribed = create_event(api, title="Subscribed event")

674 e_attending = create_event(api, title="Attending event")

675 create_event(api, title="Community event")

676 create_event(

677 api,

678 title="Other community event",

679 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),

680 )

681

682 # Approve all events so they're visible to other users

683 with session_scope() as session:

684 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()

685 for oid in occurrence_ids:

686 moderator.approve_event_occurrence(oid)

687

688 with events_session(token) as api:

689 create_event(api, title="Organized event")

690 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))

691 api.SetEventAttendance(

692 events_pb2.SetEventAttendanceReq(

693 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING

694 )

695 )

696

697 with search_session(token) as api:

698 res = api.EventSearch(search_pb2.EventSearchReq())

699 assert {event.title for event in res.events} == {

700 "Subscribed event",

701 "Attending event",

702 "Community event",

703 "Other community event",

704 "Organized event",

705 }

706

707 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))

708 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}

709

710 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))

711 assert {event.title for event in res.events} == {"Attending event", "Organized event"}

712

713 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))

714 assert {event.title for event in res.events} == {"Organized event"}

715

716 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))

717 assert {event.title for event in res.events} == {

718 "Subscribed event",

719 "Attending event",

720 "Community event",

721 "Organized event",

722 }

723

724 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))

725 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}

726

727

728def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):

729 """Test that exclude_attending removes events the user is attending or organizing."""

730 user, token = generate_user()

731 other_user, other_token = generate_user()

732

733 with communities_session(token) as api:

734 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))

735

736 with session_scope() as session:

737 create_community(session, 55, 60, "Other community", [other_user], [], None)

738

739 with events_session(other_token) as api:

740 e_attending = create_event(api, title="Attending event")

741 e_community_only = create_event(api, title="Community only event")

742 create_event(

743 api,

744 title="Other community event",

745 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),

746 )

747

748 with session_scope() as session:

749 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()

750 for oid in occurrence_ids:

751 moderator.approve_event_occurrence(oid)

752

753 with events_session(token) as api:

754 e_organized = create_event(api, title="Organized event")

755 api.SetEventAttendance(

756 events_pb2.SetEventAttendanceReq(

757 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING

758 )

759 )

760

761 with search_session(token) as api:

762 # baseline: my_communities returns all community events including attended/organized

763 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))

764 assert {event.title for event in res.events} == {

765 "Attending event",

766 "Community only event",

767 "Organized event",

768 }

769

770 # my_communities + exclude_attending: drops attended and organized events

771 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))

772 assert {event.title for event in res.events} == {"Community only event"}

773

774 # exclude_attending alone (no other filter = all events): drops attended and organized

775 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))

776 assert {event.title for event in res.events} == {"Community only event", "Other community event"}

777

778 # attending + exclude_attending is invalid

779 with pytest.raises(grpc.RpcError) as e:

780 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))

781 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT

782

783

784def test_regression_search_multiple_pages(db):

785 """

786 There was a bug when there are multiple pages of results

787 """

788 user, token = generate_user()

789 user_ids = [user.id]

790 for _ in range(10):

791 other_user, _ = generate_user()

792 user_ids.append(other_user.id)

793

794 refresh_materialized_views_rapid(empty_pb2.Empty())

795 refresh_materialized_views(empty_pb2.Empty())

796

797 with search_session(token) as api:

798 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))

799 assert [result.user_id for result in res.results] == user_ids[:5]

800 assert res.next_page_token

801

802

803def test_regression_search_no_results(db):

804 """

805 There was a bug when there were no results

806 """

807 # put us far away

808 user, token = generate_user()

809

810 refresh_materialized_views_rapid(empty_pb2.Empty())

811 refresh_materialized_views(empty_pb2.Empty())

812

813 with search_session(token) as api:

814 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))

815 assert len(res.results) == 0

816

817

818def test_user_filter_same_gender_only(db):

819 """Test that same_gender_only filter works correctly"""

820 # Create users with different genders and strong verification status

821 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")

822 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")

823 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")

824 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")

825 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")

826

827 refresh_materialized_views_rapid(empty_pb2.Empty())

828 refresh_materialized_views(empty_pb2.Empty())

829

830 # Test 1: Woman with strong verification should see only women when same_gender_only=True

831 with search_session(token_woman_with_sv) as api:

832 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

833 result_ids = [result.user.user_id for result in res.results]

834 assert woman_with_sv.id in result_ids

835 assert woman_without_sv.id in result_ids

836 assert other_woman_with_sv.id in result_ids

837 assert man_with_sv.id not in result_ids

838 assert man_without_sv.id not in result_ids

839

840 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

841 result_ids = [result.user_id for result in res.results]

842 assert woman_with_sv.id in result_ids

843 assert woman_without_sv.id in result_ids

844 assert other_woman_with_sv.id in result_ids

845 assert man_with_sv.id not in result_ids

846 assert man_without_sv.id not in result_ids

847

848 # Test 2: Man with strong verification should see only men when same_gender_only=True

849 with search_session(token_man_with_sv) as api:

850 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

851 result_ids = [result.user.user_id for result in res.results]

852 assert man_with_sv.id in result_ids

853 assert man_without_sv.id in result_ids

854 assert woman_with_sv.id not in result_ids

855 assert woman_without_sv.id not in result_ids

856 assert other_woman_with_sv.id not in result_ids

857

858 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

859 result_ids = [result.user_id for result in res.results]

860 assert man_with_sv.id in result_ids

861 assert man_without_sv.id in result_ids

862 assert woman_with_sv.id not in result_ids

863 assert woman_without_sv.id not in result_ids

864 assert other_woman_with_sv.id not in result_ids

865

866 # Test 3: Woman without strong verification should get an error

867 with search_session(token_woman_without_sv) as api:

868 with pytest.raises(Exception) as e:

869 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

870 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)

871

872 with pytest.raises(Exception) as e:

873 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

874 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)

875

876 # Test 4: When same_gender_only=False, should see all users

877 with search_session(token_woman_with_sv) as api:

878 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))

879 result_ids = [result.user.user_id for result in res.results]

880 assert woman_with_sv.id in result_ids

881 assert woman_without_sv.id in result_ids

882 assert other_woman_with_sv.id in result_ids

883 assert man_with_sv.id in result_ids

884 assert man_without_sv.id in result_ids

885

886 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))

887 result_ids = [result.user_id for result in res.results]

888 assert woman_with_sv.id in result_ids

889 assert woman_without_sv.id in result_ids

890 assert other_woman_with_sv.id in result_ids

891 assert man_with_sv.id in result_ids

892 assert man_without_sv.id in result_ids

893

894

895def test_user_filter_same_gender_only_with_other_filters(db):

896 """Test that same_gender_only filter works correctly combined with other filters"""

897 # Create users with different properties

898 woman_host, token_woman = generate_user(

899 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host

900 )

901 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)

902 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)

903

904 refresh_materialized_views_rapid(empty_pb2.Empty())

905 refresh_materialized_views(empty_pb2.Empty())

906

907 # Test: Combine same_gender_only with hosting_status filter

908 with search_session(token_woman) as api:

909 res = api.UserSearch(

910 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])

911 )

912 result_ids = [result.user.user_id for result in res.results]

913 # Should only see woman who can host

914 assert woman_host.id in result_ids

915 assert woman_cant_host.id not in result_ids

916 assert man_host.id not in result_ids

917

918 res = api.UserSearchV2(

919 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])

920 )

921 result_ids = [result.user_id for result in res.results]

922 assert woman_host.id in result_ids

923 assert woman_cant_host.id not in result_ids

924 assert man_host.id not in result_ids

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

452 statements