Coverage for app / backend / src / tests / test_search.py: 100%
436 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 14:14 +0000
1from datetime import timedelta
2from typing import Any
4import pytest
5from google.protobuf import empty_pb2, wrappers_pb2
6from sqlalchemy import select
8from couchers.db import session_scope
9from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid
10from couchers.models import EventOccurrence, HostingStatus, LanguageAbility, LanguageFluency, MeetupStatus
11from couchers.proto import api_pb2, communities_pb2, events_pb2, search_pb2
12from couchers.utils import Timestamp_from_datetime, create_coordinate, millis_from_dt, now
13from tests.fixtures.db import generate_user
14from tests.fixtures.misc import Moderator
15from tests.fixtures.sessions import communities_session, events_session, search_session
16from tests.test_communities import create_community, testing_communities # noqa
17from tests.test_references import create_friend_reference
20@pytest.fixture(autouse=True)
21def _(testconfig):
22 pass
25def test_Search(testing_communities):
26 user, token = generate_user()
27 with search_session(token) as api:
28 res = api.Search(
29 search_pb2.SearchReq(
30 query="Country 1, Region 1",
31 include_users=True,
32 include_communities=True,
33 include_groups=True,
34 include_places=True,
35 include_guides=True,
36 )
37 )
38 res = api.Search(
39 search_pb2.SearchReq(
40 query="Country 1, Region 1, Attraction",
41 title_only=True,
42 include_users=True,
43 include_communities=True,
44 include_groups=True,
45 include_places=True,
46 include_guides=True,
47 )
48 )
51def test_UserSearch(testing_communities):
52 """Test that UserSearch returns all users if no filter is set."""
53 user, token = generate_user()
55 refresh_materialized_views_rapid(empty_pb2.Empty())
56 refresh_materialized_views(empty_pb2.Empty())
58 with search_session(token) as api:
59 res = api.UserSearch(search_pb2.UserSearchReq())
60 assert len(res.results) > 0
61 assert res.total_items == len(res.results)
62 res = api.UserSearchV2(search_pb2.UserSearchReq())
63 assert len(res.results) > 0
64 assert res.total_items == len(res.results)
67def test_regression_search_in_area(db):
68 """
69 Makes sure search_in_area works.
71 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
72 """
74 # outside
75 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
76 # outside
77 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
78 # inside
79 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
80 # inside
81 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
82 # outside
83 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
85 refresh_materialized_views_rapid(empty_pb2.Empty())
86 refresh_materialized_views(empty_pb2.Empty())
88 with search_session(token5) as api:
89 res = api.UserSearch(
90 search_pb2.UserSearchReq(
91 search_in_area=search_pb2.Area(
92 lat=0,
93 lng=0,
94 radius=100000,
95 )
96 )
97 )
98 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
100 res = api.UserSearchV2(
101 search_pb2.UserSearchReq(
102 search_in_area=search_pb2.Area(
103 lat=0,
104 lng=0,
105 radius=100000,
106 )
107 )
108 )
109 assert [result.user_id for result in res.results] == [user3.id, user4.id]
112def test_user_search_in_rectangle(db):
113 """
114 Makes sure search_in_rectangle works as expected.
115 """
117 # outside
118 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
119 # outside
120 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
121 # inside
122 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
123 # inside
124 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
125 # outside (not fully inside)
126 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
127 # outside
128 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
129 # outside
130 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
132 refresh_materialized_views_rapid(empty_pb2.Empty())
133 refresh_materialized_views(empty_pb2.Empty())
135 with search_session(token5) as api:
136 res = api.UserSearch(
137 search_pb2.UserSearchReq(
138 search_in_rectangle=search_pb2.RectArea(
139 lat_min=0,
140 lat_max=2,
141 lng_min=0,
142 lng_max=1,
143 )
144 )
145 )
146 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
148 res = api.UserSearchV2(
149 search_pb2.UserSearchReq(
150 search_in_rectangle=search_pb2.RectArea(
151 lat_min=0,
152 lat_max=2,
153 lng_min=0,
154 lng_max=1,
155 )
156 )
157 )
158 assert [result.user_id for result in res.results] == [user3.id, user4.id]
161def test_user_filter_complete_profile(db):
162 """
163 Make sure the completed profile flag returns only completed user profile
164 """
165 user_complete_profile, token6 = generate_user(complete_profile=True)
167 user_incomplete_profile, token7 = generate_user(complete_profile=False)
169 refresh_materialized_views_rapid(empty_pb2.Empty())
170 refresh_materialized_views(empty_pb2.Empty())
172 with search_session(token7) as api:
173 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
174 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
176 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
177 assert user_incomplete_profile.id in [result.user_id for result in res.results]
179 with search_session(token6) as api:
180 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
181 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
183 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
184 assert [result.user_id for result in res.results] == [user_complete_profile.id]
187def test_user_filter_meetup_status(db):
188 """
189 Make sure the completed profile flag returns only completed user profile
190 """
191 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
193 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
195 refresh_materialized_views_rapid(empty_pb2.Empty())
196 refresh_materialized_views(empty_pb2.Empty())
198 with search_session(token8) as api:
199 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
200 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
202 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
203 assert user_wants_to_meetup.id in [result.user_id for result in res.results]
205 with search_session(token9) as api:
206 res = api.UserSearch(
207 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
208 )
209 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
211 res = api.UserSearchV2(
212 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
213 )
214 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]
217def test_user_filter_language(db):
218 """
219 Test filtering users by language ability.
220 """
221 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
222 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
223 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
225 with session_scope() as session:
226 session.add(
227 LanguageAbility(
228 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
229 ),
230 )
231 session.add(
232 LanguageAbility(
233 user_id=user_with_japanese_conversational.id,
234 language_code="jpn",
235 fluency=LanguageFluency.fluent,
236 )
237 )
238 session.add(
239 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
240 )
242 refresh_materialized_views_rapid(empty_pb2.Empty())
243 refresh_materialized_views(empty_pb2.Empty())
245 with search_session(token11) as api:
246 res = api.UserSearch(
247 search_pb2.UserSearchReq(
248 language_ability_filter=[
249 api_pb2.LanguageAbility(
250 code="deu",
251 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
252 )
253 ]
254 )
255 )
256 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
258 res = api.UserSearchV2(
259 search_pb2.UserSearchReq(
260 language_ability_filter=[
261 api_pb2.LanguageAbility(
262 code="deu",
263 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
264 )
265 ]
266 )
267 )
268 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]
270 res = api.UserSearch(
271 search_pb2.UserSearchReq(
272 language_ability_filter=[
273 api_pb2.LanguageAbility(
274 code="jpn",
275 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
276 )
277 ]
278 )
279 )
280 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
282 res = api.UserSearchV2(
283 search_pb2.UserSearchReq(
284 language_ability_filter=[
285 api_pb2.LanguageAbility(
286 code="jpn",
287 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
288 )
289 ]
290 )
291 )
292 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]
295def test_user_filter_strong_verification(db):
296 user1, token1 = generate_user()
297 user2, _ = generate_user(strong_verification=True)
298 user3, _ = generate_user()
299 user4, _ = generate_user(strong_verification=True)
300 user5, _ = generate_user(strong_verification=True)
302 refresh_materialized_views_rapid(empty_pb2.Empty())
303 refresh_materialized_views(empty_pb2.Empty())
305 with search_session(token1) as api:
306 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
307 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
309 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))
310 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
312 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
313 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
315 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))
316 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]
319def test_regression_search_only_with_references(db):
320 user1, token1 = generate_user()
321 user2, _ = generate_user()
322 user3, _ = generate_user()
323 user4, _ = generate_user(delete_user=True)
325 refresh_materialized_views_rapid(empty_pb2.Empty())
326 refresh_materialized_views(empty_pb2.Empty())
328 with session_scope() as session:
329 # user 2 has references
330 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
331 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
332 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
334 # user 3 only has reference from a deleted user
335 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
337 with search_session(token1) as api:
338 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
339 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
341 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))
342 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]
344 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
345 assert [result.user.user_id for result in res.results] == [user2.id]
347 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
348 assert [result.user_id for result in res.results] == [user2.id]
351def test_user_search_exactly_user_ids(db):
352 """
353 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
354 """
355 # Create users with different properties
356 user1, token1 = generate_user()
357 user2, _ = generate_user(strong_verification=True)
358 user3, _ = generate_user(complete_profile=True)
359 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
360 user5, _ = generate_user(delete_user=True) # Deleted user
362 refresh_materialized_views_rapid(empty_pb2.Empty())
363 refresh_materialized_views(empty_pb2.Empty())
365 with search_session(token1) as api:
366 # Test that exactly_user_ids returns only the specified users
367 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
368 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
370 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
371 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
373 # Test that exactly_user_ids ignores other filters
374 res = api.UserSearch(
375 search_pb2.UserSearchReq(
376 exactly_user_ids=[user2.id, user3.id, user4.id],
377 only_with_strong_verification=True, # This would normally filter out user3 and user4
378 )
379 )
380 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
382 res = api.UserSearchV2(
383 search_pb2.UserSearchReq(
384 exactly_user_ids=[user2.id, user3.id, user4.id],
385 only_with_strong_verification=True, # This would normally filter out user3 and user4
386 )
387 )
388 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
390 # Test with non-existent user IDs (should be ignored)
391 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
392 assert [result.user.user_id for result in res.results] == [user1.id]
394 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
395 assert [result.user_id for result in res.results] == [user1.id]
397 # Test with deleted user ID (should be ignored due to visibility filter)
398 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
399 assert [result.user.user_id for result in res.results] == [user1.id]
401 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
402 assert [result.user_id for result in res.results] == [user1.id]
405@pytest.fixture
406def sample_event_data() -> dict[str, Any]:
407 """Dummy data for creating events."""
408 start_time = now() + timedelta(hours=2)
409 end_time = start_time + timedelta(hours=3)
410 return {
411 "title": "Dummy Title",
412 "content": "Dummy content.",
413 "photo_key": None,
414 "offline_information": events_pb2.OfflineEventInformation(address="Near Null Island", lat=0.1, lng=0.2),
415 "start_time": Timestamp_from_datetime(start_time),
416 "end_time": Timestamp_from_datetime(end_time),
417 "timezone": "UTC",
418 }
421@pytest.fixture
422def create_event(sample_event_data):
423 """Factory for creating events."""
425 def _create_event(event_api, **kwargs) -> EventOccurrence:
426 """Create an event with default values, unless overridden by kwargs."""
427 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore
429 return _create_event
432@pytest.fixture
433def sample_community(db) -> int:
434 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""
435 user, _ = generate_user()
436 with session_scope() as session:
437 return create_community(session, -50, 50, "Community", [user], [], None).id
440def test_EventSearch_no_filters(testing_communities):
441 """Test that EventSearch returns all events if no filter is set."""
442 user, token = generate_user()
443 with search_session(token) as api:
444 res = api.EventSearch(search_pb2.EventSearchReq())
445 assert len(res.events) > 0
448def test_event_search_by_query(sample_community, create_event):
449 """Test that EventSearch finds events by title (and content if query_title_only=False)."""
450 user, token = generate_user()
452 with events_session(token) as api:
453 event1 = create_event(api, title="Lorem Ipsum")
454 event2 = create_event(api, content="Lorem Ipsum")
455 create_event(api)
457 with search_session(token) as api:
458 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))
459 assert len(res.events) == 2
460 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
462 res = api.EventSearch(
463 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)
464 )
465 assert len(res.events) == 1
466 assert res.events[0].event_id == event1.event_id
469def test_event_search_by_time(sample_community, create_event):
470 """Test that EventSearch filters with the given time range."""
471 user, token = generate_user()
473 with events_session(token) as api:
474 event1 = create_event(
475 api,
476 start_time=Timestamp_from_datetime(now() + timedelta(hours=1)),
477 end_time=Timestamp_from_datetime(now() + timedelta(hours=2)),
478 )
479 event2 = create_event(
480 api,
481 start_time=Timestamp_from_datetime(now() + timedelta(hours=4)),
482 end_time=Timestamp_from_datetime(now() + timedelta(hours=5)),
483 )
484 event3 = create_event(
485 api,
486 start_time=Timestamp_from_datetime(now() + timedelta(hours=7)),
487 end_time=Timestamp_from_datetime(now() + timedelta(hours=8)),
488 )
490 with search_session(token) as api:
491 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))
492 assert len(res.events) == 2
493 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
495 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))
496 assert len(res.events) == 2
497 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}
499 res = api.EventSearch(
500 search_pb2.EventSearchReq(
501 before=Timestamp_from_datetime(now() + timedelta(hours=6)),
502 after=Timestamp_from_datetime(now() + timedelta(hours=3)),
503 )
504 )
505 assert len(res.events) == 1
506 assert res.events[0].event_id == event2.event_id
509def test_event_search_by_circle(sample_community, create_event):
510 """Test that EventSearch only returns events within the given circle."""
511 user, token = generate_user()
513 with events_session(token) as api:
514 inside_pts = [(0.1, 0.01), (0.01, 0.1)]
515 for i, (lat, lng) in enumerate(inside_pts):
516 create_event(
517 api,
518 title=f"Inside area {i}",
519 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),
520 )
522 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]
523 for i, (lat, lng) in enumerate(outside_pts):
524 create_event(
525 api,
526 title=f"Outside area {i}",
527 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),
528 )
530 with search_session(token) as api:
531 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))
532 assert len(res.events) == len(inside_pts)
533 assert all(event.title.startswith("Inside area") for event in res.events)
536def test_event_search_by_rectangle(sample_community, create_event):
537 """Test that EventSearch only returns events within the given rectangular area."""
538 user, token = generate_user()
540 with events_session(token) as api:
541 inside_pts = [(0.1, 0.2), (1.2, 0.2)]
542 for i, (lat, lng) in enumerate(inside_pts):
543 create_event(
544 api,
545 title=f"Inside area {i}",
546 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),
547 )
549 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]
550 for i, (lat, lng) in enumerate(outside_pts):
551 create_event(
552 api,
553 title=f"Outside area {i}",
554 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),
555 )
557 with search_session(token) as api:
558 res = api.EventSearch(
559 search_pb2.EventSearchReq(
560 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)
561 )
562 )
563 assert len(res.events) == len(inside_pts)
564 assert all(event.title.startswith("Inside area") for event in res.events)
567def test_event_search_pagination(sample_community, create_event):
568 """Test that EventSearch paginates correctly.
570 Check that
571 - <page_size> events are returned, if available
572 - sort order is applied (default: past=False)
573 - the next page token is correct
574 """
575 user, token = generate_user()
577 anchor_time = now()
578 with events_session(token) as api:
579 for i in range(5):
580 create_event(
581 api,
582 title=f"Event {i + 1}",
583 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
584 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),
585 )
587 with search_session(token) as api:
588 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))
589 assert len(res.events) == 4
590 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]
591 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=5, minutes=30)))
593 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))
594 assert len(res.events) == 1
595 assert res.events[0].title == "Event 5"
596 assert res.next_page_token == ""
598 res = api.EventSearch(
599 search_pb2.EventSearchReq(
600 past=True, page_size=2, page_token=str(millis_from_dt(anchor_time + timedelta(hours=4, minutes=30)))
601 )
602 )
603 assert len(res.events) == 2
604 assert [event.title for event in res.events] == ["Event 4", "Event 3"]
605 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=2, minutes=30)))
607 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
608 assert len(res.events) == 2
609 assert [event.title for event in res.events] == ["Event 2", "Event 1"]
610 assert res.next_page_token == ""
613def test_event_search_pagination_with_page_number(sample_community, create_event):
614 """Test that EventSearch paginates correctly with page number.
616 Check that
617 - <page_size> events are returned, if available
618 - sort order is applied (default: past=False)
619 - <page_number> is respected
620 - <total_items> is correct
621 """
622 user, token = generate_user()
624 anchor_time = now()
625 with events_session(token) as api:
626 for i in range(5):
627 create_event(
628 api,
629 title=f"Event {i + 1}",
630 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
631 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),
632 )
634 with search_session(token) as api:
635 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
636 assert len(res.events) == 2
637 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
638 assert res.total_items == 5
640 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
641 assert len(res.events) == 2
642 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
643 assert res.total_items == 5
645 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
646 assert len(res.events) == 1
647 assert [event.title for event in res.events] == ["Event 5"]
648 assert res.total_items == 5
650 # Verify no more pages
651 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
652 assert not res.events
653 assert res.total_items == 5
656def test_event_search_online_status(sample_community, create_event):
657 """Test that EventSearch respects only_online and only_offline filters and by default returns both."""
658 user, token = generate_user()
660 with events_session(token) as api:
661 create_event(api, title="Offline event")
663 create_event(
664 api,
665 title="Online event",
666 online_information=events_pb2.OnlineEventInformation(link="https://couchers.org/meet/"),
667 parent_community_id=sample_community,
668 offline_information=events_pb2.OfflineEventInformation(),
669 )
671 with search_session(token) as api:
672 res = api.EventSearch(search_pb2.EventSearchReq())
673 assert len(res.events) == 2
674 assert {event.title for event in res.events} == {"Offline event", "Online event"}
676 res = api.EventSearch(search_pb2.EventSearchReq(only_online=True))
677 assert {event.title for event in res.events} == {"Online event"}
679 res = api.EventSearch(search_pb2.EventSearchReq(only_offline=True))
680 assert {event.title for event in res.events} == {"Offline event"}
683def test_event_search_filter_subscription_attendance_organizing_my_communities(
684 sample_community, create_event, moderator: Moderator
685):
686 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
687 returns all events.
688 """
689 _, token = generate_user()
690 other_user, other_token = generate_user()
692 with communities_session(token) as api:
693 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
695 with session_scope() as session:
696 create_community(session, 55, 60, "Other community", [other_user], [], None)
698 with events_session(other_token) as api:
699 e_subscribed = create_event(api, title="Subscribed event")
700 e_attending = create_event(api, title="Attending event")
701 create_event(api, title="Community event")
702 create_event(
703 api,
704 title="Other community event",
705 offline_information=events_pb2.OfflineEventInformation(lat=58, lng=1, address="Somewhere"),
706 )
708 # Approve all events so they're visible to other users
709 with session_scope() as session:
710 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
711 for oid in occurrence_ids:
712 moderator.approve_event_occurrence(oid)
714 with events_session(token) as api:
715 create_event(api, title="Organized event")
716 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
717 api.SetEventAttendance(
718 events_pb2.SetEventAttendanceReq(
719 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
720 )
721 )
723 with search_session(token) as api:
724 res = api.EventSearch(search_pb2.EventSearchReq())
725 assert {event.title for event in res.events} == {
726 "Subscribed event",
727 "Attending event",
728 "Community event",
729 "Other community event",
730 "Organized event",
731 }
733 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
734 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
736 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
737 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
739 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
740 assert {event.title for event in res.events} == {"Organized event"}
742 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
743 assert {event.title for event in res.events} == {
744 "Subscribed event",
745 "Attending event",
746 "Community event",
747 "Organized event",
748 }
750 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
751 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}
754def test_regression_search_multiple_pages(db):
755 """
756 There was a bug when there are multiple pages of results
757 """
758 user, token = generate_user()
759 user_ids = [user.id]
760 for _ in range(10):
761 other_user, _ = generate_user()
762 user_ids.append(other_user.id)
764 refresh_materialized_views_rapid(empty_pb2.Empty())
765 refresh_materialized_views(empty_pb2.Empty())
767 with search_session(token) as api:
768 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))
769 assert [result.user_id for result in res.results] == user_ids[:5]
770 assert res.next_page_token
773def test_regression_search_no_results(db):
774 """
775 There was a bug when there were no results
776 """
777 # put us far away
778 user, token = generate_user()
780 refresh_materialized_views_rapid(empty_pb2.Empty())
781 refresh_materialized_views(empty_pb2.Empty())
783 with search_session(token) as api:
784 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
785 assert len(res.results) == 0
788def test_user_filter_same_gender_only(db):
789 """Test that same_gender_only filter works correctly"""
790 # Create users with different genders and strong verification status
791 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")
792 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")
793 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")
794 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")
795 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")
797 refresh_materialized_views_rapid(empty_pb2.Empty())
798 refresh_materialized_views(empty_pb2.Empty())
800 # Test 1: Woman with strong verification should see only women when same_gender_only=True
801 with search_session(token_woman_with_sv) as api:
802 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
803 result_ids = [result.user.user_id for result in res.results]
804 assert woman_with_sv.id in result_ids
805 assert woman_without_sv.id in result_ids
806 assert other_woman_with_sv.id in result_ids
807 assert man_with_sv.id not in result_ids
808 assert man_without_sv.id not in result_ids
810 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
811 result_ids = [result.user_id for result in res.results]
812 assert woman_with_sv.id in result_ids
813 assert woman_without_sv.id in result_ids
814 assert other_woman_with_sv.id in result_ids
815 assert man_with_sv.id not in result_ids
816 assert man_without_sv.id not in result_ids
818 # Test 2: Man with strong verification should see only men when same_gender_only=True
819 with search_session(token_man_with_sv) as api:
820 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
821 result_ids = [result.user.user_id for result in res.results]
822 assert man_with_sv.id in result_ids
823 assert man_without_sv.id in result_ids
824 assert woman_with_sv.id not in result_ids
825 assert woman_without_sv.id not in result_ids
826 assert other_woman_with_sv.id not in result_ids
828 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
829 result_ids = [result.user_id for result in res.results]
830 assert man_with_sv.id in result_ids
831 assert man_without_sv.id in result_ids
832 assert woman_with_sv.id not in result_ids
833 assert woman_without_sv.id not in result_ids
834 assert other_woman_with_sv.id not in result_ids
836 # Test 3: Woman without strong verification should get an error
837 with search_session(token_woman_without_sv) as api:
838 with pytest.raises(Exception) as e:
839 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
840 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
842 with pytest.raises(Exception) as e:
843 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
844 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
846 # Test 4: When same_gender_only=False, should see all users
847 with search_session(token_woman_with_sv) as api:
848 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))
849 result_ids = [result.user.user_id for result in res.results]
850 assert woman_with_sv.id in result_ids
851 assert woman_without_sv.id in result_ids
852 assert other_woman_with_sv.id in result_ids
853 assert man_with_sv.id in result_ids
854 assert man_without_sv.id in result_ids
856 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))
857 result_ids = [result.user_id for result in res.results]
858 assert woman_with_sv.id in result_ids
859 assert woman_without_sv.id in result_ids
860 assert other_woman_with_sv.id in result_ids
861 assert man_with_sv.id in result_ids
862 assert man_without_sv.id in result_ids
865def test_user_filter_same_gender_only_with_other_filters(db):
866 """Test that same_gender_only filter works correctly combined with other filters"""
867 # Create users with different properties
868 woman_host, token_woman = generate_user(
869 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host
870 )
871 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)
872 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)
874 refresh_materialized_views_rapid(empty_pb2.Empty())
875 refresh_materialized_views(empty_pb2.Empty())
877 # Test: Combine same_gender_only with hosting_status filter
878 with search_session(token_woman) as api:
879 res = api.UserSearch(
880 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
881 )
882 result_ids = [result.user.user_id for result in res.results]
883 # Should only see woman who can host
884 assert woman_host.id in result_ids
885 assert woman_cant_host.id not in result_ids
886 assert man_host.id not in result_ids
888 res = api.UserSearchV2(
889 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
890 )
891 result_ids = [result.user_id for result in res.results]
892 assert woman_host.id in result_ids
893 assert woman_cant_host.id not in result_ids
894 assert man_host.id not in result_ids