Coverage for app/backend/src/tests/test

1from datetime import timedelta

2from typing import Any

4import pytest

5from google.protobuf import empty_pb2, wrappers_pb2

7from couchers.db import session_scope

8from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid

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

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

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

12from tests.fixtures.db import generate_user

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

14from tests.test_communities import create_community, testing_communities # noqa

15from tests.test_references import create_friend_reference

18@pytest.fixture(autouse=True)

19def _(testconfig):

20 pass

23def test_Search(testing_communities):

24 user, token = generate_user()

25 with search_session(token) as api:

26 res = api.Search(

27 search_pb2.SearchReq(

28 query="Country 1, Region 1",

29 include_users=True,

30 include_communities=True,

31 include_groups=True,

32 include_places=True,

33 include_guides=True,

34 )

35 )

36 res = api.Search(

37 search_pb2.SearchReq(

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

39 title_only=True,

40 include_users=True,

41 include_communities=True,

42 include_groups=True,

43 include_places=True,

44 include_guides=True,

45 )

46 )

49def test_UserSearch(testing_communities):

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

51 user, token = generate_user()

53 refresh_materialized_views_rapid(empty_pb2.Empty())

54 refresh_materialized_views(empty_pb2.Empty())

56 with search_session(token) as api:

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

58 assert len(res.results) > 0

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

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

61 assert len(res.results) > 0

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

65def test_regression_search_in_area(db):

66 """

67 Makes sure search_in_area works.

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

70 """

72 # outside

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

74 # outside

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

76 # inside

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

78 # inside

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

80 # outside

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

83 refresh_materialized_views_rapid(empty_pb2.Empty())

84 refresh_materialized_views(empty_pb2.Empty())

86 with search_session(token5) as api:

87 res = api.UserSearch(

88 search_pb2.UserSearchReq(

89 search_in_area=search_pb2.Area(

90 lat=0,

91 lng=0,

92 radius=100000,

93 )

94 )

95 )

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

98 res = api.UserSearchV2(

99 search_pb2.UserSearchReq(

100 search_in_area=search_pb2.Area(

101 lat=0,

102 lng=0,

103 radius=100000,

104 )

105 )

106 )

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

108

109

110def test_user_search_in_rectangle(db):

111 """

112 Makes sure search_in_rectangle works as expected.

113 """

114

115 # outside

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

117 # outside

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

119 # inside

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

121 # inside

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

123 # outside (not fully inside)

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

125 # outside

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

127 # outside

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

129

130 refresh_materialized_views_rapid(empty_pb2.Empty())

131 refresh_materialized_views(empty_pb2.Empty())

132

133 with search_session(token5) as api:

134 res = api.UserSearch(

135 search_pb2.UserSearchReq(

136 search_in_rectangle=search_pb2.RectArea(

137 lat_min=0,

138 lat_max=2,

139 lng_min=0,

140 lng_max=1,

141 )

142 )

143 )

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

145

146 res = api.UserSearchV2(

147 search_pb2.UserSearchReq(

148 search_in_rectangle=search_pb2.RectArea(

149 lat_min=0,

150 lat_max=2,

151 lng_min=0,

152 lng_max=1,

153 )

154 )

155 )

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

157

158

159def test_user_filter_complete_profile(db):

160 """

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

162 """

163 user_complete_profile, token6 = generate_user(complete_profile=True)

164

165 user_incomplete_profile, token7 = generate_user(complete_profile=False)

166

167 refresh_materialized_views_rapid(empty_pb2.Empty())

168 refresh_materialized_views(empty_pb2.Empty())

169

170 with search_session(token7) as api:

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

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

173

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

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

176

177 with search_session(token6) as api:

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

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

180

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

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

183

184

185def test_user_filter_meetup_status(db):

186 """

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

188 """

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

190

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

192

193 refresh_materialized_views_rapid(empty_pb2.Empty())

194 refresh_materialized_views(empty_pb2.Empty())

195

196 with search_session(token8) as api:

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

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

199

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

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

202

203 with search_session(token9) as api:

204 res = api.UserSearch(

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

206 )

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

208

209 res = api.UserSearchV2(

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

211 )

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

213

214

215def test_user_filter_language(db):

216 """

217 Test filtering users by language ability.

218 """

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

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

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

222

223 with session_scope() as session:

224 session.add(

225 LanguageAbility(

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

227 ),

228 )

229 session.add(

230 LanguageAbility(

231 user_id=user_with_japanese_conversational.id,

232 language_code="jpn",

233 fluency=LanguageFluency.fluent,

234 )

235 )

