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

1from datetime import date, timedelta 

2 

3import grpc 

4import pytest 

5from google.protobuf import empty_pb2, wrappers_pb2 

6from psycopg.types.range import TimestamptzRange 

7from sqlalchemy import select, update 

8 

9from couchers.db import session_scope 

10from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid 

11from couchers.models import ( 

12 EventOccurrence, 

13 HostingStatus, 

14 LanguageAbility, 

15 LanguageFluency, 

16 MeetupStatus, 

17 PassportSex, 

18 StrongVerificationAttempt, 

19 StrongVerificationAttemptStatus, 

20 User, 

21 UserRankingOffset, 

22) 

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

24from couchers.servicers.search import USER_SEARCH_NOISE_PERIOD_DAYS 

25from couchers.utils import Timestamp_from_datetime, create_coordinate, now, today 

26from tests.fixtures.api import events as events_api 

27from tests.fixtures.db import generate_user 

28from tests.fixtures.misc import Moderator 

29from tests.fixtures.sessions import communities_session, search_session 

30from tests.test_communities import create_community, testing_communities # noqa 

31from tests.test_references import create_friend_reference 

32from tests.utils import local_date_time 

33 

34 

35class TestSearchInCommunities: 

36 """The tests that search the whole community tree, grouped so they share one copy of it.""" 

37 

38 @staticmethod 

39 def test_Search(testing_communities): 

40 user, token = generate_user() 

41 with search_session(token) as api: 

42 res = api.Search( 

43 search_pb2.SearchReq( 

44 query="Country 1, Region 1", 

45 include_users=True, 

46 include_communities=True, 

47 include_groups=True, 

48 include_places=True, 

49 include_guides=True, 

50 ) 

51 ) 

52 res = api.Search( 

53 search_pb2.SearchReq( 

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

55 title_only=True, 

56 include_users=True, 

57 include_communities=True, 

58 include_groups=True, 

59 include_places=True, 

60 include_guides=True, 

61 ) 

62 ) 

63 

64 @staticmethod 

65 def test_UserSearch(testing_communities): 

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

67 user, token = generate_user() 

68 

69 refresh_materialized_views_rapid(empty_pb2.Empty()) 

70 refresh_materialized_views(empty_pb2.Empty()) 

71 

72 with search_session(token) as api: 

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

74 assert len(res.results) > 0 

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

76 

77 @staticmethod 

78 def test_EventSearch_no_filters(testing_communities): 

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

80 user, token = generate_user() 

81 with search_session(token) as api: 

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

83 assert len(res.events) > 0 

84 

85 

86def test_regression_search_in_area(db): 

87 """ 

88 Makes sure search_in_area works. 

89 

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

91 """ 

92 

93 # outside 

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

95 # outside 

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

97 # inside 

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

99 # inside 

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

101 # outside 

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

103 

104 refresh_materialized_views_rapid(empty_pb2.Empty()) 

105 refresh_materialized_views(empty_pb2.Empty()) 

106 

107 with search_session(token5) as api: 

108 res = api.UserSearchV2( 

109 search_pb2.UserSearchReq( 

110 search_in_area=search_pb2.Area( 

111 lat=0, 

112 lng=0, 

113 radius=100000, 

114 ) 

115 ) 

116 ) 

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

118 

119 

120def test_user_search_in_rectangle(db): 

121 """ 

122 Makes sure search_in_rectangle works as expected. 

123 """ 

124 

125 # outside 

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

127 # outside 

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

129 # inside 

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

131 # inside 

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

133 # outside (not fully inside) 

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

135 # outside 

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

137 # outside 

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

139 

140 refresh_materialized_views_rapid(empty_pb2.Empty()) 

141 refresh_materialized_views(empty_pb2.Empty()) 

142 

143 with search_session(token5) as api: 

144 res = api.UserSearchV2( 

145 search_pb2.UserSearchReq( 

146 search_in_rectangle=search_pb2.RectArea( 

147 lat_min=0, 

148 lat_max=2, 

149 lng_min=0, 

150 lng_max=1, 

151 ) 

152 ) 

153 ) 

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

