Coverage for app/backend/src/tests/test_search.py: 100%
462 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-10 12:25 +0000
1from datetime import timedelta
2from typing import Any
4import grpc
5import pytest
6from google.protobuf import empty_pb2, wrappers_pb2
7from psycopg.types.range import TimestamptzRange
8from sqlalchemy import select
10from couchers.db import session_scope
11from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid
12from couchers.models import EventOccurrence, HostingStatus, LanguageAbility, LanguageFluency, MeetupStatus
13from couchers.proto import api_pb2, communities_pb2, events_pb2, search_pb2
14from couchers.utils import Timestamp_from_datetime, create_coordinate, datetime_to_iso8601_local, now
15from tests.fixtures.db import generate_user
16from tests.fixtures.misc import Moderator
17from tests.fixtures.sessions import communities_session, events_session, search_session
18from tests.test_communities import create_community, testing_communities # noqa
19from tests.test_references import create_friend_reference
22@pytest.fixture(autouse=True)
23def _(testconfig):
24 pass
27class TestSearchInCommunities:
28 """The tests that search the whole community tree, grouped so they share one copy of it."""
30 @staticmethod
31 def test_Search(testing_communities):
32 user, token = generate_user()
33 with search_session(token) as api:
34 res = api.Search(
35 search_pb2.SearchReq(
36 query="Country 1, Region 1",
37 include_users=True,
38 include_communities=True,
39 include_groups=True,
40 include_places=True,
41 include_guides=True,
42 )
43 )
44 res = api.Search(
45 search_pb2.SearchReq(
46 query="Country 1, Region 1, Attraction",
47 title_only=True,
48 include_users=True,
49 include_communities=True,
50 include_groups=True,
51 include_places=True,
52 include_guides=True,
53 )
54 )
56 @staticmethod
57 def test_UserSearch(testing_communities):
58 """Test that UserSearch returns all users if no filter is set."""
59 user, token = generate_user()
61 refresh_materialized_views_rapid(empty_pb2.Empty())
62 refresh_materialized_views(empty_pb2.Empty())
64 with search_session(token) as api:
65 res = api.UserSearch(search_pb2.UserSearchReq())
66 assert len(res.results) > 0
67 assert res.total_items == len(res.results)
68 res = api.UserSearchV2(search_pb2.UserSearchReq())
69 assert len(res.results) > 0
70 assert res.total_items == len(res.results)
72 @staticmethod
73 def test_EventSearch_no_filters(testing_communities):
74 """Test that EventSearch returns all events if no filter is set."""
75 user, token = generate_user()
76 with search_session(token) as api:
77 res = api.EventSearch(search_pb2.EventSearchReq())
78 assert len(res.events) > 0
81def test_regression_search_in_area(db):
82 """
83 Makes sure search_in_area works.
85 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
86 """
88 # outside
89 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
90 # outside
91 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
92 # inside
93 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
94 # inside
95 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
96 # outside
97 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
99 refresh_materialized_views_rapid(empty_pb2.Empty())
100 refresh_materialized_views(empty_pb2.Empty())
102 with search_session(token5) as api:
103 res = api.UserSearch(
104 search_pb2.UserSearchReq(
105 search_in_area=search_pb2.Area(
106 lat=0,
107 lng=0,
108 radius=100000,
109 )
110 )
111 )
112 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
114 res = api.UserSearchV2(
115 search_pb2.UserSearchReq(
116 search_in_area=search_pb2.Area(
117 lat=0,
118 lng=0,
119 radius=100000,
120 )
121 )
122 )
123 assert [result.user_id for result in res.results] == [user3.id, user4.id]
126def test_user_search_in_rectangle(db):
127 """
128 Makes sure search_in_rectangle works as expected.
129 """
131 # outside
132 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
133 # outside
134 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
135 # inside
136 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
137 # inside
138 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
139 # outside (not fully inside)
140 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
141 # outside
142 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
143 # outside
144 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
146 refresh_materialized_views_rapid(empty_pb2.Empty())
147 refresh_materialized_views(empty_pb2.Empty())
149 with search_session(token5) as api:
150 res = api.UserSearch(
151 search_pb2.UserSearchReq(
152 search_in_rectangle=search_pb2.RectArea(
153 lat_min=0,
154 lat_max=2,
155 lng_min=0,
156 lng_max=1,
157 )
158 )
159 )
160 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
162 res = api.UserSearchV2(
163 search_pb2.UserSearchReq(
164 search_in_rectangle=search_pb2.RectArea(
165 lat_min=0,
166 lat_max=2,
167 lng_min=0,
168 lng_max=1,
169 )
170 )
171 )
172 assert [result.user_id for result in res.results] == [user3.id, user4.id]
175def test_user_filter_complete_profile(db):
176 """
177 Make sure the completed profile flag returns only completed user profile
178 """
179 user_complete_profile, token6 = generate_user(complete_profile=True)
181 user_incomplete_profile, token7 = generate_user(complete_profile=False)
183 refresh_materialized_views_rapid(empty_pb2.Empty())
184 refresh_materialized_views(empty_pb2.Empty())
186 with search_session(token7) as api:
187 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
188 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
190 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
191 assert user_incomplete_profile.id in [result.user_id for result in res.results]
193 with search_session(token6) as api:
194 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
195 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
197 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
198 assert [result.user_id for result in res.results] == [user_complete_profile.id]
201def test_user_filter_meetup_status(db):
202 """
203 Make sure the completed profile flag returns only completed user profile
204 """
205 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
207 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
209 refresh_materialized_views_rapid(empty_pb2.Empty())
210 refresh_materialized_views(empty_pb2.Empty())
212 with search_session(token8) as api:
213 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
214 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
216 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
217 assert user_wants_to_meetup.id in [result.user_id for result in res.results]
219 with search_session(token9) as api:
220 res = api.UserSearch(
221 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
222 )
223 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
225 res = api.UserSearchV2(
226 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
227 )
228 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]
231def test_user_filter_language(db):
232 """
233 Test filtering users by language ability.
234 """
235 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
236 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
237 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
239 with session_scope() as session:
240 session.add(
241 LanguageAbility(
242 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
243 ),
244 )
245 session.add(
246 LanguageAbility(
247 user_id=user_with_japanese_conversational.id,
248 language_code="jpn",
249 fluency=LanguageFluency.fluent,
250 )
251 )
252 session.add(
253 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
254 )
256 refresh_materialized_views_rapid(empty_pb2.Empty())
257 refresh_materialized_views(empty_pb2.Empty())
259 with search_session(token11) as api:
260 res = api.UserSearch(
261 search_pb2.UserSearchReq(
262 language_ability_filter=[
263 api_pb2.LanguageAbility(
264 code="deu",
265 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
266 )
267 ]
268 )
269 )
270 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
272 res = api.UserSearchV2(
273 search_pb2.UserSearchReq(
274 language_ability_filter=[
275 api_pb2.LanguageAbility(
276 code="deu",
277 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
278 )
279 ]
280 )
281 )
282 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]
284 res = api.UserSearch(
285 search_pb2.UserSearchReq(
286 language_ability_filter=[
287 api_pb2.LanguageAbility(
288 code="jpn",
289 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
290 )
291 ]
292 )
293 )
294 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
296 res = api.UserSearchV2(
297 search_pb2.UserSearchReq(
298 language_ability_filter=[
299 api_pb2.LanguageAbility(
300 code="jpn",
301 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
302 )
303 ]
304 )
305 )
306 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]
309def test_user_filter_strong_verification(db):
310 user1, token1 = generate_user()
311 user2, _ = generate_user(strong_verification=True)
312 user3, _ = generate_user()
313 user4, _ = generate_user(strong_verification=True)
314 user5, _ = generate_user(strong_verification=True)
316 refresh_materialized_views_rapid(empty_pb2.Empty())
317 refresh_materialized_views(empty_pb2.Empty())
319 with search_session(token1) as api:
320 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
321 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
323 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))
324 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
326 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
327 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
329 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))
330 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]
333def test_regression_search_only_with_references(db):
334 user1, token1 = generate_user()
335 user2, _ = generate_user()
336 user3, _ = generate_user()
337 user4, _ = generate_user(delete_user=True)
339 refresh_materialized_views_rapid(empty_pb2.Empty())
340 refresh_materialized_views(empty_pb2.Empty())
342 with session_scope() as session:
343 # user 2 has references
344 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
345 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
346 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
348 # user 3 only has reference from a deleted user
349 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
351 with search_session(token1) as api:
352 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
353 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
355 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))
356 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]
358 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
359 assert [result.user.user_id for result in res.results] == [user2.id]
361 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
362 assert [result.user_id for result in res.results] == [user2.id]
365def test_user_search_exactly_user_ids(db):
366 """
367 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
368 """
369 # Create users with different properties
370 user1, token1 = generate_user()
371 user2, _ = generate_user(strong_verification=True)
372 user3, _ = generate_user(complete_profile=True)
373 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
374 user5, _ = generate_user(delete_user=True) # Deleted user
376 refresh_materialized_views_rapid(empty_pb2.Empty())
377 refresh_materialized_views(empty_pb2.Empty())
379 with search_session(token1) as api:
380 # Test that exactly_user_ids returns only the specified users
381 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
382 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
384 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
385 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
387 # Test that exactly_user_ids ignores other filters
388 res = api.UserSearch(
389 search_pb2.UserSearchReq(
390 exactly_user_ids=[user2.id, user3.id, user4.id],
391 only_with_strong_verification=True, # This would normally filter out user3 and user4
392 )
393 )
394 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
396 res = api.UserSearchV2(
397 search_pb2.UserSearchReq(
398 exactly_user_ids=[user2.id, user3.id, user4.id],
399 only_with_strong_verification=True, # This would normally filter out user3 and user4
400 )
401 )
402 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
404 # Test with non-existent user IDs (should be ignored)
405 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
406 assert [result.user.user_id for result in res.results] == [user1.id]
408 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
409 assert [result.user_id for result in res.results] == [user1.id]
411 # Test with deleted user ID (should be ignored due to visibility filter)
412 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
413 assert [result.user.user_id for result in res.results] == [user1.id]
415 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
416 assert [result.user_id for result in res.results] == [user1.id]
419@pytest.fixture
420def sample_event_data() -> dict[str, Any]:
421 """Dummy data for creating events."""
422 start_time = now() + timedelta(hours=2)
423 end_time = start_time + timedelta(hours=3)
424 return {
425 "title": "Dummy Title",
426 "content": "Dummy content.",
427 "photo_key": None,
428 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),
429 "start_datetime_iso8601_local": datetime_to_iso8601_local(start_time),
430 "end_datetime_iso8601_local": datetime_to_iso8601_local(end_time),
431 }
434@pytest.fixture
435def create_event(sample_event_data):
436 """Factory for creating events."""
438 def _create_event(event_api, **kwargs) -> EventOccurrence:
439 """Create an event with default values, unless overridden by kwargs."""
440 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore
442 return _create_event
445@pytest.fixture
446def sample_community(db) -> int:
447 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""
448 user, _ = generate_user()
449 with session_scope() as session:
450 return create_community(session, -50, 50, "Community", [user], [], None).id
453def test_event_search_by_query(sample_community, create_event):
454 """Test that EventSearch finds events by title (and content if query_title_only=False)."""
455 user, token = generate_user()
457 with events_session(token) as api:
458 event1 = create_event(api, title="Lorem Ipsum")
459 event2 = create_event(api, content="Lorem Ipsum")
460 create_event(api)
462 with search_session(token) as api:
463 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))
464 assert len(res.events) == 2
465 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
467 res = api.EventSearch(
468 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)
469 )
470 assert len(res.events) == 1
471 assert res.events[0].event_id == event1.event_id
474def test_event_search_by_time(sample_community, create_event):
475 """Test that EventSearch filters with the given time range."""
476 user, token = generate_user()
478 with events_session(token) as api:
479 event1 = create_event(
480 api,
481 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=1)),
482 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=2)),
483 )
484 event2 = create_event(
485 api,
486 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=4)),
487 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=5)),
488 )
489 event3 = create_event(
490 api,
491 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=7)),
492 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=8)),
493 )
495 with search_session(token) as api:
496 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))
497 assert len(res.events) == 2
498 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
500 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))
501 assert len(res.events) == 2
502 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}
504 res = api.EventSearch(
505 search_pb2.EventSearchReq(
506 before=Timestamp_from_datetime(now() + timedelta(hours=6)),
507 after=Timestamp_from_datetime(now() + timedelta(hours=3)),
508 )
509 )
510 assert len(res.events) == 1
511 assert res.events[0].event_id == event2.event_id
514def test_event_search_by_circle(sample_community, create_event):
515 """Test that EventSearch only returns events within the given circle."""
516 user, token = generate_user()
518 with events_session(token) as api:
519 inside_pts = [(0.1, 0.01), (0.01, 0.1)]
520 for i, (lat, lng) in enumerate(inside_pts):
521 create_event(
522 api,
523 title=f"Inside area {i}",
524 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),
525 )
527 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]
528 for i, (lat, lng) in enumerate(outside_pts):
529 create_event(
530 api,
531 title=f"Outside area {i}",
532 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),
533 )
535 with search_session(token) as api:
536 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))
537 assert len(res.events) == len(inside_pts)
538 assert all(event.title.startswith("Inside area") for event in res.events)
541def test_event_search_by_rectangle(sample_community, create_event):
542 """Test that EventSearch only returns events within the given rectangular area."""
543 user, token = generate_user()
545 with events_session(token) as api:
546 inside_pts = [(0.1, 0.2), (1.2, 0.2)]
547 for i, (lat, lng) in enumerate(inside_pts):
548 create_event(
549 api,
550 title=f"Inside area {i}",
551 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),
552 )
554 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]
555 for i, (lat, lng) in enumerate(outside_pts):
556 create_event(
557 api,
558 title=f"Outside area {i}",
559 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),
560 )
562 with search_session(token) as api:
563 res = api.EventSearch(
564 search_pb2.EventSearchReq(
565 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)
566 )
567 )
568 assert len(res.events) == len(inside_pts)
569 assert all(event.title.startswith("Inside area") for event in res.events)
572def test_event_search_pagination(sample_community, create_event):
573 """Test that EventSearch paginates correctly.
575 Check that
576 - <page_size> events are returned, if available
577 - sort order is applied (default: past=False)
578 - the next page token continues where the previous page left off
579 """
580 user, token = generate_user()
582 anchor_time = now().replace(second=0, microsecond=0) # Events are created at minute granularity
583 with events_session(token) as api:
584 for i in range(5):
585 create_event(
586 api,
587 title=f"Event {i + 1}",
588 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
589 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),
590 )
592 with search_session(token) as api:
593 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))
594 assert len(res.events) == 4
595 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]
596 assert res.next_page_token
598 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))
599 assert len(res.events) == 1
600 assert res.events[0].title == "Event 5"
601 assert res.next_page_token == ""
603 # move all the events into the past to test past pagination
604 with session_scope() as session:
605 for occurrence in session.execute(select(EventOccurrence)).scalars().all():
606 occurrence.during = TimestamptzRange(
607 occurrence.start_time - timedelta(days=30), occurrence.end_time - timedelta(days=30)
608 )
610 with search_session(token) as api:
611 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2))
612 assert [event.title for event in res.events] == ["Event 5", "Event 4"]
613 assert res.next_page_token
615 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
616 assert [event.title for event in res.events] == ["Event 3", "Event 2"]
617 assert res.next_page_token
619 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
620 assert [event.title for event in res.events] == ["Event 1"]
621 assert res.next_page_token == ""
624def test_event_search_pagination_with_page_number(sample_community, create_event):
625 """Test that EventSearch paginates correctly with page number.
627 Check that
628 - <page_size> events are returned, if available
629 - sort order is applied (default: past=False)
630 - <page_number> is respected
631 - <total_items> is correct
632 """
633 user, token = generate_user()
635 anchor_time = now()
636 with events_session(token) as api:
637 for i in range(5):
638 create_event(
639 api,
640 title=f"Event {i + 1}",
641 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
642 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),
643 )
645 with search_session(token) as api:
646 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
647 assert len(res.events) == 2
648 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
649 assert res.total_items == 5
651 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
652 assert len(res.events) == 2
653 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
654 assert res.total_items == 5
656 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
657 assert len(res.events) == 1
658 assert [event.title for event in res.events] == ["Event 5"]
659 assert res.total_items == 5
661 # Verify no more pages
662 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
663 assert not res.events
664 assert res.total_items == 5
667def test_event_search_filter_subscription_attendance_organizing_my_communities(
668 sample_community, create_event, moderator: Moderator
669):
670 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
671 returns all events.
672 """
673 _, token = generate_user()
674 other_user, other_token = generate_user()
676 with communities_session(token) as api:
677 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
679 with session_scope() as session:
680 create_community(session, 55, 60, "Other community", [other_user], [], None)
682 with events_session(other_token) as api:
683 e_subscribed = create_event(api, title="Subscribed event")
684 e_attending = create_event(api, title="Attending event")
685 create_event(api, title="Community event")
686 create_event(
687 api,
688 title="Other community event",
689 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
690 )
692 # Approve all events so they're visible to other users
693 with session_scope() as session:
694 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
695 for oid in occurrence_ids:
696 moderator.approve_event_occurrence(oid)
698 with events_session(token) as api:
699 create_event(api, title="Organized event")
700 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
701 api.SetEventAttendance(
702 events_pb2.SetEventAttendanceReq(
703 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
704 )
705 )
707 with search_session(token) as api:
708 res = api.EventSearch(search_pb2.EventSearchReq())
709 assert {event.title for event in res.events} == {
710 "Subscribed event",
711 "Attending event",
712 "Community event",
713 "Other community event",
714 "Organized event",
715 }
717 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
718 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
720 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
721 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
723 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
724 assert {event.title for event in res.events} == {"Organized event"}
726 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
727 assert {event.title for event in res.events} == {
728 "Subscribed event",
729 "Attending event",
730 "Community event",
731 "Organized event",
732 }
734 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
735 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}
738def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):
739 """Test that exclude_attending removes events the user is attending or organizing."""
740 user, token = generate_user()
741 other_user, other_token = generate_user()
743 with communities_session(token) as api:
744 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
746 with session_scope() as session:
747 create_community(session, 55, 60, "Other community", [other_user], [], None)
749 with events_session(other_token) as api:
750 e_attending = create_event(api, title="Attending event")
751 e_community_only = create_event(api, title="Community only event")
752 create_event(
753 api,
754 title="Other community event",
755 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
756 )
758 with session_scope() as session:
759 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
760 for oid in occurrence_ids:
761 moderator.approve_event_occurrence(oid)
763 with events_session(token) as api:
764 e_organized = create_event(api, title="Organized event")
765 api.SetEventAttendance(
766 events_pb2.SetEventAttendanceReq(
767 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
768 )
769 )
771 with search_session(token) as api:
772 # baseline: my_communities returns all community events including attended/organized
773 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
774 assert {event.title for event in res.events} == {
775 "Attending event",
776 "Community only event",
777 "Organized event",
778 }
780 # my_communities + exclude_attending: drops attended and organized events
781 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))
782 assert {event.title for event in res.events} == {"Community only event"}
784 # exclude_attending alone (no other filter = all events): drops attended and organized
785 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))
786 assert {event.title for event in res.events} == {"Community only event", "Other community event"}
788 # attending + exclude_attending is invalid
789 with pytest.raises(grpc.RpcError) as e:
790 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))
791 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
794def test_regression_search_multiple_pages(db):
795 """
796 There was a bug when there are multiple pages of results
797 """
798 user, token = generate_user()
799 user_ids = [user.id]
800 for _ in range(10):
801 other_user, _ = generate_user()
802 user_ids.append(other_user.id)
804 refresh_materialized_views_rapid(empty_pb2.Empty())
805 refresh_materialized_views(empty_pb2.Empty())
807 with search_session(token) as api:
808 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))
809 assert [result.user_id for result in res.results] == user_ids[:5]
810 assert res.next_page_token
813def test_regression_search_no_results(db):
814 """
815 There was a bug when there were no results
816 """
817 # put us far away
818 user, token = generate_user()
820 refresh_materialized_views_rapid(empty_pb2.Empty())
821 refresh_materialized_views(empty_pb2.Empty())
823 with search_session(token) as api:
824 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
825 assert len(res.results) == 0
828def test_user_filter_same_gender_only(db):
829 """Test that same_gender_only filter works correctly"""
830 # Create users with different genders and strong verification status
831 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")
832 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")
833 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")
834 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")
835 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")
837 refresh_materialized_views_rapid(empty_pb2.Empty())
838 refresh_materialized_views(empty_pb2.Empty())
840 # Test 1: Woman with strong verification should see only women when same_gender_only=True
841 with search_session(token_woman_with_sv) as api:
842 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
843 result_ids = [result.user.user_id for result in res.results]
844 assert woman_with_sv.id in result_ids
845 assert woman_without_sv.id in result_ids
846 assert other_woman_with_sv.id in result_ids
847 assert man_with_sv.id not in result_ids
848 assert man_without_sv.id not in result_ids
850 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
851 result_ids = [result.user_id for result in res.results]
852 assert woman_with_sv.id in result_ids
853 assert woman_without_sv.id in result_ids
854 assert other_woman_with_sv.id in result_ids
855 assert man_with_sv.id not in result_ids
856 assert man_without_sv.id not in result_ids
858 # Test 2: Man with strong verification should see only men when same_gender_only=True
859 with search_session(token_man_with_sv) as api:
860 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
861 result_ids = [result.user.user_id for result in res.results]
862 assert man_with_sv.id in result_ids
863 assert man_without_sv.id in result_ids
864 assert woman_with_sv.id not in result_ids
865 assert woman_without_sv.id not in result_ids
866 assert other_woman_with_sv.id not in result_ids
868 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
869 result_ids = [result.user_id for result in res.results]
870 assert man_with_sv.id in result_ids
871 assert man_without_sv.id in result_ids
872 assert woman_with_sv.id not in result_ids
873 assert woman_without_sv.id not in result_ids
874 assert other_woman_with_sv.id not in result_ids
876 # Test 3: Woman without strong verification should get an error
877 with search_session(token_woman_without_sv) as api:
878 with pytest.raises(Exception) as e:
879 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
880 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
882 with pytest.raises(Exception) as e:
883 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
884 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
886 # Test 4: When same_gender_only=False, should see all users
887 with search_session(token_woman_with_sv) as api:
888 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))
889 result_ids = [result.user.user_id for result in res.results]
890 assert woman_with_sv.id in result_ids
891 assert woman_without_sv.id in result_ids
892 assert other_woman_with_sv.id in result_ids
893 assert man_with_sv.id in result_ids
894 assert man_without_sv.id in result_ids
896 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))
897 result_ids = [result.user_id for result in res.results]
898 assert woman_with_sv.id in result_ids
899 assert woman_without_sv.id in result_ids
900 assert other_woman_with_sv.id in result_ids
901 assert man_with_sv.id in result_ids
902 assert man_without_sv.id in result_ids
905def test_user_filter_same_gender_only_with_other_filters(db):
906 """Test that same_gender_only filter works correctly combined with other filters"""
907 # Create users with different properties
908 woman_host, token_woman = generate_user(
909 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host
910 )
911 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)
912 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)
914 refresh_materialized_views_rapid(empty_pb2.Empty())
915 refresh_materialized_views(empty_pb2.Empty())
917 # Test: Combine same_gender_only with hosting_status filter
918 with search_session(token_woman) as api:
919 res = api.UserSearch(
920 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
921 )
922 result_ids = [result.user.user_id for result in res.results]
923 # Should only see woman who can host
924 assert woman_host.id in result_ids
925 assert woman_cant_host.id not in result_ids
926 assert man_host.id not in result_ids
928 res = api.UserSearchV2(
929 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
930 )
931 result_ids = [result.user_id for result in res.results]
932 assert woman_host.id in result_ids
933 assert woman_cant_host.id not in result_ids
934 assert man_host.id not in result_ids