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, 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 testconfig, 

18) 

19 

20 

21@pytest.fixture(autouse=True) 

22def _(testconfig): 

23 pass 

24 

25 

26def test_Search(testing_communities): 

27 user, token = generate_user() 

28 with search_session(token) as api: 

29 res = api.Search( 

30 search_pb2.SearchReq( 

31 query="Country 1, Region 1", 

32 include_users=True, 

33 include_communities=True, 

34 include_groups=True, 

35 include_places=True, 

36 include_guides=True, 

37 ) 

38 ) 

39 res = api.Search( 

40 search_pb2.SearchReq( 

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

42 title_only=True, 

43 include_users=True, 

44 include_communities=True, 

45 include_groups=True, 

46 include_places=True, 

47 include_guides=True, 

48 ) 

49 ) 

50 

51 

52def test_UserSearch(testing_communities): 

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

54 user, token = generate_user() 

55 with search_session(token) as api: 

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

57 assert len(res.results) > 0 

58 

59 

60def test_regression_search_in_area(db): 

61 """ 

62 Makes sure search_in_area works. 

63 

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

65 """ 

66 

67 # outside 

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

69 # outside 

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

71 # inside 

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

73 # inside 

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

75 # outside 

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

77 

78 with search_session(token5) as api: 

79 res = api.UserSearch( 

80 search_pb2.UserSearchReq( 

81 search_in_area=search_pb2.Area( 

82 lat=0, 

83 lng=0, 

84 radius=100000, 

85 ) 

86 ) 

87 ) 

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

89 

90 

91def test_user_search_in_rectangle(db): 

92 """ 

93 Makes sure search_in_rectangle works as expected. 

94 """ 

95 

96 # outside 

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

98 # outside 

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

100 # inside 

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

102 # inside 

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

104 # outside (not fully inside) 

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

106 # outside 

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

108 # outside 

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

110 

111 with search_session(token5) as api: 

112 res = api.UserSearch( 

113 search_pb2.UserSearchReq( 

114 search_in_rectangle=search_pb2.RectArea( 

115 lat_min=0, 

116 lat_max=2, 

117 lng_min=0, 

118 lng_max=1, 

119 ) 

120 ) 

121 ) 

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

123 

124 

125def test_user_filter_complete_profile(db): 

126 """ 

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

128 """ 

129 user_complete_profile, token6 = generate_user(complete_profile=True) 

130 

131 user_incomplete_profile, token7 = generate_user(complete_profile=False) 

132 

133 with search_session(token7) as api: 

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

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

136 

137 with search_session(token6) as api: 

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

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

140 

141 

142def test_user_filter_meetup_status(db): 

143 """ 

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

145 """ 

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

147 

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

149 

150 with search_session(token8) as api: 

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

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

153 

154 with search_session(token9) as api: 

155 res = api.UserSearch( 

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

157 ) 

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

159 

160 

161@pytest.fixture 

162def sample_event_data() -> dict: 

163 """Dummy data for creating events.""" 

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

165 end_time = start_time + timedelta(hours=3) 

166 return { 

167 "title": "Dummy Title", 

168 "content": "Dummy content.", 

169 "photo_key": None, 

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

171 "start_time": Timestamp_from_datetime(start_time), 

172 "end_time": Timestamp_from_datetime(end_time), 

173 "timezone": "UTC", 

174 } 

175 

176 

177@pytest.fixture 

178def create_event(sample_event_data): 

179 """Factory for creating events.""" 

180 

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

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

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

184 

185 return _create_event 

186 

187 

188@pytest.fixture 

189def sample_community(db) -> int: 

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

191 user, _ = generate_user() 

192 with session_scope() as session: 

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

194 

195 

196def test_EventSearch_no_filters(testing_communities): 

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

198 user, token = generate_user() 

199 with search_session(token) as api: 

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

201 assert len(res.events) > 0 

202 

203 

204def test_event_search_by_query(sample_community, create_event): 

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

206 user, token = generate_user() 

207 

208 with events_session(token) as api: 

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

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

211 create_event(api) 

212 

213 with search_session(token) as api: 

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

215 assert len(res.events) == 2 

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

217 

218 res = api.EventSearch( 

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

220 ) 

221 assert len(res.events) == 1 

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

223 

224 

225def test_event_search_by_time(sample_community, create_event): 

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

227 user, token = generate_user() 

228 

229 with events_session(token) as api: 

230 event1 = create_event( 

231 api, 

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

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

234 ) 

235 event2 = create_event( 

236 api, 

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

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

239 ) 