155 

156 

157def test_user_filter_complete_profile(db): 

158 """ 

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

160 """ 

161 user_complete_profile, token6 = generate_user(complete_profile=True) 

162 

163 user_incomplete_profile, token7 = generate_user(complete_profile=False) 

164 

165 refresh_materialized_views_rapid(empty_pb2.Empty()) 

166 refresh_materialized_views(empty_pb2.Empty()) 

167 

168 with search_session(token7) as api: 

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

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

171 

172 with search_session(token6) as api: 

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

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

175 

176 

177def test_user_filter_meetup_status(db): 

178 """ 

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

180 """ 

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

182 

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

184 

185 refresh_materialized_views_rapid(empty_pb2.Empty()) 

186 refresh_materialized_views(empty_pb2.Empty()) 

187 

188 with search_session(token8) as api: 

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

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

191 

192 with search_session(token9) as api: 

193 res = api.UserSearchV2( 

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

195 ) 

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

197 

198 

199def test_user_filter_language(db): 

200 """ 

201 Test filtering users by language ability. 

202 """ 

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

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

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

206 

207 with session_scope() as session: 

208 session.add( 

209 LanguageAbility( 

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

211 ), 

212 ) 

213 session.add( 

214 LanguageAbility( 

215 user_id=user_with_japanese_conversational.id, 

216 language_code="jpn", 

217 fluency=LanguageFluency.fluent, 

218 ) 

219 ) 

220 session.add( 

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

222 ) 

223 

224 refresh_materialized_views_rapid(empty_pb2.Empty()) 

225 refresh_materialized_views(empty_pb2.Empty()) 

226 

227 with search_session(token11) as api: 

228 res = api.UserSearchV2( 

229 search_pb2.UserSearchReq( 

230 language_ability_filter=[ 

231 api_pb2.LanguageAbility( 

232 code="deu", 

233 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT, 

234 ) 

235 ] 

236 ) 

237 ) 

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

239 

240 res = api.UserSearchV2( 

241 search_pb2.UserSearchReq( 

242 language_ability_filter=[ 

243 api_pb2.LanguageAbility( 

244 code="jpn", 

245 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL, 

246 ) 

247 ] 

248 ) 

249 ) 

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

251 

252 

253def test_user_filter_strong_verification(db): 

254 user1, token1 = generate_user() 

255 user2, _ = generate_user(strong_verification=True) 

256 user3, _ = generate_user() 

257 user4, _ = generate_user(strong_verification=True) 

258 user5, _ = generate_user(strong_verification=True) 

259 

260 refresh_materialized_views_rapid(empty_pb2.Empty()) 

261 refresh_materialized_views(empty_pb2.Empty()) 

262 

263 with search_session(token1) as api: 

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

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

266 

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

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

269 

270 

271def test_user_search_no_duplicates_from_multiple_matching_rows(db): 

272 """ 

273 A user matching several requested languages, or with several valid strong verifications, is returned once. 

274 """ 

275 languages = [("deu", LanguageFluency.fluent), ("jpn", LanguageFluency.fluent)] 

276 user1, token1 = generate_user() 

277 user2, _ = generate_user(strong_verification=True, language_abilities=languages) 

278 user3, _ = generate_user(strong_verification=True, language_abilities=languages) 

279 

280 with session_scope() as session: 

281 session.add( 

282 StrongVerificationAttempt( 

283 verification_attempt_token=f"second_verification_attempt_token_{user2.id}", 

284 user_id=user2.id, 

285 status=StrongVerificationAttemptStatus.succeeded, 

286 has_full_data=True, 

287 passport_encrypted_data=b"not real", 

288 passport_date_of_birth=user2.birthdate, 

289 passport_sex=PassportSex.unspecified, 

290 has_minimal_data=True, 

291 passport_expiry_date=date.today() + timedelta(days=20), 

292 passport_nationality="UTO", 

293 passport_last_three_document_chars="xyz", 

294 iris_token=f"second_iris_token_{user2.id}", 

295 iris_session_id=1000 + user2.id, 

296 ) 

297 ) 

