Coverage for app/backend/src/tests/test_search.py: 100%
459 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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
22class TestSearchInCommunities:
23 """The tests that search the whole community tree, grouped so they share one copy of it."""
25 @staticmethod
26 def test_Search(testing_communities):
27 user, token = generate_user()
28 with search_session(token) as api:
29 res = api.Search(
30 search_pb2.SearchReq(
31 query="Country 1, Region 1",
32 include_users=True,
33 include_communities=True,
34 include_groups=True,
35 include_places=True,
36 include_guides=True,
37 )
38 )
39 res = api.Search(
40 search_pb2.SearchReq(
41 query="Country 1, Region 1, Attraction",
42 title_only=True,
43 include_users=True,
44 include_communities=True,
45 include_groups=True,
46 include_places=True,
47 include_guides=True,
48 )
49 )
51 @staticmethod
52 def test_UserSearch(testing_communities):
53 """Test that UserSearch returns all users if no filter is set."""
54 user, token = generate_user()
56 refresh_materialized_views_rapid(empty_pb2.Empty())
57 refresh_materialized_views(empty_pb2.Empty())
59 with search_session(token) as api:
60 res = api.UserSearch(search_pb2.UserSearchReq())
61 assert len(res.results) > 0
62 assert res.total_items == len(res.results)
63 res = api.UserSearchV2(search_pb2.UserSearchReq())
64 assert len(res.results) > 0
65 assert res.total_items == len(res.results)
67 @staticmethod
68 def test_EventSearch_no_filters(testing_communities):
69 """Test that EventSearch returns all events if no filter is set."""
70 user, token = generate_user()
71 with search_session(token) as api:
72 res = api.EventSearch(search_pb2.EventSearchReq())
73 assert len(res.events) > 0
76def test_regression_search_in_area(db):
77 """
78 Makes sure search_in_area works.
80 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
81 """
83 # outside
84 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
85 # outside
86 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
87 # inside
88 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
89 # inside
90 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
91 # outside
92 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
94 refresh_materialized_views_rapid(empty_pb2.Empty())
95 refresh_materialized_views(empty_pb2.Empty())
97 with search_session(token5) as api:
98 res = api.UserSearch(
99 search_pb2.UserSearchReq(
100 search_in_area=search_pb2.Area(
101 lat=0,
102 lng=0,
103 radius=100000,
104 )
105 )
106 )
107 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
109 res = api.UserSearchV2(
110 search_pb2.UserSearchReq(
111 search_in_area=search_pb2.Area(
112 lat=0,
113 lng=0,
114 radius=100000,
115 )
116 )
117 )
118 assert [result.user_id for result in res.results] == [user3.id, user4.id]
121def test_user_search_in_rectangle(db):
122 """
123 Makes sure search_in_rectangle works as expected.
124 """
126 # outside
127 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
128 # outside
129 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
130 # inside
131 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
132 # inside
133 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
134 # outside (not fully inside)
135 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
136 # outside
137 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
138 # outside
139 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
141 refresh_materialized_views_rapid(empty_pb2.Empty())
142 refresh_materialized_views(empty_pb2.Empty())
144 with search_session(token5) as api:
145 res = api.UserSearch(
146 search_pb2.UserSearchReq(
147 search_in_rectangle=search_pb2.RectArea(
148 lat_min=0,
149 lat_max=2,
150 lng_min=0,
151 lng_max=1,
152 )
153 )
154 )
155 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
157 res = api.UserSearchV2(
158 search_pb2.UserSearchReq(
159 search_in_rectangle=search_pb2.RectArea(
160 lat_min=0,
161 lat_max=2,
162 lng_min=0,
163 lng_max=1,
164 )
165 )
166 )
167 assert [result.user_id for result in res.results] == [user3.id, user4.id]
170def test_user_filter_complete_profile(db):
171 """
172 Make sure the completed profile flag returns only completed user profile
173 """
174 user_complete_profile, token6 = generate_user(complete_profile=True)
176 user_incomplete_profile, token7 = generate_user(complete_profile=False)
178 refresh_materialized_views_rapid(empty_pb2.Empty())
179 refresh_materialized_views(empty_pb2.Empty())
181 with search_session(token7) as api:
182 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
183 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
185 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
186 assert user_incomplete_profile.id in [result.user_id for result in res.results]
188 with search_session(token6) as api:
189 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
190 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
192 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
193 assert [result.user_id for result in res.results] == [user_complete_profile.id]
196def test_user_filter_meetup_status(db):
197 """
198 Make sure the completed profile flag returns only completed user profile
199 """
200 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
202 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
204 refresh_materialized_views_rapid(empty_pb2.Empty())
205 refresh_materialized_views(empty_pb2.Empty())
207 with search_session(token8) as api:
208 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
209 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
211 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
212 assert user_wants_to_meetup.id in [result.user_id for result in res.results]
214 with search_session(token9) as api:
215 res = api.UserSearch(
216 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
217 )
218 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
220 res = api.UserSearchV2(
221 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
222 )
223 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]
226def test_user_filter_language(db):
227 """
228 Test filtering users by language ability.
229 """
230 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
231 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
232 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
234 with session_scope() as session:
235 session.add(
236 LanguageAbility(
237 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
238 ),
239 )
240 session.add(
241 LanguageAbility(
242 user_id=user_with_japanese_conversational.id,
243 language_code="jpn",
244 fluency=LanguageFluency.fluent,
245 )
246 )
247 session.add(
248 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
249 )
251 refresh_materialized_views_rapid(empty_pb2.Empty())
252 refresh_materialized_views(empty_pb2.Empty())
254 with search_session(token11) as api:
255 res = api.UserSearch(
256 search_pb2.UserSearchReq(
257 language_ability_filter=[
258 api_pb2.LanguageAbility(
259 code="deu",
260 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
261 )
262 ]
263 )
264 )
265 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
267 res = api.UserSearchV2(
268 search_pb2.UserSearchReq(
269 language_ability_filter=[
270 api_pb2.LanguageAbility(
271 code="deu",
272 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
273 )
274 ]
275 )
276 )
277 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]
279 res = api.UserSearch(
280 search_pb2.UserSearchReq(
281 language_ability_filter=[
282 api_pb2.LanguageAbility(
283 code="jpn",
284 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
285 )
286 ]
287 )
288 )
289 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
291 res = api.UserSearchV2(
292 search_pb2.UserSearchReq(
293 language_ability_filter=[
294 api_pb2.LanguageAbility(
295 code="jpn",
296 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
297 )
298 ]
299 )
300 )
301 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]
304def test_user_filter_strong_verification(db):
305 user1, token1 = generate_user()
306 user2, _ = generate_user(strong_verification=True)
307 user3, _ = generate_user()
308 user4, _ = generate_user(strong_verification=True)
309 user5, _ = generate_user(strong_verification=True)
311 refresh_materialized_views_rapid(empty_pb2.Empty())
312 refresh_materialized_views(empty_pb2.Empty())
314 with search_session(token1) as api:
315 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
316 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
318 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))
319 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
321 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
322 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
324 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))
325 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]
328def test_regression_search_only_with_references(db):
329 user1, token1 = generate_user()
330 user2, _ = generate_user()
331 user3, _ = generate_user()
332 user4, _ = generate_user(delete_user=True)
334 refresh_materialized_views_rapid(empty_pb2.Empty())
335 refresh_materialized_views(empty_pb2.Empty())
337 with session_scope() as session:
338 # user 2 has references
339 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
340 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
341 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
343 # user 3 only has reference from a deleted user
344 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
346 with search_session(token1) as api:
347 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
348 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
350 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))
351 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]
353 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
354 assert [result.user.user_id for result in res.results] == [user2.id]
356 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
357 assert [result.user_id for result in res.results] == [user2.id]
360def test_user_search_exactly_user_ids(db):
361 """
362 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
363 """
364 # Create users with different properties
365 user1, token1 = generate_user()
366 user2, _ = generate_user(strong_verification=True)
367 user3, _ = generate_user(complete_profile=True)
368 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
369 user5, _ = generate_user(delete_user=True) # Deleted user
371 refresh_materialized_views_rapid(empty_pb2.Empty())
372 refresh_materialized_views(empty_pb2.Empty())
374 with search_session(token1) as api:
375 # Test that exactly_user_ids returns only the specified users
376 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
377 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
379 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
380 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
382 # Test that exactly_user_ids ignores other filters
383 res = api.UserSearch(
384 search_pb2.UserSearchReq(
385 exactly_user_ids=[user2.id, user3.id, user4.id],
386 only_with_strong_verification=True, # This would normally filter out user3 and user4
387 )
388 )
389 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
391 res = api.UserSearchV2(
392 search_pb2.UserSearchReq(
393 exactly_user_ids=[user2.id, user3.id, user4.id],
394 only_with_strong_verification=True, # This would normally filter out user3 and user4
395 )
396 )
397 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
399 # Test with non-existent user IDs (should be ignored)
400 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
401 assert [result.user.user_id for result in res.results] == [user1.id]
403 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
404 assert [result.user_id for result in res.results] == [user1.id]
406 # Test with deleted user ID (should be ignored due to visibility filter)
407 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
408 assert [result.user.user_id for result in res.results] == [user1.id]
410 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
411 assert [result.user_id for result in res.results] == [user1.id]
414@pytest.fixture
415def sample_event_data() -> dict[str, Any]:
416 """Dummy data for creating events."""
417 start_time = now() + timedelta(hours=2)
418 end_time = start_time + timedelta(hours=3)
419 return {
420 "title": "Dummy Title",
421 "content": "Dummy content.",
422 "photo_key": None,
423 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),
424 "start_datetime_iso8601_local": datetime_to_iso8601_local(start_time),
425 "end_datetime_iso8601_local": datetime_to_iso8601_local(end_time),
426 }
429@pytest.fixture
430def create_event(sample_event_data):
431 """Factory for creating events."""
433 def _create_event(event_api, **kwargs) -> EventOccurrence:
434 """Create an event with default values, unless overridden by kwargs."""
435 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore
437 return _create_event
440@pytest.fixture
441def sample_community(db) -> int:
442 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""
443 user, _ = generate_user()
444 with session_scope() as session:
445 return create_community(session, -50, 50, "Community", [user], [], None).id
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_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=1)),
477 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=2)),
478 )
479 event2 = create_event(
480 api,
481 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=4)),
482 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=5)),
483 )
484 event3 = create_event(
485 api,
486 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=7)),
487 end_datetime_iso8601_local=datetime_to_iso8601_local(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 location=events_pb2.EventLocation(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 location=events_pb2.EventLocation(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 location=events_pb2.EventLocation(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 location=events_pb2.EventLocation(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 continues where the previous page left off
574 """
575 user, token = generate_user()
577 anchor_time = now().replace(second=0, microsecond=0) # Events are created at minute granularity
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_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
584 end_datetime_iso8601_local=datetime_to_iso8601_local(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
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 # move all the events into the past to test past pagination
599 with session_scope() as session:
600 for occurrence in session.execute(select(EventOccurrence)).scalars().all():
601 occurrence.during = TimestamptzRange(
602 occurrence.start_time - timedelta(days=30), occurrence.end_time - timedelta(days=30)
603 )
605 with search_session(token) as api:
606 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2))
607 assert [event.title for event in res.events] == ["Event 5", "Event 4"]
608 assert res.next_page_token
610 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
611 assert [event.title for event in res.events] == ["Event 3", "Event 2"]
612 assert res.next_page_token
614 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
615 assert [event.title for event in res.events] == ["Event 1"]
616 assert res.next_page_token == ""
619def test_event_search_pagination_with_page_number(sample_community, create_event):
620 """Test that EventSearch paginates correctly with page number.
622 Check that
623 - <page_size> events are returned, if available
624 - sort order is applied (default: past=False)
625 - <page_number> is respected
626 - <total_items> is correct
627 """
628 user, token = generate_user()
630 anchor_time = now()
631 with events_session(token) as api:
632 for i in range(5):
633 create_event(
634 api,
635 title=f"Event {i + 1}",
636 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
637 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),
638 )
640 with search_session(token) as api:
641 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
642 assert len(res.events) == 2
643 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
644 assert res.total_items == 5
646 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
647 assert len(res.events) == 2
648 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
649 assert res.total_items == 5
651 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
652 assert len(res.events) == 1
653 assert [event.title for event in res.events] == ["Event 5"]
654 assert res.total_items == 5
656 # Verify no more pages
657 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
658 assert not res.events
659 assert res.total_items == 5
662def test_event_search_filter_subscription_attendance_organizing_my_communities(
663 sample_community, create_event, moderator: Moderator
664):
665 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
666 returns all events.
667 """
668 _, token = generate_user()
669 other_user, other_token = generate_user()
671 with communities_session(token) as api:
672 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
674 with session_scope() as session:
675 create_community(session, 55, 60, "Other community", [other_user], [], None)
677 with events_session(other_token) as api:
678 e_subscribed = create_event(api, title="Subscribed event")
679 e_attending = create_event(api, title="Attending event")
680 create_event(api, title="Community event")
681 create_event(
682 api,
683 title="Other community event",
684 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
685 )
687 # Approve all events so they're visible to other users
688 with session_scope() as session:
689 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
690 for oid in occurrence_ids:
691 moderator.approve_event_occurrence(oid)
693 with events_session(token) as api:
694 create_event(api, title="Organized event")
695 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
696 api.SetEventAttendance(
697 events_pb2.SetEventAttendanceReq(
698 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
699 )
700 )
702 with search_session(token) as api:
703 res = api.EventSearch(search_pb2.EventSearchReq())
704 assert {event.title for event in res.events} == {
705 "Subscribed event",
706 "Attending event",
707 "Community event",
708 "Other community event",
709 "Organized event",
710 }
712 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
713 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
715 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
716 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
718 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
719 assert {event.title for event in res.events} == {"Organized event"}
721 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
722 assert {event.title for event in res.events} == {
723 "Subscribed event",
724 "Attending event",
725 "Community event",
726 "Organized event",
727 }
729 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
730 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}
733def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):
734 """Test that exclude_attending removes events the user is attending or organizing."""
735 user, token = generate_user()
736 other_user, other_token = generate_user()
738 with communities_session(token) as api:
739 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
741 with session_scope() as session:
742 create_community(session, 55, 60, "Other community", [other_user], [], None)
744 with events_session(other_token) as api:
745 e_attending = create_event(api, title="Attending event")
746 e_community_only = create_event(api, title="Community only event")
747 create_event(
748 api,
749 title="Other community event",
750 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
751 )
753 with session_scope() as session:
754 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
755 for oid in occurrence_ids:
756 moderator.approve_event_occurrence(oid)
758 with events_session(token) as api:
759 e_organized = create_event(api, title="Organized event")
760 api.SetEventAttendance(
761 events_pb2.SetEventAttendanceReq(
762 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
763 )
764 )
766 with search_session(token) as api:
767 # baseline: my_communities returns all community events including attended/organized
768 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
769 assert {event.title for event in res.events} == {
770 "Attending event",
771 "Community only event",
772 "Organized event",
773 }
775 # my_communities + exclude_attending: drops attended and organized events
776 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))
777 assert {event.title for event in res.events} == {"Community only event"}
779 # exclude_attending alone (no other filter = all events): drops attended and organized
780 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))
781 assert {event.title for event in res.events} == {"Community only event", "Other community event"}
783 # attending + exclude_attending is invalid
784 with pytest.raises(grpc.RpcError) as e:
785 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))
786 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
789def test_regression_search_multiple_pages(db):
790 """
791 There was a bug when there are multiple pages of results
792 """
793 user, token = generate_user()
794 user_ids = [user.id]
795 for _ in range(10):
796 other_user, _ = generate_user()
797 user_ids.append(other_user.id)
799 refresh_materialized_views_rapid(empty_pb2.Empty())
800 refresh_materialized_views(empty_pb2.Empty())
802 with search_session(token) as api:
803 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))
804 assert [result.user_id for result in res.results] == user_ids[:5]
805 assert res.next_page_token
808def test_regression_search_no_results(db):
809 """
810 There was a bug when there were no results
811 """
812 # put us far away
813 user, token = generate_user()
815 refresh_materialized_views_rapid(empty_pb2.Empty())
816 refresh_materialized_views(empty_pb2.Empty())
818 with search_session(token) as api:
819 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
820 assert len(res.results) == 0
823def test_user_filter_same_gender_only(db):
824 """Test that same_gender_only filter works correctly"""
825 # Create users with different genders and strong verification status
826 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")
827 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")
828 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")
829 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")
830 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")
832 refresh_materialized_views_rapid(empty_pb2.Empty())
833 refresh_materialized_views(empty_pb2.Empty())
835 # Test 1: Woman with strong verification should see only women when same_gender_only=True
836 with search_session(token_woman_with_sv) as api:
837 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
838 result_ids = [result.user.user_id for result in res.results]
839 assert woman_with_sv.id in result_ids
840 assert woman_without_sv.id in result_ids
841 assert other_woman_with_sv.id in result_ids
842 assert man_with_sv.id not in result_ids
843 assert man_without_sv.id not in result_ids
845 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
846 result_ids = [result.user_id for result in res.results]
847 assert woman_with_sv.id in result_ids
848 assert woman_without_sv.id in result_ids
849 assert other_woman_with_sv.id in result_ids
850 assert man_with_sv.id not in result_ids
851 assert man_without_sv.id not in result_ids
853 # Test 2: Man with strong verification should see only men when same_gender_only=True
854 with search_session(token_man_with_sv) as api:
855 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
856 result_ids = [result.user.user_id for result in res.results]
857 assert man_with_sv.id in result_ids
858 assert man_without_sv.id in result_ids
859 assert woman_with_sv.id not in result_ids
860 assert woman_without_sv.id not in result_ids
861 assert other_woman_with_sv.id not in result_ids
863 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
864 result_ids = [result.user_id for result in res.results]
865 assert man_with_sv.id in result_ids
866 assert man_without_sv.id in result_ids
867 assert woman_with_sv.id not in result_ids
868 assert woman_without_sv.id not in result_ids
869 assert other_woman_with_sv.id not in result_ids
871 # Test 3: Woman without strong verification should get an error
872 with search_session(token_woman_without_sv) as api:
873 with pytest.raises(Exception) as e:
874 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
875 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
877 with pytest.raises(Exception) as e:
878 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
879 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
881 # Test 4: When same_gender_only=False, should see all users
882 with search_session(token_woman_with_sv) as api:
883 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))
884 result_ids = [result.user.user_id for result in res.results]
885 assert woman_with_sv.id in result_ids
886 assert woman_without_sv.id in result_ids
887 assert other_woman_with_sv.id in result_ids
888 assert man_with_sv.id in result_ids
889 assert man_without_sv.id in result_ids
891 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))
892 result_ids = [result.user_id for result in res.results]
893 assert woman_with_sv.id in result_ids
894 assert woman_without_sv.id in result_ids
895 assert other_woman_with_sv.id in result_ids
896 assert man_with_sv.id in result_ids
897 assert man_without_sv.id in result_ids
900def test_user_filter_same_gender_only_with_other_filters(db):
901 """Test that same_gender_only filter works correctly combined with other filters"""
902 # Create users with different properties
903 woman_host, token_woman = generate_user(
904 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host
905 )
906 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)
907 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)
909 refresh_materialized_views_rapid(empty_pb2.Empty())
910 refresh_materialized_views(empty_pb2.Empty())
912 # Test: Combine same_gender_only with hosting_status filter
913 with search_session(token_woman) as api:
914 res = api.UserSearch(
915 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
916 )
917 result_ids = [result.user.user_id for result in res.results]
918 # Should only see woman who can host
919 assert woman_host.id in result_ids
920 assert woman_cant_host.id not in result_ids
921 assert man_host.id not in result_ids
923 res = api.UserSearchV2(
924 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
925 )
926 result_ids = [result.user_id for result in res.results]
927 assert woman_host.id in result_ids
928 assert woman_cant_host.id not in result_ids
929 assert man_host.id not in result_ids