Coverage for src/tests/test_search.py: 100%
278 statements
« prev ^ index » next coverage.py v7.6.10, created at 2025-04-16 15:13 +0000
1from datetime import timedelta
3import pytest
4from google.protobuf import wrappers_pb2
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 testconfig,
18)
19from tests.test_references import create_friend_reference
22@pytest.fixture(autouse=True)
23def _(testconfig):
24 pass
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 )
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 assert res.total_items == len(res.results)
62def test_regression_search_in_area(db):
63 """
64 Makes sure search_in_area works.
66 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
67 """
69 # outside
70 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
71 # outside
72 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
73 # inside
74 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
75 # inside
76 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
77 # outside
78 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
80 with search_session(token5) as api:
81 res = api.UserSearch(
82 search_pb2.UserSearchReq(
83 search_in_area=search_pb2.Area(
84 lat=0,
85 lng=0,
86 radius=100000,
87 )
88 )
89 )
90 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
93def test_user_search_in_rectangle(db):
94 """
95 Makes sure search_in_rectangle works as expected.
96 """
98 # outside
99 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
100 # outside
101 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
102 # inside
103 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
104 # inside
105 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
106 # outside (not fully inside)
107 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
108 # outside
109 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
110 # outside
111 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
113 with search_session(token5) as api:
114 res = api.UserSearch(
115 search_pb2.UserSearchReq(
116 search_in_rectangle=search_pb2.RectArea(
117 lat_min=0,
118 lat_max=2,
119 lng_min=0,
120 lng_max=1,
121 )
122 )
123 )
124 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
127def test_user_filter_complete_profile(db):
128 """
129 Make sure the completed profile flag returns only completed user profile
130 """
131 user_complete_profile, token6 = generate_user(complete_profile=True)
133 user_incomplete_profile, token7 = generate_user(complete_profile=False)
135 with search_session(token7) as api:
136 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
137 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
139 with search_session(token6) as api:
140 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
141 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
144def test_user_filter_meetup_status(db):
145 """
146 Make sure the completed profile flag returns only completed user profile
147 """
148 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
150 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
152 with search_session(token8) as api:
153 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
154 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
156 with search_session(token9) as api:
157 res = api.UserSearch(
158 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
159 )
160 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
163def test_user_filter_language(db):
164 """
165 Test filtering users by language ability.
166 """
167 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
168 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
169 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
171 with session_scope() as session:
172 session.add(
173 LanguageAbility(
174 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
175 ),
176 )
177 session.add(
178 LanguageAbility(
179 user_id=user_with_japanese_conversational.id,
180 language_code="jpn",
181 fluency=LanguageFluency.fluent,
182 )
183 )
184 session.add(
185 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
186 )
188 with search_session(token11) as api:
189 res = api.UserSearch(
190 search_pb2.UserSearchReq(
191 language_ability_filter=[
192 api_pb2.LanguageAbility(
193 code="deu",
194 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
195 )
196 ]
197 )
198 )
199 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
201 res = api.UserSearch(
202 search_pb2.UserSearchReq(
203 language_ability_filter=[
204 api_pb2.LanguageAbility(
205 code="jpn",
206 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
207 )
208 ]
209 )
210 )
211 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
214def test_user_filter_strong_verification(db):
215 user1, token1 = generate_user()
216 user2, _ = generate_user(strong_verification=True)
217 user3, _ = generate_user()
218 user4, _ = generate_user(strong_verification=True)
219 user5, _ = generate_user(strong_verification=True)
221 with search_session(token1) as api:
222 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
223 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
225 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
226 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
229def test_regression_search_only_with_references(db):
230 user1, token1 = generate_user()
231 user2, _ = generate_user()
232 user3, _ = generate_user()
233 user4, _ = generate_user(delete_user=True)
235 with session_scope() as session:
236 # user 2 has references
237 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
238 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
239 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
241 # user 3 only has reference from a deleted user
242 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
244 with search_session(token1) as api:
245 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
246 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
248 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
249 assert [result.user.user_id for result in res.results] == [user2.id]
252def test_user_search_exactly_user_ids(db):
253 """
254 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
255 """
256 # Create users with different properties
257 user1, token1 = generate_user()
258 user2, _ = generate_user(strong_verification=True)
259 user3, _ = generate_user(complete_profile=True)
260 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
261 user5, _ = generate_user(delete_user=True) # Deleted user
263 with search_session(token1) as api:
264 # Test that exactly_user_ids returns only the specified users
265 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
266 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
268 # Test that exactly_user_ids ignores other filters
269 res = api.UserSearch(
270 search_pb2.UserSearchReq(
271 exactly_user_ids=[user2.id, user3.id, user4.id],
272 only_with_strong_verification=True, # This would normally filter out user3 and user4
273 )
274 )
275 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
277 # Test with non-existent user IDs (should be ignored)
278 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
279 assert [result.user.user_id for result in res.results] == [user1.id]
281 # Test with deleted user ID (should be ignored due to visibility filter)
282 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
283 assert [result.user.user_id for result in res.results] == [user1.id]
286@pytest.fixture
287def sample_event_data() -> dict:
288 """Dummy data for creating events."""
289 start_time = now() + timedelta(hours=2)
290 end_time = start_time + timedelta(hours=3)
291 return {
292 "title": "Dummy Title",
293 "content": "Dummy content.",
294 "photo_key": None,
295 "offline_information": events_pb2.OfflineEventInformation(address="Near Null Island", lat=0.1, lng=0.2),
296 "start_time": Timestamp_from_datetime(start_time),
297 "end_time": Timestamp_from_datetime(end_time),
298 "timezone": "UTC",
299 }
302@pytest.fixture
303def create_event(sample_event_data):
304 """Factory for creating events."""
306 def _create_event(event_api, **kwargs) -> EventOccurrence:
307 """Create an event with default values, unless overridden by kwargs."""
308 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs}))
310 return _create_event
313@pytest.fixture
314def sample_community(db) -> int:
315 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""
316 user, _ = generate_user()
317 with session_scope() as session:
318 return create_community(session, -50, 50, "Community", [user], [], None).id
321def test_EventSearch_no_filters(testing_communities):
322 """Test that EventSearch returns all events if no filter is set."""
323 user, token = generate_user()
324 with search_session(token) as api:
325 res = api.EventSearch(search_pb2.EventSearchReq())
326 assert len(res.events) > 0
329def test_event_search_by_query(sample_community, create_event):
330 """Test that EventSearch finds events by title (and content if query_title_only=False)."""
331 user, token = generate_user()
333 with events_session(token) as api:
334 event1 = create_event(api, title="Lorem Ipsum")
335 event2 = create_event(api, content="Lorem Ipsum")
336 create_event(api)
338 with search_session(token) as api:
339 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))
340 assert len(res.events) == 2
341 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
343 res = api.EventSearch(
344 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)
345 )
346 assert len(res.events) == 1
347 assert res.events[0].event_id == event1.event_id
350def test_event_search_by_time(sample_community, create_event):
351 """Test that EventSearch filters with the given time range."""
352 user, token = generate_user()
354 with events_session(token) as api:
355 event1 = create_event(
356 api,
357 start_time=Timestamp_from_datetime(now() + timedelta(hours=1)),
358 end_time=Timestamp_from_datetime(now() + timedelta(hours=2)),
359 )
360 event2 = create_event(
361 api,
362 start_time=Timestamp_from_datetime(now() + timedelta(hours=4)),
363 end_time=Timestamp_from_datetime(now() + timedelta(hours=5)),
364 )
365 event3 = create_event(
366 api,
367 start_time=Timestamp_from_datetime(now() + timedelta(hours=7)),
368 end_time=Timestamp_from_datetime(now() + timedelta(hours=8)),
369 )
371 with search_session(token) as api:
372 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))
373 assert len(res.events) == 2
374 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
376 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))
377 assert len(res.events) == 2
378 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}
380 res = api.EventSearch(
381 search_pb2.EventSearchReq(
382 before=Timestamp_from_datetime(now() + timedelta(hours=6)),
383 after=Timestamp_from_datetime(now() + timedelta(hours=3)),
384 )
385 )
386 assert len(res.events) == 1
387 assert res.events[0].event_id == event2.event_id
390def test_event_search_by_circle(sample_community, create_event):
391 """Test that EventSearch only returns events within the given circle."""
392 user, token = generate_user()
394 with events_session(token) as api:
395 inside_pts = [(0.1, 0.01), (0.01, 0.1)]
396 for i, (lat, lng) in enumerate(inside_pts):
397 create_event(
398 api,
399 title=f"Inside area {i}",
400 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),
401 )
403 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]
404 for i, (lat, lng) in enumerate(outside_pts):
405 create_event(
406 api,
407 title=f"Outside area {i}",
408 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),
409 )
411 with search_session(token) as api:
412 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))
413 assert len(res.events) == len(inside_pts)
414 assert all(event.title.startswith("Inside area") for event in res.events)
417def test_event_search_by_rectangle(sample_community, create_event):
418 """Test that EventSearch only returns events within the given rectangular area."""
419 user, token = generate_user()
421 with events_session(token) as api:
422 inside_pts = [(0.1, 0.2), (1.2, 0.2)]
423 for i, (lat, lng) in enumerate(inside_pts):
424 create_event(
425 api,
426 title=f"Inside area {i}",
427 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),
428 )
430 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]
431 for i, (lat, lng) in enumerate(outside_pts):
432 create_event(
433 api,
434 title=f"Outside area {i}",
435 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),
436 )
438 with search_session(token) as api:
439 res = api.EventSearch(
440 search_pb2.EventSearchReq(
441 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)
442 )
443 )
444 assert len(res.events) == len(inside_pts)
445 assert all(event.title.startswith("Inside area") for event in res.events)
448def test_event_search_pagination(sample_community, create_event):
449 """Test that EventSearch paginates correctly.
451 Check that
452 - <page_size> events are returned, if available
453 - sort order is applied (default: past=False)
454 - the next page token is correct
455 """
456 user, token = generate_user()
458 anchor_time = now()
459 with events_session(token) as api:
460 for i in range(5):
461 create_event(
462 api,
463 title=f"Event {i + 1}",
464 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
465 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),
466 )
468 with search_session(token) as api:
469 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))
470 assert len(res.events) == 4
471 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]
472 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=5, minutes=30)))
474 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))
475 assert len(res.events) == 1
476 assert res.events[0].title == "Event 5"
477 assert res.next_page_token == ""
479 res = api.EventSearch(
480 search_pb2.EventSearchReq(
481 past=True, page_size=2, page_token=str(millis_from_dt(anchor_time + timedelta(hours=4, minutes=30)))
482 )
483 )
484 assert len(res.events) == 2
485 assert [event.title for event in res.events] == ["Event 4", "Event 3"]
486 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=2, minutes=30)))
488 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
489 assert len(res.events) == 2
490 assert [event.title for event in res.events] == ["Event 2", "Event 1"]
491 assert res.next_page_token == ""
494def test_event_search_pagination_with_page_number(sample_community, create_event):
495 """Test that EventSearch paginates correctly with page number.
497 Check that
498 - <page_size> events are returned, if available
499 - sort order is applied (default: past=False)
500 - <page_number> is respected
501 - <total_items> is correct
502 """
503 user, token = generate_user()
505 anchor_time = now()
506 with events_session(token) as api:
507 for i in range(5):
508 create_event(
509 api,
510 title=f"Event {i + 1}",
511 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
512 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),
513 )
515 with search_session(token) as api:
516 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
517 assert len(res.events) == 2
518 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
519 assert res.total_items == 5
521 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
522 assert len(res.events) == 2
523 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
524 assert res.total_items == 5
526 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
527 assert len(res.events) == 1
528 assert [event.title for event in res.events] == ["Event 5"]
529 assert res.total_items == 5
531 # Verify no more pages
532 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
533 assert not res.events
534 assert res.total_items == 5
537def test_event_search_online_status(sample_community, create_event):
538 """Test that EventSearch respects only_online and only_offline filters and by default returns both."""
539 user, token = generate_user()
541 with events_session(token) as api:
542 create_event(api, title="Offline event")
544 create_event(
545 api,
546 title="Online event",
547 online_information=events_pb2.OnlineEventInformation(link="https://couchers.org/meet/"),
548 parent_community_id=sample_community,
549 offline_information=events_pb2.OfflineEventInformation(),
550 )
552 with search_session(token) as api:
553 res = api.EventSearch(search_pb2.EventSearchReq())
554 assert len(res.events) == 2
555 assert {event.title for event in res.events} == {"Offline event", "Online event"}
557 res = api.EventSearch(search_pb2.EventSearchReq(only_online=True))
558 assert {event.title for event in res.events} == {"Online event"}
560 res = api.EventSearch(search_pb2.EventSearchReq(only_offline=True))
561 assert {event.title for event in res.events} == {"Offline event"}
564def test_event_search_filter_subscription_attendance_organizing_my_communities(sample_community, create_event):
565 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
566 returns all events.
567 """
568 _, token = generate_user()
569 other_user, other_token = generate_user()
571 with communities_session(token) as api:
572 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
574 with session_scope() as session:
575 create_community(session, 55, 60, "Other community", [other_user], [], None)
577 with events_session(other_token) as api:
578 e_subscribed = create_event(api, title="Subscribed event")
579 e_attending = create_event(api, title="Attending event")
580 create_event(api, title="Community event")
581 create_event(
582 api,
583 title="Other community event",
584 offline_information=events_pb2.OfflineEventInformation(lat=58, lng=1, address="Somewhere"),
585 )
587 with events_session(token) as api:
588 create_event(api, title="Organized event")
589 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
590 api.SetEventAttendance(
591 events_pb2.SetEventAttendanceReq(
592 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
593 )
594 )
596 with search_session(token) as api:
597 res = api.EventSearch(search_pb2.EventSearchReq())
598 assert {event.title for event in res.events} == {
599 "Subscribed event",
600 "Attending event",
601 "Community event",
602 "Other community event",
603 "Organized event",
604 }
606 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
607 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
609 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
610 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
612 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
613 assert {event.title for event in res.events} == {"Organized event"}
615 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
616 assert {event.title for event in res.events} == {
617 "Subscribed event",
618 "Attending event",
619 "Community event",
620 "Organized event",
621 }
623 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
624 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}