298 

299 refresh_materialized_views_rapid(empty_pb2.Empty()) 

300 refresh_materialized_views(empty_pb2.Empty()) 

301 

302 language_filter = [ 

303 api_pb2.LanguageAbility(code="deu", fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL), 

304 api_pb2.LanguageAbility(code="jpn", fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL), 

305 ] 

306 with search_session(token1) as api: 

307 for req in [ 

308 search_pb2.UserSearchReq(language_ability_filter=language_filter), 

309 search_pb2.UserSearchReq(only_with_strong_verification=True), 

310 ]: 

311 res = api.UserSearchV2(req) 

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

313 assert res.total_items == 2 

314 

315 

316def test_regression_search_only_with_references(db): 

317 user1, token1 = generate_user() 

318 user2, _ = generate_user() 

319 user3, _ = generate_user() 

320 user4, _ = generate_user(delete_user=True) 

321 

322 refresh_materialized_views_rapid(empty_pb2.Empty()) 

323 refresh_materialized_views(empty_pb2.Empty()) 

324 

325 with session_scope() as session: 

326 # user 2 has references 

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

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

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

330 

331 # user 3 only has reference from a deleted user 

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

333 

334 with search_session(token1) as api: 

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

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

337 

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

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

340 

341 

342def test_user_search_exactly_user_ids(db): 

343 """ 

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

345 """ 

346 # Create users with different properties 

347 user1, token1 = generate_user() 

348 user2, _ = generate_user(strong_verification=True) 

349 user3, _ = generate_user(complete_profile=True) 

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

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

352 

353 refresh_materialized_views_rapid(empty_pb2.Empty()) 

354 refresh_materialized_views(empty_pb2.Empty()) 

355 

356 with search_session(token1) as api: 

357 # Test that exactly_user_ids returns only the specified users 

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

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

360 

361 # Test that exactly_user_ids ignores other filters 

362 res = api.UserSearchV2( 

363 search_pb2.UserSearchReq( 

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

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

366 ) 

367 ) 

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

369 

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

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

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

373 

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

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

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

377 

378 

379def test_user_search_pagination_with_tied_scores(db, flags): 

380 """ 

381 Paging through tied recommendation scores, including a tie block larger than the page size, returns every user 

382 exactly once in a stable order. 

383 """ 

384 flags.set_float("user_search_noise_scale", 0.0) 

385 users = [generate_user()[0] for _ in range(7)] 

386 token = generate_user()[1] 

387 scores = [10.0, 5.0, 5.0, 5.0, 5.0, 5.0, -1.0] 

388 

389 with session_scope() as session: 

390 for user, score in zip(users, scores): 

391 session.execute(update(User).where(User.id == user.id).values(recommendation_score=score)) 

392 session.execute(update(User).where(User.id.not_in([u.id for u in users])).values(recommendation_score=0.0)) 

393 

394 refresh_materialized_views_rapid(empty_pb2.Empty()) 

395 refresh_materialized_views(empty_pb2.Empty()) 

396 

397 with session_scope() as session: 

398 expected = list( 

399 session.execute(select(User.id).order_by(User.recommendation_score.desc(), User.id.desc())).scalars() 

400 ) 

401 

402 with search_session(token) as api: 

403 seen = [] 

404 page_token = "" 

405 while True: 

406 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=2, page_token=page_token)) 

407 seen += [result.user_id for result in res.results] 

408 assert len(seen) <= len(expected) 

409 page_token = res.next_page_token 

410 if not page_token: 

411 break 

412 

413 assert seen == expected 

414 

415 

416def _user_search_order(token: str, page_size: int = 100) -> list[int]: 

417 order: list[int] = [] 

418 page_token = "" 

419 with search_session(token) as api: 

420 while True: 

