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

1from datetime import timedelta 

2 

3import pytest 

4from google.protobuf import wrappers_pb2 

5 

6from couchers.db import session_scope 

7from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid 

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

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

10from proto import api_pb2, communities_pb2, events_pb2, search_pb2 

11from tests.test_communities import create_community, testing_communities # noqa 

12from tests.test_fixtures import ( # noqa 

13 communities_session, 

14 db, 

15 events_session, 

16 generate_user, 

17 search_session, 

18 testconfig, 

19) 

20from tests.test_references import create_friend_reference 

21 

22 

23@pytest.fixture(autouse=True) 

24def _(testconfig): 

25 pass 

26 

27 

28def test_Search(testing_communities): 

29 user, token = generate_user() 

30 with search_session(token) as api: 

31 res = api.Search( 

32 search_pb2.SearchReq( 

33 query="Country 1, Region 1", 

34 include_users=True, 

35 include_communities=True, 

36 include_groups=True, 

37 include_places=True, 

38 include_guides=True, 

39 ) 

40 ) 

41 res = api.Search( 

42 search_pb2.SearchReq( 

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

44 title_only=True, 

45 include_users=True, 

46 include_communities=True, 

47 include_groups=True, 

48 include_places=True, 

49 include_guides=True, 

50 ) 

51 ) 

52 

53 

54def test_UserSearch(testing_communities): 

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

56 user, token = generate_user() 

57 

58 refresh_materialized_views_rapid(None) 

59 refresh_materialized_views(None) 

60 

61 with search_session(token) as api: 

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

63 assert len(res.results) > 0 

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

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

66 assert len(res.results) > 0 

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

68 

69 

70def test_regression_search_in_area(db): 

71 """ 

72 Makes sure search_in_area works. 

73 

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

75 """ 

76 

77 # outside 

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

79 # outside 

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

81 # inside 

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

83 # inside 

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

85 # outside 

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

87 

88 refresh_materialized_views_rapid(None) 

89 refresh_materialized_views(None) 

90 

91 with search_session(token5) as api: 

92 res = api.UserSearch( 

93 search_pb2.UserSearchReq( 

94 search_in_area=search_pb2.Area( 

95 lat=0, 

96 lng=0, 

97 radius=100000, 

98 ) 

99 ) 

100 ) 

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

102 

103 res = api.UserSearchV2( 

104 search_pb2.UserSearchReq( 

105 search_in_area=search_pb2.Area( 

106 lat=0, 

107 lng=0, 

108 radius=100000, 

109 ) 

110 ) 

111 ) 

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

113 

114 

115def test_user_search_in_rectangle(db): 

116 """ 

117 Makes sure search_in_rectangle works as expected. 

118 """ 

119 

120 # outside 

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

122 # outside 

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

124 # inside 

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

126 # inside 

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

128 # outside (not fully inside) 

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

130 # outside 

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

132 # outside 

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

134 

135 refresh_materialized_views_rapid(None) 

136 refresh_materialized_views(None) 

137 

138 with search_session(token5) as api: 

139 res = api.UserSearch( 

140 search_pb2.UserSearchReq( 

141 search_in_rectangle=search_pb2.RectArea( 

142 lat_min=0, 

143 lat_max=2, 

144 lng_min=0, 

145 lng_max=1, 

146 ) 

147 ) 

148 ) 

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

150 

151 res = api.UserSearchV2( 

152 search_pb2.UserSearchReq( 

153 search_in_rectangle=search_pb2.RectArea( 

154 lat_min=0, 

155 lat_max=2, 

156 lng_min=0, 

157 lng_max=1, 

158 ) 

159 ) 

160 ) 

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

162 

163 

164def test_user_filter_complete_profile(db): 

165 """ 

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

167 """ 

168 user_complete_profile, token6 = generate_user(complete_profile=True) 

169 

170 user_incomplete_profile, token7 = generate_user(complete_profile=False) 

171 

172 refresh_materialized_views_rapid(None) 

173 refresh_materialized_views(None) 

174 

175 with search_session(token7) as api: 

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

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

178 

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

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

181 

182 with search_session(token6) as api: 

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

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

185 

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

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

188 

189 

190def test_user_filter_meetup_status(db): 

191 """ 

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

193 """ 

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

195 

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

197 

198 refresh_materialized_views_rapid(None) 

199 refresh_materialized_views(None) 

200 

201 with search_session(token8) as api: 

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

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

204 

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

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

207 

208 with search_session(token9) as api: 

209 res = api.UserSearch( 

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

211 ) 

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

