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.models import EventOccurrence, HostingStatus, LanguageAbility, LanguageFluency, MeetupStatus 

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

9from proto import api_pb2, communities_pb2, events_pb2, search_pb2 

10from tests.test_communities import create_community, testing_communities # noqa 

11from tests.test_fixtures import ( # noqa 

12 communities_session, 

13 db, 

14 events_session, 

15 generate_user, 

16 search_session, 

17 session_scope, 

18 testconfig, 

19) 

20 

21 

22@pytest.fixture(autouse=True) 

23def _(testconfig): 

24 pass 

25 

26 

27def test_Search(testing_communities): 

28 user, token = generate_user() 

29 with search_session(token) as api: 

30 res = api.Search( 

31 search_pb2.SearchReq( 

32 query="Country 1, Region 1", 

33 include_users=True, 

34 include_communities=True, 

35 include_groups=True, 

36 include_places=True, 

37 include_guides=True, 

38 ) 

39 ) 

40 res = api.Search( 

41 search_pb2.SearchReq( 

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

43 title_only=True, 

44 include_users=True, 

45 include_communities=True, 

46 include_groups=True, 

47 include_places=True, 

48 include_guides=True, 

49 ) 

50 ) 

51 

52 

53def test_UserSearch(testing_communities): 

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

55 user, token = generate_user() 

56 with search_session(token) as api: 

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

58 assert len(res.results) > 0 

59 

60 

61def test_regression_search_in_area(db): 

62 """ 

63 Makes sure search_in_area works. 

64 

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

66 """ 

67 

68 # outside 

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

70 # outside 

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

72 # inside 

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

74 # inside 

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

76 # outside 

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

78 

79 with search_session(token5) as api: 

80 res = api.UserSearch( 

81 search_pb2.UserSearchReq( 

82 search_in_area=search_pb2.Area( 

83 lat=0, 

84 lng=0, 

85 radius=100000, 

86 ) 

87 ) 

88 ) 

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

90 

91 

92def test_user_search_in_rectangle(db): 

93 """ 

94 Makes sure search_in_rectangle works as expected. 

95 """ 

96 

97 # outside 

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

99 # outside 

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

101 # inside 

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

103 # inside 

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

105 # outside (not fully inside) 

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

107 # outside 

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

109 # outside 

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

111 

112 with search_session(token5) as api: 

113 res = api.UserSearch( 

114 search_pb2.UserSearchReq( 

115 search_in_rectangle=search_pb2.RectArea( 

116 lat_min=0, 

117 lat_max=2, 

118 lng_min=0, 

119 lng_max=1, 

120 ) 

121 ) 

122 ) 

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

124 

125 

126def test_user_filter_complete_profile(db): 

127 """ 

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

129 """ 

130 user_complete_profile, token6 = generate_user(complete_profile=True) 

131 

132 user_incomplete_profile, token7 = generate_user(complete_profile=False) 

133 

134 with search_session(token7) as api: 

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

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

137 

138 with search_session(token6) as api: 

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

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

141 

142 

143def test_user_filter_meetup_status(db): 

144 """ 

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

146 """ 

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

148 

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

150 

151 with search_session(token8) as api: 

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

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

154 

155 with search_session(token9) as api: 

156 res = api.UserSearch( 

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

158 ) 

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

160 

161 

162def test_user_filter_language(db): 

163 """ 

164 Test filtering users by language ability. 

165 """ 

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

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

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

169 

170 with session_scope() as session: 

171 session.add( 

172 LanguageAbility( 

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

174 ), 

175 ) 

176 session.add( 

177 LanguageAbility( 

178 user_id=user_with_japanese_conversational.id, 

179 language_code="jpn", 

180 fluency=LanguageFluency.fluent, 

181 ) 

182 ) 

183 session.add( 

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

185 ) 

186 

187 with search_session(token11) as api: 

188 res = api.UserSearch( 

189 search_pb2.UserSearchReq( 

190 language_ability_filter=[ 

191 api_pb2.LanguageAbility( 

192 code="deu", 

193 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT, 

194 ) 

195 ] 

196 ) 

197 ) 

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

199 

200 res = api.UserSearch( 

201 search_pb2.UserSearchReq( 

202 language_ability_filter=[ 

203 api_pb2.LanguageAbility( 

204 code="jpn", 

205 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL, 

206 ) 

207 ] 

208 ) 

209 ) 

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

211 

212 

213@pytest.fixture 

214def sample_event_data() -> dict: 

215 """Dummy data for creating events.""" 

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

217 end_time = start_time + timedelta(hours=3) 

218 return { 

219 "title": "Dummy Title", 

220 "content": "Dummy content.", 

221 "photo_key": None, 

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

223 "start_time": Timestamp_from_datetime(start_time), 

224 "end_time": Timestamp_from_datetime(end_time), 

225 "timezone": "UTC", 

226 } 

227 

228 

229@pytest.fixture 

230def create_event(sample_event_data): 

231 """Factory for creating events.""" 

232 

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

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

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

236 

237 return _create_event 

238 

239 

240@pytest.fixture 

241def sample_community(db) -> int: 

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

243 user, _ = generate_user() 

244 with session_scope() as session: 

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

246 

247 

248def test_EventSearch_no_filters(testing_communities): 

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