421 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=page_size, page_token=page_token)) 

422 order += [result.user_id for result in res.results] 

423 page_token = res.next_page_token 

424 if not page_token: 

425 return order 

426 

427 

428def test_user_search_noise_is_per_viewer_and_stable_within_period(db, frozen_timewarp): 

429 viewer1, token1 = generate_user() 

430 _, token2 = generate_user() 

431 candidate_ids = {generate_user()[0].id for _ in range(20)} 

432 

433 with session_scope() as session: 

434 session.execute(update(User).values(recommendation_score=0)) 

435 

436 refresh_materialized_views_rapid(empty_pb2.Empty()) 

437 

438 order1 = _user_search_order(token1) 

439 assert candidate_ids <= set(order1) 

440 assert _user_search_order(token1) == order1 

441 assert _user_search_order(token2) != order1 

442 

443 # step a day at a time, which also keeps the short-lived session alive, until this viewer's period rolls over 

444 days_left = USER_SEARCH_NOISE_PERIOD_DAYS - (today().toordinal() + viewer1.id) % USER_SEARCH_NOISE_PERIOD_DAYS 

445 for _ in range(days_left - 1): 

446 frozen_timewarp.advance(timedelta(days=1)) 

447 assert _user_search_order(token1) == order1 

448 frozen_timewarp.advance(timedelta(days=1)) 

449 assert _user_search_order(token1) != order1 

450 

451 

452def test_user_search_pagination_with_noise(db): 

453 _, token = generate_user() 

454 for _ in range(10): 

455 generate_user() 

456 

457 with session_scope() as session: 

458 session.execute(update(User).values(recommendation_score=0)) 

459 

460 refresh_materialized_views_rapid(empty_pb2.Empty()) 

461 

462 full_order = _user_search_order(token) 

463 assert len(full_order) == 11 

464 assert _user_search_order(token, page_size=3) == full_order 

465 

466 

467def test_user_search_noise_scale_flag(db, flags): 

468 _, token = generate_user() 

469 user_ids = [generate_user()[0].id for _ in range(10)] 

470 

471 with session_scope() as session: 

472 session.execute(update(User).values(recommendation_score=-User.id)) 

473 

474 refresh_materialized_views_rapid(empty_pb2.Empty()) 

475 

476 def order() -> list[int]: 

477 return [user_id for user_id in _user_search_order(token) if user_id in user_ids] 

478 

479 assert sorted(order()) == user_ids 

480 assert order() != user_ids 

481 

482 flags.set_float("user_search_noise_scale", 0.0) 

483 assert order() == user_ids 

484 

485 

486def test_user_search_ranking_offsets_variant(db, flags): 

487 flags.set_float("user_search_noise_scale", 0.0) 

488 _, token = generate_user() 

489 user_ids = [generate_user()[0].id for _ in range(5)] 

490 

491 with session_scope() as session: 

492 session.execute(update(User).values(recommendation_score=-User.id)) 

493 # in variant "a", offsets reverse the order; variant "b" bumps only the last user to the top 

494 for i, user_id in enumerate(user_ids): 

495 session.add(UserRankingOffset(variant="a", user_id=user_id, offset=2.0 * i)) 

496 session.add(UserRankingOffset(variant="b", user_id=user_ids[-1], offset=100.0)) 

497 

498 refresh_materialized_views_rapid(empty_pb2.Empty()) 

499 

500 def order() -> list[int]: 

501 return [user_id for user_id in _user_search_order(token, page_size=2) if user_id in user_ids] 

502 

503 assert order() == user_ids 

504 

505 flags.set_string("search_ranking_offsets_variant", "a") 

506 assert order() == user_ids[::-1] 

507 

508 flags.set_string("search_ranking_offsets_variant", "b") 

509 assert order() == user_ids[-1:] + user_ids[:-1] 

510 

511 flags.set_string("search_ranking_offsets_variant", "nonexistent") 

512 assert order() == user_ids 

513 

514 

515@pytest.fixture 