236 session.add(

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

238 )

239

240 refresh_materialized_views_rapid(empty_pb2.Empty())

241 refresh_materialized_views(empty_pb2.Empty())

242

243 with search_session(token11) as api:

244 res = api.UserSearch(

245 search_pb2.UserSearchReq(

246 language_ability_filter=[

247 api_pb2.LanguageAbility(

248 code="deu",

249 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

250 )

251 ]

252 )

253 )

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

255

256 res = api.UserSearchV2(

257 search_pb2.UserSearchReq(

258 language_ability_filter=[

259 api_pb2.LanguageAbility(

260 code="deu",

261 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

262 )

263 ]

264 )

265 )

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

267

268 res = api.UserSearch(

269 search_pb2.UserSearchReq(

270 language_ability_filter=[

271 api_pb2.LanguageAbility(

272 code="jpn",

273 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

274 )

275 ]

276 )

277 )

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

279

280 res = api.UserSearchV2(

281 search_pb2.UserSearchReq(

282 language_ability_filter=[

283 api_pb2.LanguageAbility(

284 code="jpn",

285 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

286 )

287 ]

288 )

289 )

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

291

292

293def test_user_filter_strong_verification(db):

294 user1, token1 = generate_user()

295 user2, _ = generate_user(strong_verification=True)

296 user3, _ = generate_user()

297 user4, _ = generate_user(strong_verification=True)

298 user5, _ = generate_user(strong_verification=True)

299

300 refresh_materialized_views_rapid(empty_pb2.Empty())

301 refresh_materialized_views(empty_pb2.Empty())

302

303 with search_session(token1) as api:

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

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

306

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

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

309

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

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

312

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

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

315

316

317def test_regression_search_only_with_references(db):

318 user1, token1 = generate_user()

319 user2, _ = generate_user()

320 user3, _ = generate_user()

321 user4, _ = generate_user(delete_user=True)

322

323 refresh_materialized_views_rapid(empty_pb2.Empty())

324 refresh_materialized_views(empty_pb2.Empty())

325

326 with session_scope() as session:

327 # user 2 has references

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

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

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

331

332 # user 3 only has reference from a deleted user

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

334

335 with search_session(token1) as api:

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

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

338

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

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

341

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

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

344

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

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

347

348

349def test_user_search_exactly_user_ids(db):

350 """

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

352 """

353 # Create users with different properties

354 user1, token1 = generate_user()

355 user2, _ = generate_user(strong_verification=True)

356 user3, _ = generate_user(complete_profile=True)

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

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

359

360 refresh_materialized_views_rapid(empty_pb2.Empty())

361 refresh_materialized_views(empty_pb2.Empty())

362

363 with search_session(token1) as api:

364 # Test that exactly_user_ids returns only the specified users

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

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

367

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

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

370

371 # Test that exactly_user_ids ignores other filters

372 res = api.UserSearch(

373 search_pb2.UserSearchReq(

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

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

376 )

377 )

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

379

380 res = api.UserSearchV2(

381 search_pb2.UserSearchReq(

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

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

384 )

385 )

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

387

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

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

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

391

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

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

394

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

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

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

398

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

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

401

402

403@pytest.fixture

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

405 """Dummy data for creating events."""

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

407 end_time = start_time + timedelta(hours=3)

408 return {

409 "title": "Dummy Title",

410 "content": "Dummy content.",

411 "photo_key": None,

412 "offline_information": events_pb2.OfflineEventInformation(address="Near Null Island", lat=0.1, lng=0.2),

413 "start_time": Timestamp_from_datetime(start_time),

414 "end_time": Timestamp_from_datetime(end_time),

415 "timezone": "UTC",

416 }

417

418

419@pytest.fixture

420def create_event(sample_event_data):

421 """Factory for creating events."""

422

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

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

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

426

427 return _create_event

428

429

430@pytest.fixture

431def sample_community(db) -> int:

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

433 user, _ = generate_user()

434 with session_scope() as session:

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

436

437

438def test_EventSearch_no_filters(testing_communities):

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

440 user, token = generate_user()

441 with search_session(token) as api:

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

443 assert len(res.events) > 0

444

445

446def test_event_search_by_query(sample_community, create_event):

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

448 user, token = generate_user()

449

450 with events_session(token) as api:

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

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

453 create_event(api)

454

455 with search_session(token) as api:

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

457 assert len(res.events) == 2

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