213 

214 res = api.UserSearchV2( 

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

216 ) 

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

218 

219 

220def test_user_filter_language(db): 

221 """ 

222 Test filtering users by language ability. 

223 """ 

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

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

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

227 

228 with session_scope() as session: 

229 session.add( 

230 LanguageAbility( 

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

232 ), 

233 ) 

234 session.add( 

235 LanguageAbility( 

236 user_id=user_with_japanese_conversational.id, 

237 language_code="jpn", 

238 fluency=LanguageFluency.fluent, 

239 ) 

240 ) 

241 session.add( 

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

243 ) 

244 

245 refresh_materialized_views_rapid(None) 

246 refresh_materialized_views(None) 

247 

248 with search_session(token11) as api: 

249 res = api.UserSearch( 

250 search_pb2.UserSearchReq( 

251 language_ability_filter=[ 

252 api_pb2.LanguageAbility( 

253 code="deu", 

254 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT, 

255 ) 

256 ] 

257 ) 

258 ) 

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

260 

261 res = api.UserSearchV2( 

262 search_pb2.UserSearchReq( 

263 language_ability_filter=[ 

264 api_pb2.LanguageAbility( 

265 code="deu", 

266 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT, 

267 ) 

268 ] 

269 ) 

270 ) 

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

272 

273 res = api.UserSearch( 

274 search_pb2.UserSearchReq( 

275 language_ability_filter=[ 

276 api_pb2.LanguageAbility( 

277 code="jpn", 

278 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL, 

279 ) 

280 ] 

281 ) 

282 ) 

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

284 

285 res = api.UserSearchV2( 

286 search_pb2.UserSearchReq( 

287 language_ability_filter=[ 

288 api_pb2.LanguageAbility( 

289 code="jpn", 

290 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL, 

291 ) 

292 ] 

293 ) 

294 ) 

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

296 

297 

298def test_user_filter_strong_verification(db): 

299 user1, token1 = generate_user() 

300 user2, _ = generate_user(strong_verification=True) 

301 user3, _ = generate_user() 

302 user4, _ = generate_user(strong_verification=True) 

303 user5, _ = generate_user(strong_verification=True) 

304 

305 refresh_materialized_views_rapid(None) 

306 refresh_materialized_views(None) 

307 

308 with search_session(token1) as api: 

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

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

311 

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

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

314 

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

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

317 

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

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

320 

321 

322def test_regression_search_only_with_references(db): 

323 user1, token1 = generate_user() 

324 user2, _ = generate_user() 

325 user3, _ = generate_user() 

326 user4, _ = generate_user(delete_user=True) 

327 

328 refresh_materialized_views_rapid(None) 

329 refresh_materialized_views(None) 

330 

331 with session_scope() as session: 

332 # user 2 has references 

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

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

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

336 

337 # user 3 only has reference from a deleted user 

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

339 

340 with search_session(token1) as api: 

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

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

343 

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

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

346 

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

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

349 

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

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

352 

353 

354def test_user_search_exactly_user_ids(db): 

355 """ 

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

357 """ 

358 # Create users with different properties 

359 user1, token1 = generate_user() 

360 user2, _ = generate_user(strong_verification=True) 

361 user3, _ = generate_user(complete_profile=True) 

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

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

364 

365 refresh_materialized_views_rapid(None) 

366 refresh_materialized_views(None) 

367 

368 with search_session(token1) as api: 

369 # Test that exactly_user_ids returns only the specified users 

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

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

372 

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

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

375 

376 # Test that exactly_user_ids ignores other filters 

377 res = api.UserSearch( 

378 search_pb2.UserSearchReq( 

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

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

381 ) 

382 ) 

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

384 

385 res = api.UserSearchV2( 

386 search_pb2.UserSearchReq( 

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

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

389 ) 

390 ) 

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

392 

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

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

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

396 

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

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

399 

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

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

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

403 

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

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

406 

407 

408@pytest.fixture 

409def sample_event_data() -> dict: 

410 """Dummy data for creating events.""" 

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

412 end_time = start_time + timedelta(hours=3) 

413 return { 

414 "title": "Dummy Title", 

415 "content": "Dummy content.", 

416 "photo_key": None, 

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

418 "start_time": Timestamp_from_datetime(start_time), 

419 "end_time": Timestamp_from_datetime(end_time), 

420 "timezone": "UTC", 

421 } 

422 

423 

424@pytest.fixture 

425def create_event(sample_event_data): 

426 """Factory for creating events.""" 