250 user, token = generate_user() 

251 with search_session(token) as api: 

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

253 assert len(res.events) > 0 

254 

255 

256def test_event_search_by_query(sample_community, create_event): 

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

258 user, token = generate_user() 

259 

260 with events_session(token) as api: 

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

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

263 create_event(api) 

264 

265 with search_session(token) as api: 

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

267 assert len(res.events) == 2 

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

269 

270 res = api.EventSearch( 

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

272 ) 

273 assert len(res.events) == 1 

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

275 

276 

277def test_event_search_by_time(sample_community, create_event): 

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

279 user, token = generate_user() 

280 

281 with events_session(token) as api: 

282 event1 = create_event( 

283 api, 

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

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

286 ) 

287 event2 = create_event( 

288 api, 

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

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

291 ) 

292 event3 = create_event( 

293 api, 

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

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

296 ) 

297 

298 with search_session(token) as api: 

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

300 assert len(res.events) == 2 

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

302 

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

304 assert len(res.events) == 2 

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

306 

307 res = api.EventSearch( 

308 search_pb2.EventSearchReq( 

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

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

311 ) 

312 ) 

313 assert len(res.events) == 1 

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

315 

316 

317def test_event_search_by_circle(sample_community, create_event): 

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

319 user, token = generate_user() 

320 

321 with events_session(token) as api: 

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

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

324 create_event( 

325 api, 

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

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

328 ) 

329 

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

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

332 create_event( 

333 api, 

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

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

336 ) 

337 

338 with search_session(token) as api: 

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

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

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

342 

343 

344def test_event_search_by_rectangle(sample_community, create_event): 

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

346 user, token = generate_user() 

347 

348 with events_session(token) as api: 

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

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

351 create_event( 

352 api, 

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

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

355 ) 

356 

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

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

359 create_event( 

360 api, 

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

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

363 ) 

364 

365 with search_session(token) as api: 

366 res = api.EventSearch( 

367 search_pb2.EventSearchReq( 

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

369 ) 

370 ) 

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

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

373 

374 

375def test_event_search_pagination(sample_community, create_event): 

376 """Test that EventSearch paginates correctly. 

377 

378 Check that 

379 - <page_size> events are returned, if available 

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

381 - the next page token is correct 

382 """ 

383 user, token = generate_user() 

384 

385 anchor_time = now() 

386 with events_session(token) as api: 

387 for i in range(5): 

388 create_event( 

389 api, 

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

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

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

393 ) 

394 

395 with search_session(token) as api: 

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

397 assert len(res.events) == 4 

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

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

400 

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

402 assert len(res.events) == 1 

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

404 assert res.next_page_token == "" 

405 

406 res = api.EventSearch( 

407 search_pb2.EventSearchReq( 

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

409 ) 

410 ) 

411 assert len(res.events) == 2 

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

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

414 

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

416 assert len(res.events) == 2 

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

418 assert res.next_page_token == "" 

419 

420 

421def test_event_search_pagination_with_page_number(sample_community, create_event): 

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

423 

424 Check that 

425 - <page_size> events are returned, if available 

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

427 - <page_number> is respected 

428 - <total_items> is correct 

429 """ 

430 user, token = generate_user() 

431 

432 anchor_time = now() 

433 with events_session(token) as api: 

434 for i in range(5): 

435 create_event( 

436 api, 

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

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

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

440 ) 

441 

442 with search_session(token) as api: 

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

444 assert len(res.events) == 2 

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

446 assert res.total_items == 5 

447 

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

449 assert len(res.events) == 2 

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

451 assert res.total_items == 5 

452 

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

454 assert len(res.events) == 1 

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

456 assert res.total_items == 5 

457 

458 # Verify no more pages 

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

460 assert not res.events 

461 assert res.total_items == 5 

462 

463 

464def test_event_search_online_status(sample_community, create_event): 

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

466 user, token = generate_user() 

467 

468 with events_session(token) as api: 

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

470 

471 create_event( 

472 api, 

473 title="Online event", 

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

475 parent_community_id=sample_community, 

476 offline_information=events_pb2.OfflineEventInformation(), 

477 ) 

478 

479 with search_session(token) as api: 

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

481 assert len(res.events) == 2 

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

483 

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

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

486 

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

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

489 

490 

491def test_event_search_filter_subscription_attendance_organizing_my_communities(sample_community, create_event): 

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

493 returns all events. 

494 """ 

495 _, token = generate_user() 

496 other_user, other_token = generate_user() 

497 

498 with communities_session(token) as api: 

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

500 

501 with session_scope() as session: 

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

503 

504 with events_session(other_token) as api: 

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

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

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

508 create_event( 

509 api, 

510 title="Other community event", 

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

512 ) 

513 

514 with events_session(token) as api: 

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

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

517 api.SetEventAttendance( 

518 events_pb2.SetEventAttendanceReq( 

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

520 ) 

521 ) 

522 

523 with search_session(token) as api: 

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

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

526 "Subscribed event", 

527 "Attending event", 

528 "Community event", 

529 "Other community event", 

530 "Organized event", 

531 } 

532 

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

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

535 

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

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

538 

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

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

541 

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

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

544 "Subscribed event", 

545 "Attending event", 

546 "Community event", 

547 "Organized event", 

548 } 

549 

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

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