459

460 res = api.EventSearch(

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

462 )

463 assert len(res.events) == 1

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

465

466

467def test_event_search_by_time(sample_community, create_event):

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

469 user, token = generate_user()

470

471 with events_session(token) as api:

472 event1 = create_event(

473 api,

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

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

476 )

477 event2 = create_event(

478 api,

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

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

481 )

482 event3 = create_event(

483 api,

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

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

486 )

487

488 with search_session(token) as api:

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

490 assert len(res.events) == 2

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

492

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

494 assert len(res.events) == 2

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

496

497 res = api.EventSearch(

498 search_pb2.EventSearchReq(

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

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

501 )

502 )

503 assert len(res.events) == 1

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

505

506

507def test_event_search_by_circle(sample_community, create_event):

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

509 user, token = generate_user()

510

511 with events_session(token) as api:

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

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

514 create_event(

515 api,

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

517 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),

518 )

519

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

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

522 create_event(

523 api,

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

525 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),

526 )

527

528 with search_session(token) as api:

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

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

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

532

533

534def test_event_search_by_rectangle(sample_community, create_event):

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

536 user, token = generate_user()

537

538 with events_session(token) as api:

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

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

541 create_event(

542 api,

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

544 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),

545 )

546

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

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

549 create_event(

550 api,

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

552 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),

553 )

554

555 with search_session(token) as api:

556 res = api.EventSearch(

557 search_pb2.EventSearchReq(

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

559 )

560 )

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

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

563

564

565def test_event_search_pagination(sample_community, create_event):

566 """Test that EventSearch paginates correctly.

567

568 Check that

569 - <page_size> events are returned, if available

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

571 - the next page token is correct

572 """

573 user, token = generate_user()

574

575 anchor_time = now()

576 with events_session(token) as api:

577 for i in range(5):

578 create_event(

579 api,

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

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

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

583 )

584

585 with search_session(token) as api:

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

587 assert len(res.events) == 4

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

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

590

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

592 assert len(res.events) == 1

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

594 assert res.next_page_token == ""

595

596 res = api.EventSearch(

597 search_pb2.EventSearchReq(

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

599 )

600 )

601 assert len(res.events) == 2

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

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

604

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

606 assert len(res.events) == 2

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

608 assert res.next_page_token == ""

609

610

611def test_event_search_pagination_with_page_number(sample_community, create_event):

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

613

614 Check that

615 - <page_size> events are returned, if available

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

617 - <page_number> is respected

618 - <total_items> is correct

619 """

620 user, token = generate_user()

621

622 anchor_time = now()

623 with events_session(token) as api:

624 for i in range(5):

625 create_event(

626 api,

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

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

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

630 )

631

632 with search_session(token) as api:

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

634 assert len(res.events) == 2

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

636 assert res.total_items == 5

637

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

639 assert len(res.events) == 2

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

641 assert res.total_items == 5

642

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

644 assert len(res.events) == 1

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

646 assert res.total_items == 5

647

648 # Verify no more pages

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

650 assert not res.events

651 assert res.total_items == 5

652

653

654def test_event_search_online_status(sample_community, create_event):

655 """Test that EventSearch respects only_online and only_offline filters and by default returns both."""

656 user, token = generate_user()

657

658 with events_session(token) as api:

659 create_event(api, title="Offline event")

660

661 create_event(

662 api,

663 title="Online event",

664 online_information=events_pb2.OnlineEventInformation(link="https://couchers.org/meet/"),

665 parent_community_id=sample_community,

666 offline_information=events_pb2.OfflineEventInformation(),

667 )

668

669 with search_session(token) as api:

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

671 assert len(res.events) == 2

672 assert {event.title for event in res.events} == {"Offline event", "Online event"}

673

674 res = api.EventSearch(search_pb2.EventSearchReq(only_online=True))

675 assert {event.title for event in res.events} == {"Online event"}

676

677 res = api.EventSearch(search_pb2.EventSearchReq(only_offline=True))

678 assert {event.title for event in res.events} == {"Offline event"}

679

680

681def test_event_search_filter_subscription_attendance_organizing_my_communities(sample_community, create_event):

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

683 returns all events.

684 """

685 _, token = generate_user()

686 other_user, other_token = generate_user()

687

688 with communities_session(token) as api:

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

690

691 with session_scope() as session:

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

693

694 with events_session(other_token) as api:

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

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

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

698 create_event(

699 api,

700 title="Other community event",

701 offline_information=events_pb2.OfflineEventInformation(lat=58, lng=1, address="Somewhere"),

702 )