427 

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

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

430 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) 

431 

432 return _create_event 

433 

434 

435@pytest.fixture 

436def sample_community(db) -> int: 

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

438 user, _ = generate_user() 

439 with session_scope() as session: 

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

441 

442 

443def test_EventSearch_no_filters(testing_communities): 

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

445 user, token = generate_user() 

446 with search_session(token) as api: 

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

448 assert len(res.events) > 0 

449 

450 

451def test_event_search_by_query(sample_community, create_event): 

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

453 user, token = generate_user() 

454 

455 with events_session(token) as api: 

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

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

458 create_event(api) 

459 

460 with search_session(token) as api: 

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

462 assert len(res.events) == 2 

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

464 

465 res = api.EventSearch( 

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

467 ) 

468 assert len(res.events) == 1 

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

470 

471 

472def test_event_search_by_time(sample_community, create_event): 

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

474 user, token = generate_user() 

475 

476 with events_session(token) as api: 

477 event1 = create_event( 

478 api, 

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

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

481 ) 

482 event2 = create_event( 

483 api, 

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

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

486 ) 

487 event3 = create_event( 

488 api, 

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

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

491 ) 

492 

493 with search_session(token) as api: 

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

495 assert len(res.events) == 2 

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

497 

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

499 assert len(res.events) == 2 

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

501 

502 res = api.EventSearch( 

503 search_pb2.EventSearchReq( 

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

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

506 ) 

507 ) 

508 assert len(res.events) == 1 

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

510 

511 

512def test_event_search_by_circle(sample_community, create_event): 

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

514 user, token = generate_user() 

515 

516 with events_session(token) as api: 

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

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

519 create_event( 

520 api, 

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

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

523 ) 

524 

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

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

527 create_event( 

528 api, 

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

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

531 ) 

532 

533 with search_session(token) as api: 

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

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

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

537 

538 

539def test_event_search_by_rectangle(sample_community, create_event): 

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

541 user, token = generate_user() 

542 

543 with events_session(token) as api: 

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

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

546 create_event( 

547 api, 

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

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

550 ) 

551 

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

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

554 create_event( 

555 api, 

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

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

558 ) 

559 

560 with search_session(token) as api: 

561 res = api.EventSearch( 

562 search_pb2.EventSearchReq( 

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

564 ) 

565 ) 

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

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

568 

569 

570def test_event_search_pagination(sample_community, create_event): 

571 """Test that EventSearch paginates correctly. 

572 

573 Check that 

574 - <page_size> events are returned, if available 

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

576 - the next page token is correct 

577 """ 

578 user, token = generate_user() 

579 

580 anchor_time = now() 

581 with events_session(token) as api: 

582 for i in range(5): 

583 create_event( 

584 api, 

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

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

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

588 ) 

589 

590 with search_session(token) as api: 

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

592 assert len(res.events) == 4 

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

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

595 

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

597 assert len(res.events) == 1 

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

599 assert res.next_page_token == "" 

600 

601 res = api.EventSearch( 

602 search_pb2.EventSearchReq( 

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

604 ) 

605 ) 

606 assert len(res.events) == 2 

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

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

609 

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

611 assert len(res.events) == 2 

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

613 assert res.next_page_token == "" 

614 

615 

616def test_event_search_pagination_with_page_number(sample_community, create_event): 

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

618 

619 Check that 

620 - <page_size> events are returned, if available 

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

622 - <page_number> is respected 

623 - <total_items> is correct 

624 """ 

625 user, token = generate_user() 

626 

627 anchor_time = now() 

628 with events_session(token) as api: 

629 for i in range(5): 

630 create_event( 

631 api, 

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

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

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

635 ) 

636 

637 with search_session(token) as api: 

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

639 assert len(res.events) == 2 

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

641 assert res.total_items == 5 

642 

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

644 assert len(res.events) == 2 

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

646 assert res.total_items == 5 

647 

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

649 assert len(res.events) == 1 

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

651 assert res.total_items == 5 

652 

653 # Verify no more pages 

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

655 assert not res.events 

656 assert res.total_items == 5 

657 

658 

659def test_event_search_online_status(sample_community, create_event): 

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

661 user, token = generate_user() 

662 

663 with events_session(token) as api: 

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

665 

666 create_event( 

667 api, 

668 title="Online event", 

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

670 parent_community_id=sample_community, 

671 offline_information=events_pb2.OfflineEventInformation(), 

672 ) 

673 

674 with search_session(token) as api: 

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

676 assert len(res.events) == 2 

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

678 

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