Coverage for app/backend/src/tests/test_search.py: 100%
458 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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
27def test_Search(testing_communities):
28 user, token = generate_user()
29 with search_session(token) as api:
30 res = api.Search(
31 search_pb2.SearchReq(
32 query="Country 1, Region 1",
33 include_users=True,
34 include_communities=True,
35 include_groups=True,
36 include_places=True,
37 include_guides=True,
38 )
39 )
40 res = api.Search(
41 search_pb2.SearchReq(
42 query="Country 1, Region 1, Attraction",
43 title_only=True,
44 include_users=True,
45 include_communities=True,
46 include_groups=True,
47 include_places=True,
48 include_guides=True,
49 )
50 )
53def test_UserSearch(testing_communities):
54 """Test that UserSearch returns all users if no filter is set."""
55 user, token = generate_user()
57 refresh_materialized_views_rapid(empty_pb2.Empty())
58 refresh_materialized_views(empty_pb2.Empty())
60 with search_session(token) as api:
61 res = api.UserSearch(search_pb2.UserSearchReq())
62 assert len(res.results) > 0
63 assert res.total_items == len(res.results)
64 res = api.UserSearchV2(search_pb2.UserSearchReq())
65 assert len(res.results) > 0
66 assert res.total_items == len(res.results)
69def test_regression_search_in_area(db):
70 """
71 Makes sure search_in_area works.
73 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
74 """
76 # outside
77 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
78 # outside
79 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
80 # inside
81 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
82 # inside
83 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
84 # outside
85 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
87 refresh_materialized_views_rapid(empty_pb2.Empty())
88 refresh_materialized_views(empty_pb2.Empty())
90 with search_session(token5) as api:
91 res = api.UserSearch(
92 search_pb2.UserSearchReq(
93 search_in_area=search_pb2.Area(
94 lat=0,
95 lng=0,
96 radius=100000,
97 )
98 )
99 )
100 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
102 res = api.UserSearchV2(
103 search_pb2.UserSearchReq(
104 search_in_area=search_pb2.Area(
105 lat=0,
106 lng=0,
107 radius=100000,
108 )
109 )
110 )
111 assert [result.user_id for result in res.results] == [user3.id, user4.id]
114def test_user_search_in_rectangle(db):
115 """
116 Makes sure search_in_rectangle works as expected.
117 """
119 # outside
120 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
121 # outside
122 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
123 # inside
124 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
125 # inside
126 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
127 # outside (not fully inside)
128 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
129 # outside
130 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
131 # outside
132 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
134 refresh_materialized_views_rapid(empty_pb2.Empty())
135 refresh_materialized_views(empty_pb2.Empty())
137 with search_session(token5) as api:
138 res = api.UserSearch(
139 search_pb2.UserSearchReq(
140 search_in_rectangle=search_pb2.RectArea(
141 lat_min=0,
142 lat_max=2,
143 lng_min=0,
144 lng_max=1,
145 )
146 )
147 )
148 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
150 res = api.UserSearchV2(
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_id for result in res.results] == [user3.id, user4.id]
163def test_user_filter_complete_profile(db):
164 """
165 Make sure the completed profile flag returns only completed user profile
166 """
167 user_complete_profile, token6 = generate_user(complete_profile=True)
169 user_incomplete_profile, token7 = generate_user(complete_profile=False)
171 refresh_materialized_views_rapid(empty_pb2.Empty())
172 refresh_materialized_views(empty_pb2.Empty())
174 with search_session(token7) as api:
175 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
176 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
178 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
179 assert user_incomplete_profile.id in [result.user_id for result in res.results]
181 with search_session(token6) as api:
182 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
183 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
185 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
186 assert [result.user_id for result in res.results] == [user_complete_profile.id]
189def test_user_filter_meetup_status(db):
190 """
191 Make sure the completed profile flag returns only completed user profile
192 """
193 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
195 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
197 refresh_materialized_views_rapid(empty_pb2.Empty())
198 refresh_materialized_views(empty_pb2.Empty())
200 with search_session(token8) as api:
201 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
202 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
204 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
205 assert user_wants_to_meetup.id in [result.user_id for result in res.results]
207 with search_session(token9) as api:
208 res = api.UserSearch(
209 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
210 )
211 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
213 res = api.UserSearchV2(
214 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
215 )
216 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]
219def test_user_filter_language(db):
220 """
221 Test filtering users by language ability.
222 """
223 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
224 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
225 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
227 with session_scope() as session:
228 session.add(
229 LanguageAbility(
230 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
231 ),
232 )
233 session.add(
234 LanguageAbility(
235 user_id=user_with_japanese_conversational.id,
236 language_code="jpn",
237 fluency=LanguageFluency.fluent,
238 )
239 )
240 session.add(
241 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
242 )
244 refresh_materialized_views_rapid(empty_pb2.Empty())
245 refresh_materialized_views(empty_pb2.Empty())
247 with search_session(token11) as api:
248 res = api.UserSearch(
249 search_pb2.UserSearchReq(
250 language_ability_filter=[
251 api_pb2.LanguageAbility(
252 code="deu",
253 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
254 )
255 ]
256 )
257 )
258 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
260 res = api.UserSearchV2(
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_id for result in res.results] == [user_with_german_fluent.id]
272 res = api.UserSearch(
273 search_pb2.UserSearchReq(
274 language_ability_filter=[
275 api_pb2.LanguageAbility(
276 code="jpn",
277 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
278 )
279 ]
280 )
281 )
282 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
284 res = api.UserSearchV2(
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_id for result in res.results] == [user_with_japanese_conversational.id]
297def test_user_filter_strong_verification(db):
298 user1, token1 = generate_user()
299 user2, _ = generate_user(strong_verification=True)
300 user3, _ = generate_user()
301 user4, _ = generate_user(strong_verification=True)
302 user5, _ = generate_user(strong_verification=True)
304 refresh_materialized_views_rapid(empty_pb2.Empty())
305 refresh_materialized_views(empty_pb2.Empty())
307 with search_session(token1) as api:
308 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
309 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
311 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))
312 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
314 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
315 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
317 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))
318 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]
321def test_regression_search_only_with_references(db):
322 user1, token1 = generate_user()
323 user2, _ = generate_user()
324 user3, _ = generate_user()
325 user4, _ = generate_user(delete_user=True)
327 refresh_materialized_views_rapid(empty_pb2.Empty())
328 refresh_materialized_views(empty_pb2.Empty())
330 with session_scope() as session:
331 # user 2 has references
332 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
333 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
334 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
336 # user 3 only has reference from a deleted user
337 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
339 with search_session(token1) as api:
340 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
341 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
343 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))
344 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]
346 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
347 assert [result.user.user_id for result in res.results] == [user2.id]
349 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
350 assert [result.user_id for result in res.results] == [user2.id]
353def test_user_search_exactly_user_ids(db):
354 """
355 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
356 """
357 # Create users with different properties
358 user1, token1 = generate_user()
359 user2, _ = generate_user(strong_verification=True)
360 user3, _ = generate_user(complete_profile=True)
361 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
362 user5, _ = generate_user(delete_user=True) # Deleted user
364 refresh_materialized_views_rapid(empty_pb2.Empty())
365 refresh_materialized_views(empty_pb2.Empty())
367 with search_session(token1) as api:
368 # Test that exactly_user_ids returns only the specified users
369 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
370 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
372 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
373 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
375 # Test that exactly_user_ids ignores other filters
376 res = api.UserSearch(
377 search_pb2.UserSearchReq(
378 exactly_user_ids=[user2.id, user3.id, user4.id],
379 only_with_strong_verification=True, # This would normally filter out user3 and user4
380 )
381 )
382 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
384 res = api.UserSearchV2(
385 search_pb2.UserSearchReq(
386 exactly_user_ids=[user2.id, user3.id, user4.id],
387 only_with_strong_verification=True, # This would normally filter out user3 and user4
388 )
389 )
390 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
392 # Test with non-existent user IDs (should be ignored)
393 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
394 assert [result.user.user_id for result in res.results] == [user1.id]
396 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
397 assert [result.user_id for result in res.results] == [user1.id]
399 # Test with deleted user ID (should be ignored due to visibility filter)
400 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
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, user5.id]))
404 assert [result.user_id for result in res.results] == [user1.id]
407@pytest.fixture
408def sample_event_data() -> dict[str, Any]:
409 """Dummy data for creating events."""
410 start_time = now() + timedelta(hours=2)
411 end_time = start_time + timedelta(hours=3)
412 return {
413 "title": "Dummy Title",
414 "content": "Dummy content.",
415 "photo_key": None,
416 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),
417 "start_datetime_iso8601_local": datetime_to_iso8601_local(start_time),
418 "end_datetime_iso8601_local": datetime_to_iso8601_local(end_time),
419 }
422@pytest.fixture
423def create_event(sample_event_data):
424 """Factory for creating events."""
426 def _create_event(event_api, **kwargs) -> EventOccurrence:
427 """Create an event with default values, unless overridden by kwargs."""
428 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore
430 return _create_event
433@pytest.fixture
434def sample_community(db) -> int:
435 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""
436 user, _ = generate_user()
437 with session_scope() as session:
438 return create_community(session, -50, 50, "Community", [user], [], None).id
441def test_EventSearch_no_filters(testing_communities):
442 """Test that EventSearch returns all events if no filter is set."""
443 user, token = generate_user()
444 with search_session(token) as api:
445 res = api.EventSearch(search_pb2.EventSearchReq())
446 assert len(res.events) > 0
449def test_event_search_by_query(sample_community, create_event):
450 """Test that EventSearch finds events by title (and content if query_title_only=False)."""
451 user, token = generate_user()
453 with events_session(token) as api:
454 event1 = create_event(api, title="Lorem Ipsum")
455 event2 = create_event(api, content="Lorem Ipsum")
456 create_event(api)
458 with search_session(token) as api:
459 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))
460 assert len(res.events) == 2
461 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
463 res = api.EventSearch(
464 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)
465 )
466 assert len(res.events) == 1
467 assert res.events[0].event_id == event1.event_id
470def test_event_search_by_time(sample_community, create_event):
471 """Test that EventSearch filters with the given time range."""
472 user, token = generate_user()
474 with events_session(token) as api:
475 event1 = create_event(
476 api,
477 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=1)),
478 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=2)),
479 )
480 event2 = create_event(
481 api,
482 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=4)),
483 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=5)),
484 )
485 event3 = create_event(
486 api,
487 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=7)),
488 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=8)),
489 )
491 with search_session(token) as api:
492 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))
493 assert len(res.events) == 2
494 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}
496 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))
497 assert len(res.events) == 2
498 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}
500 res = api.EventSearch(
501 search_pb2.EventSearchReq(
502 before=Timestamp_from_datetime(now() + timedelta(hours=6)),
503 after=Timestamp_from_datetime(now() + timedelta(hours=3)),
504 )
505 )
506 assert len(res.events) == 1
507 assert res.events[0].event_id == event2.event_id
510def test_event_search_by_circle(sample_community, create_event):
511 """Test that EventSearch only returns events within the given circle."""
512 user, token = generate_user()
514 with events_session(token) as api:
515 inside_pts = [(0.1, 0.01), (0.01, 0.1)]
516 for i, (lat, lng) in enumerate(inside_pts):
517 create_event(
518 api,
519 title=f"Inside area {i}",
520 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),
521 )
523 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]
524 for i, (lat, lng) in enumerate(outside_pts):
525 create_event(
526 api,
527 title=f"Outside area {i}",
528 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),
529 )
531 with search_session(token) as api:
532 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))
533 assert len(res.events) == len(inside_pts)
534 assert all(event.title.startswith("Inside area") for event in res.events)
537def test_event_search_by_rectangle(sample_community, create_event):
538 """Test that EventSearch only returns events within the given rectangular area."""
539 user, token = generate_user()
541 with events_session(token) as api:
542 inside_pts = [(0.1, 0.2), (1.2, 0.2)]
543 for i, (lat, lng) in enumerate(inside_pts):
544 create_event(
545 api,
546 title=f"Inside area {i}",
547 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),
548 )
550 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]
551 for i, (lat, lng) in enumerate(outside_pts):
552 create_event(
553 api,
554 title=f"Outside area {i}",
555 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),
556 )
558 with search_session(token) as api:
559 res = api.EventSearch(
560 search_pb2.EventSearchReq(
561 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)
562 )
563 )
564 assert len(res.events) == len(inside_pts)
565 assert all(event.title.startswith("Inside area") for event in res.events)
568def test_event_search_pagination(sample_community, create_event):
569 """Test that EventSearch paginates correctly.
571 Check that
572 - <page_size> events are returned, if available
573 - sort order is applied (default: past=False)
574 - the next page token continues where the previous page left off
575 """
576 user, token = generate_user()
578 anchor_time = now().replace(second=0, microsecond=0) # Events are created at minute granularity
579 with events_session(token) as api:
580 for i in range(5):
581 create_event(
582 api,
583 title=f"Event {i + 1}",
584 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
585 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),
586 )
588 with search_session(token) as api:
589 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))
590 assert len(res.events) == 4
591 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]
592 assert res.next_page_token
594 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))
595 assert len(res.events) == 1
596 assert res.events[0].title == "Event 5"
597 assert res.next_page_token == ""
599 # move all the events into the past to test past pagination
600 with session_scope() as session:
601 for occurrence in session.execute(select(EventOccurrence)).scalars().all():
602 occurrence.during = TimestamptzRange(
603 occurrence.start_time - timedelta(days=30), occurrence.end_time - timedelta(days=30)
604 )
606 with search_session(token) as api:
607 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2))
608 assert [event.title for event in res.events] == ["Event 5", "Event 4"]
609 assert res.next_page_token
611 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
612 assert [event.title for event in res.events] == ["Event 3", "Event 2"]
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 1"]
617 assert res.next_page_token == ""
620def test_event_search_pagination_with_page_number(sample_community, create_event):
621 """Test that EventSearch paginates correctly with page number.
623 Check that
624 - <page_size> events are returned, if available
625 - sort order is applied (default: past=False)
626 - <page_number> is respected
627 - <total_items> is correct
628 """
629 user, token = generate_user()
631 anchor_time = now()
632 with events_session(token) as api:
633 for i in range(5):
634 create_event(
635 api,
636 title=f"Event {i + 1}",
637 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),
638 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),
639 )
641 with search_session(token) as api:
642 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
643 assert len(res.events) == 2
644 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
645 assert res.total_items == 5
647 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
648 assert len(res.events) == 2
649 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
650 assert res.total_items == 5
652 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
653 assert len(res.events) == 1
654 assert [event.title for event in res.events] == ["Event 5"]
655 assert res.total_items == 5
657 # Verify no more pages
658 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
659 assert not res.events
660 assert res.total_items == 5
663def test_event_search_filter_subscription_attendance_organizing_my_communities(
664 sample_community, create_event, moderator: Moderator
665):
666 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
667 returns all events.
668 """
669 _, token = generate_user()
670 other_user, other_token = generate_user()
672 with communities_session(token) as api:
673 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
675 with session_scope() as session:
676 create_community(session, 55, 60, "Other community", [other_user], [], None)
678 with events_session(other_token) as api:
679 e_subscribed = create_event(api, title="Subscribed event")
680 e_attending = create_event(api, title="Attending event")
681 create_event(api, title="Community event")
682 create_event(
683 api,
684 title="Other community event",
685 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
686 )
688 # Approve all events so they're visible to other users
689 with session_scope() as session:
690 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
691 for oid in occurrence_ids:
692 moderator.approve_event_occurrence(oid)
694 with events_session(token) as api:
695 create_event(api, title="Organized event")
696 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
697 api.SetEventAttendance(
698 events_pb2.SetEventAttendanceReq(
699 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
700 )
701 )
703 with search_session(token) as api:
704 res = api.EventSearch(search_pb2.EventSearchReq())
705 assert {event.title for event in res.events} == {
706 "Subscribed event",
707 "Attending event",
708 "Community event",
709 "Other community event",
710 "Organized event",
711 }
713 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
714 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
716 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
717 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
719 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
720 assert {event.title for event in res.events} == {"Organized event"}
722 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
723 assert {event.title for event in res.events} == {
724 "Subscribed event",
725 "Attending event",
726 "Community event",
727 "Organized event",
728 }
730 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
731 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}
734def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):
735 """Test that exclude_attending removes events the user is attending or organizing."""
736 user, token = generate_user()
737 other_user, other_token = generate_user()
739 with communities_session(token) as api:
740 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
742 with session_scope() as session:
743 create_community(session, 55, 60, "Other community", [other_user], [], None)
745 with events_session(other_token) as api:
746 e_attending = create_event(api, title="Attending event")
747 e_community_only = create_event(api, title="Community only event")
748 create_event(
749 api,
750 title="Other community event",
751 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
752 )
754 with session_scope() as session:
755 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
756 for oid in occurrence_ids:
757 moderator.approve_event_occurrence(oid)
759 with events_session(token) as api:
760 e_organized = create_event(api, title="Organized event")
761 api.SetEventAttendance(
762 events_pb2.SetEventAttendanceReq(
763 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
764 )
765 )
767 with search_session(token) as api:
768 # baseline: my_communities returns all community events including attended/organized
769 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
770 assert {event.title for event in res.events} == {
771 "Attending event",
772 "Community only event",
773 "Organized event",
774 }
776 # my_communities + exclude_attending: drops attended and organized events
777 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))
778 assert {event.title for event in res.events} == {"Community only event"}
780 # exclude_attending alone (no other filter = all events): drops attended and organized
781 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))
782 assert {event.title for event in res.events} == {"Community only event", "Other community event"}
784 # attending + exclude_attending is invalid
785 with pytest.raises(grpc.RpcError) as e:
786 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))
787 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
790def test_regression_search_multiple_pages(db):
791 """
792 There was a bug when there are multiple pages of results
793 """
794 user, token = generate_user()
795 user_ids = [user.id]
796 for _ in range(10):
797 other_user, _ = generate_user()
798 user_ids.append(other_user.id)
800 refresh_materialized_views_rapid(empty_pb2.Empty())
801 refresh_materialized_views(empty_pb2.Empty())
803 with search_session(token) as api:
804 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))
805 assert [result.user_id for result in res.results] == user_ids[:5]
806 assert res.next_page_token
809def test_regression_search_no_results(db):
810 """
811 There was a bug when there were no results
812 """
813 # put us far away
814 user, token = generate_user()
816 refresh_materialized_views_rapid(empty_pb2.Empty())
817 refresh_materialized_views(empty_pb2.Empty())
819 with search_session(token) as api:
820 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
821 assert len(res.results) == 0
824def test_user_filter_same_gender_only(db):
825 """Test that same_gender_only filter works correctly"""
826 # Create users with different genders and strong verification status
827 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")
828 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")
829 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")
830 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")
831 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")
833 refresh_materialized_views_rapid(empty_pb2.Empty())
834 refresh_materialized_views(empty_pb2.Empty())
836 # Test 1: Woman with strong verification should see only women when same_gender_only=True
837 with search_session(token_woman_with_sv) as api:
838 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
839 result_ids = [result.user.user_id for result in res.results]
840 assert woman_with_sv.id in result_ids
841 assert woman_without_sv.id in result_ids
842 assert other_woman_with_sv.id in result_ids
843 assert man_with_sv.id not in result_ids
844 assert man_without_sv.id not in result_ids
846 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
847 result_ids = [result.user_id for result in res.results]
848 assert woman_with_sv.id in result_ids
849 assert woman_without_sv.id in result_ids
850 assert other_woman_with_sv.id in result_ids
851 assert man_with_sv.id not in result_ids
852 assert man_without_sv.id not in result_ids
854 # Test 2: Man with strong verification should see only men when same_gender_only=True
855 with search_session(token_man_with_sv) as api:
856 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
857 result_ids = [result.user.user_id for result in res.results]
858 assert man_with_sv.id in result_ids
859 assert man_without_sv.id in result_ids
860 assert woman_with_sv.id not in result_ids
861 assert woman_without_sv.id not in result_ids
862 assert other_woman_with_sv.id not in result_ids
864 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
865 result_ids = [result.user_id for result in res.results]
866 assert man_with_sv.id in result_ids
867 assert man_without_sv.id in result_ids
868 assert woman_with_sv.id not in result_ids
869 assert woman_without_sv.id not in result_ids
870 assert other_woman_with_sv.id not in result_ids
872 # Test 3: Woman without strong verification should get an error
873 with search_session(token_woman_without_sv) as api:
874 with pytest.raises(Exception) as e:
875 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
876 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
878 with pytest.raises(Exception) as e:
879 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
880 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
882 # Test 4: When same_gender_only=False, should see all users
883 with search_session(token_woman_with_sv) as api:
884 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))
885 result_ids = [result.user.user_id for result in res.results]
886 assert woman_with_sv.id in result_ids
887 assert woman_without_sv.id in result_ids
888 assert other_woman_with_sv.id in result_ids
889 assert man_with_sv.id in result_ids
890 assert man_without_sv.id in result_ids
892 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))
893 result_ids = [result.user_id for result in res.results]
894 assert woman_with_sv.id in result_ids
895 assert woman_without_sv.id in result_ids
896 assert other_woman_with_sv.id in result_ids
897 assert man_with_sv.id in result_ids
898 assert man_without_sv.id in result_ids
901def test_user_filter_same_gender_only_with_other_filters(db):
902 """Test that same_gender_only filter works correctly combined with other filters"""
903 # Create users with different properties
904 woman_host, token_woman = generate_user(
905 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host
906 )
907 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)
908 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)
910 refresh_materialized_views_rapid(empty_pb2.Empty())
911 refresh_materialized_views(empty_pb2.Empty())
913 # Test: Combine same_gender_only with hosting_status filter
914 with search_session(token_woman) as api:
915 res = api.UserSearch(
916 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
917 )
918 result_ids = [result.user.user_id for result in res.results]
919 # Should only see woman who can host
920 assert woman_host.id in result_ids
921 assert woman_cant_host.id not in result_ids
922 assert man_host.id not in result_ids
924 res = api.UserSearchV2(
925 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
926 )
927 result_ids = [result.user_id for result in res.results]
928 assert woman_host.id in result_ids
929 assert woman_cant_host.id not in result_ids
930 assert man_host.id not in result_ids