240 event3 = create_event( 

241 api, 

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

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

244 ) 

245 

246 with search_session(token) as api: 

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

248 assert len(res.events) == 2 

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

250 

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

252 assert len(res.events) == 2 

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

254 

255 res = api.EventSearch( 

256 search_pb2.EventSearchReq( 

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

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

259 ) 

260 ) 

261 assert len(res.events) == 1 

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

263 

264 

265def test_event_search_by_circle(sample_community, create_event): 

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

267 user, token = generate_user() 

268 

269 with events_session(token) as api: 

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

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

272 create_event( 

273 api, 

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

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

276 ) 

277 

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

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

280 create_event( 

281 api, 

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

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

284 ) 

285 

286 with search_session(token) as api: 

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

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

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

290 

291 

292def test_event_search_by_rectangle(sample_community, create_event): 

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

294 user, token = generate_user() 

295 

296 with events_session(token) as api: 

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

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

299 create_event( 

300 api, 

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

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

303 ) 

304 

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

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

307 create_event( 

308 api, 

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

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

311 ) 

312 

313 with search_session(token) as api: 

314 res = api.EventSearch( 

315 search_pb2.EventSearchReq( 

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

317 ) 

318 ) 

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

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

321 

322 

323def test_event_search_pagination(sample_community, create_event): 

324 """Test that EventSearch paginates correctly. 

325 

326 Check that 

327 - <page_size> events are returned, if available 

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

329 - the next page token is correct 

330 """ 

331 user, token = generate_user() 

332 

333 anchor_time = now() 

334 with events_session(token) as api: 

335 for i in range(5): 

336 create_event( 

337 api, 

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

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

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

341 ) 

342 

343 with search_session(token) as api: 

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

345 assert len(res.events) == 4 

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

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

348 

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

350 assert len(res.events) == 1 

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

352 assert res.next_page_token == "" 

353 

354 res = api.EventSearch( 

355 search_pb2.EventSearchReq( 

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

357 ) 

358 ) 

359 assert len(res.events) == 2 

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

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

362 

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

364 assert len(res.events) == 2 

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

366 assert res.next_page_token == "" 

367 

368 

369def test_event_search_pagination_with_page_number(sample_community, create_event): 

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

371 

372 Check that 

373 - <page_size> events are returned, if available 

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

375 - <page_number> is respected 

376 - <total_items> is correct 

377 """ 

378 user, token = generate_user() 

379 

380 anchor_time = now() 

381 with events_session(token) as api: 

382 for i in range(5): 

383 create_event( 

384 api, 

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

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

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

388 ) 

389 

390 with search_session(token) as api: 

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

392 assert len(res.events) == 2 

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

394 assert res.total_items == 5 

395 

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

397 assert len(res.events) == 2 

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

399 assert res.total_items == 5 

400 

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

402 assert len(res.events) == 1 

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

404 assert res.total_items == 5 

405 

406 # Verify no more pages 

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

408 assert not res.events 

409 assert res.total_items == 5 

410 

411 

412def test_event_search_online_status(sample_community, create_event): 

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

414 user, token = generate_user() 

415 

416 with events_session(token) as api: 

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

418 

419 create_event( 

420 api, 

421 title="Online event", 

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

423 parent_community_id=sample_community, 

424 offline_information=events_pb2.OfflineEventInformation(), 

425 ) 

426 

427 with search_session(token) as api: 

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

429 assert len(res.events) == 2 

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

431 

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

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

434 

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

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

437 

438 

439def test_event_search_filter_subscription_attendance_organizing_my_communities(sample_community, create_event): 

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

441 returns all events. 

442 """ 

443 _, token = generate_user() 

444 other_user, other_token = generate_user() 

445 

446 with communities_session(token) as api: 

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

448 

449 with session_scope() as session: 

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

451 

452 with events_session(other_token) as api: 

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

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

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

456 create_event( 

457 api, 

458 title="Other community event", 

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

460 ) 

461 

462 with events_session(token) as api: 

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

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

465 api.SetEventAttendance( 

466 events_pb2.SetEventAttendanceReq( 

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

468 ) 

469 ) 

470 

471 with search_session(token) as api: 

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

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

474 "Subscribed event", 

475 "Attending event", 

476 "Community event", 

477 "Other community event", 

478 "Organized event", 

479 } 

480 

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

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

483 

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

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

486 

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

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

489 

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

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

492 "Subscribed event", 

493 "Attending event", 

494 "Community event", 

495 "Organized event", 

496 } 

497 

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

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