Coverage for src/tests/test_search.py: 100%
222 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-12-20 18:03 +0000
1from datetime import timedelta
3import pytest
4from google.protobuf import wrappers_pb2
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)
21@pytest.fixture(autouse=True)
22def _(testconfig):
23 pass
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 )
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
60def test_regression_search_in_area(db):
61 """
62 Makes sure search_in_area works.
64 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
65 """
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)
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]
91def test_user_search_in_rectangle(db):
92 """
93 Makes sure search_in_rectangle works as expected.
94 """
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)
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]
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)
131 user_incomplete_profile, token7 = generate_user(complete_profile=False)
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]
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]
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)
148 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
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]
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]
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 }
177@pytest.fixture
178def create_event(sample_event_data):
179 """Factory for creating events."""
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}))
185 return _create_event
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
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
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()
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)
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}
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
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()
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 )
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}
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}
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
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()
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 )
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 )
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)
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()
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 )
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 )
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)
323def test_event_search_pagination(sample_community, create_event):
324 """Test that EventSearch paginates correctly.
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()
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 )
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)))
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 == ""
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)))
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 == ""
369def test_event_search_pagination_with_page_number(sample_community, create_event):
370 """Test that EventSearch paginates correctly with page number.
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()
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 )
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
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
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
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
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()
416 with events_session(token) as api:
417 create_event(api, title="Offline event")
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 )
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"}
432 res = api.EventSearch(search_pb2.EventSearchReq(only_online=True))
433 assert {event.title for event in res.events} == {"Online event"}
435 res = api.EventSearch(search_pb2.EventSearchReq(only_offline=True))
436 assert {event.title for event in res.events} == {"Offline event"}
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()
446 with communities_session(token) as api:
447 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
449 with session_scope() as session:
450 create_community(session, 55, 60, "Other community", [other_user], [], None)
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 )
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 )
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 }
481 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
482 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
484 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
485 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
487 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
488 assert {event.title for event in res.events} == {"Organized event"}
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 }
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"}