516def sample_community(db) -> int: 

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

518 user, _ = generate_user() 

519 with session_scope() as session: 

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

521 

522 

523def test_event_search_by_query(sample_community): 

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

525 user, token = generate_user() 

526 

527 event1 = events_api.create_event(token=token, title="Lorem Ipsum") 

528 event2 = events_api.create_event(token=token, content="Lorem Ipsum") 

529 events_api.create_event(token=token) 

530 

531 with search_session(token) as api: 

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

533 assert len(res.events) == 2 

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

535 

536 res = api.EventSearch( 

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

538 ) 

539 assert len(res.events) == 1 

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

541 

542 

543def test_event_search_by_time(sample_community): 

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

545 user, token = generate_user() 

546 

547 event1 = events_api.create_event( 

548 token=token, 

549 start_date_time=local_date_time(now() + timedelta(hours=1)), 

550 end_date_time=local_date_time(now() + timedelta(hours=2)), 

551 ) 

552 event2 = events_api.create_event( 

553 token=token, 

554 start_date_time=local_date_time(now() + timedelta(hours=4)), 

555 end_date_time=local_date_time(now() + timedelta(hours=5)), 

556 ) 

557 event3 = events_api.create_event( 

558 token=token, 

559 start_date_time=local_date_time(now() + timedelta(hours=7)), 

560 end_date_time=local_date_time(now() + timedelta(hours=8)), 

561 ) 

562 

563 with search_session(token) as api: 

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

565 assert len(res.events) == 2 

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

567 

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

569 assert len(res.events) == 2 

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

571 

572 res = api.EventSearch( 

573 search_pb2.EventSearchReq( 

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

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

576 ) 

577 ) 

578 assert len(res.events) == 1 

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

580 

581 

582def test_event_search_by_circle(sample_community): 

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

584 user, token = generate_user() 

585 

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

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

588 events_api.create_event( 

589 token=token, 

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

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

592 ) 

593 

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

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

596 events_api.create_event( 

597 token=token, 

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

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

600 ) 

601 

602 with search_session(token) as api: 

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

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

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

606 

607 

608def test_event_search_by_rectangle(sample_community): 

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

610 user, token = generate_user() 

611 

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

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

614 events_api.create_event( 

615 token=token, 

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

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

618 ) 

619 

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

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

622 events_api.create_event( 

623 token=token, 

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

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

626 ) 

627 

628 with search_session(token) as api: 

629 res = api.EventSearch( 

630 search_pb2.EventSearchReq( 

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

632 ) 

633 ) 

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

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

636 

637 

638def test_event_search_pagination(sample_community): 

639 """Test that EventSearch paginates correctly. 

640 

641 Check that 

642 - <page_size> events are returned, if available 

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

644 - the next page token continues where the previous page left off 

645 """ 

646 user, token = generate_user() 

647 

648 anchor_time = now().replace(second=0, microsecond=0) # Events are created at minute granularity 

649 for i in range(5): 

650 events_api.create_event( 

651 token=token, 

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

653 start_date_time=local_date_time(anchor_time + timedelta(hours=i + 1)), 

654 end_date_time=local_date_time(anchor_time + timedelta(hours=i + 1, minutes=30)), 

655 ) 

656 

657 with search_session(token) as api: 

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

659 assert len(res.events) == 4 

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

661 assert res.next_page_token 

662 

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

664 assert len(res.events) == 1 

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

666 assert res.next_page_token == "" 

667 

668 # move all the events into the past to test past pagination 

669 with session_scope() as session: 

670 for occurrence in session.execute(select(EventOccurrence)).scalars().all(): 

671 occurrence.during = TimestamptzRange( 

672 occurrence.start_time - timedelta(days=30), occurrence.end_time - timedelta(days=30) 

673 ) 

674 

675 with search_session(token) as api: 

676 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2)) 

677 assert [event.title for event in res.events] == ["Event 5", "Event 4"] 

678 assert res.next_page_token 

679 

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