703

704 with events_session(token) as api:

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

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

707 api.SetEventAttendance(

708 events_pb2.SetEventAttendanceReq(

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

710 )

711 )

712

713 with search_session(token) as api:

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

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

716 "Subscribed event",

717 "Attending event",

718 "Community event",

719 "Other community event",

720 "Organized event",

721 }

722

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

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

725

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

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

728

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

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

731

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

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

734 "Subscribed event",

735 "Attending event",

736 "Community event",

737 "Organized event",

738 }

739

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

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

742

743

744def test_regression_search_multiple_pages(db):

745 """

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

747 """

748 user, token = generate_user()

749 user_ids = [user.id]

750 for _ in range(10):

751 other_user, _ = generate_user()

752 user_ids.append(other_user.id)

753

754 refresh_materialized_views_rapid(empty_pb2.Empty())

755 refresh_materialized_views(empty_pb2.Empty())

756

757 with search_session(token) as api:

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

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

760 assert res.next_page_token

761

762

763def test_regression_search_no_results(db):

764 """

765 There was a bug when there were no results

766 """

767 # put us far away

768 user, token = generate_user()

769

770 refresh_materialized_views_rapid(empty_pb2.Empty())

771 refresh_materialized_views(empty_pb2.Empty())

772

773 with search_session(token) as api:

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

775 assert len(res.results) == 0

776

777

778def test_user_filter_same_gender_only(db):

779 """Test that same_gender_only filter works correctly"""

780 # Create users with different genders and strong verification status

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

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

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

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

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

786

787 refresh_materialized_views_rapid(empty_pb2.Empty())

788 refresh_materialized_views(empty_pb2.Empty())

789

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

791 with search_session(token_woman_with_sv) as api:

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

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

794 assert woman_with_sv.id in result_ids

795 assert woman_without_sv.id in result_ids

796 assert other_woman_with_sv.id in result_ids

797 assert man_with_sv.id not in result_ids

798 assert man_without_sv.id not in result_ids

799

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

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

802 assert woman_with_sv.id in result_ids

803 assert woman_without_sv.id in result_ids

804 assert other_woman_with_sv.id in result_ids

805 assert man_with_sv.id not in result_ids

806 assert man_without_sv.id not in result_ids

807

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

809 with search_session(token_man_with_sv) as api:

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

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

812 assert man_with_sv.id in result_ids

813 assert man_without_sv.id in result_ids

814 assert woman_with_sv.id not in result_ids

815 assert woman_without_sv.id not in result_ids

816 assert other_woman_with_sv.id not in result_ids

817

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

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

820 assert man_with_sv.id in result_ids

821 assert man_without_sv.id in result_ids

822 assert woman_with_sv.id not in result_ids

823 assert woman_without_sv.id not in result_ids

824 assert other_woman_with_sv.id not in result_ids

825

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

827 with search_session(token_woman_without_sv) as api:

828 with pytest.raises(Exception) as e:

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

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

831

832 with pytest.raises(Exception) as e:

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

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

835

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

837 with search_session(token_woman_with_sv) as api:

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

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

840 assert woman_with_sv.id in result_ids

841 assert woman_without_sv.id in result_ids

842 assert other_woman_with_sv.id in result_ids

843 assert man_with_sv.id in result_ids

844 assert man_without_sv.id in result_ids

845

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

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

848 assert woman_with_sv.id in result_ids

849 assert woman_without_sv.id in result_ids

850 assert other_woman_with_sv.id in result_ids

851 assert man_with_sv.id in result_ids

852 assert man_without_sv.id in result_ids

853

854

855def test_user_filter_same_gender_only_with_other_filters(db):

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

857 # Create users with different properties

858 woman_host, token_woman = generate_user(

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

860 )

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

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

863

864 refresh_materialized_views_rapid(empty_pb2.Empty())

865 refresh_materialized_views(empty_pb2.Empty())

866

867 # Test: Combine same_gender_only with hosting_status filter

868 with search_session(token_woman) as api:

869 res = api.UserSearch(

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

871 )

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

873 # Should only see woman who can host

874 assert woman_host.id in result_ids

875 assert woman_cant_host.id not in result_ids

876 assert man_host.id not in result_ids

877

878 res = api.UserSearchV2(

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

880 )

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

882 assert woman_host.id in result_ids

883 assert woman_cant_host.id not in result_ids

884 assert man_host.id not in result_ids

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

430 statements