Coverage for app/backend/src/tests/test_search.py: 100%
452 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-21 21:52 +0000
1from datetime import timedelta
2from typing import Any
4import grpc
5import pytest
6from google.protobuf import empty_pb2, wrappers_pb2
7from sqlalchemy import select
9from couchers.db import session_scope
10from couchers.materialized_views import refresh_materialized_views, refresh_materialized_views_rapid
11from couchers.models import EventOccurrence, HostingStatus, LanguageAbility, LanguageFluency, MeetupStatus
12from couchers.proto import api_pb2, communities_pb2, events_pb2, search_pb2
13from couchers.utils import Timestamp_from_datetime, create_coordinate, millis_from_dt, now
14from tests.fixtures.db import generate_user
15from tests.fixtures.misc import Moderator
16from tests.fixtures.sessions import communities_session, events_session, search_session
17from tests.test_communities import create_community, testing_communities # noqa
18from tests.test_references import create_friend_reference
21@pytest.fixture(autouse=True)
22def _(testconfig):
23 pass
26def 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 )
52def 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)
68def test_regression_search_in_area(db):
69 """
70 Makes sure search_in_area works.
72 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.
73 """
75 # outside
76 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)
77 # outside
78 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)
79 # inside
80 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)
81 # inside
82 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)
83 # outside
84 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
86 refresh_materialized_views_rapid(empty_pb2.Empty())
87 refresh_materialized_views(empty_pb2.Empty())
89 with search_session(token5) as api:
90 res = api.UserSearch(
91 search_pb2.UserSearchReq(
92 search_in_area=search_pb2.Area(
93 lat=0,
94 lng=0,
95 radius=100000,
96 )
97 )
98 )
99 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
101 res = api.UserSearchV2(
102 search_pb2.UserSearchReq(
103 search_in_area=search_pb2.Area(
104 lat=0,
105 lng=0,
106 radius=100000,
107 )
108 )
109 )
110 assert [result.user_id for result in res.results] == [user3.id, user4.id]
113def test_user_search_in_rectangle(db):
114 """
115 Makes sure search_in_rectangle works as expected.
116 """
118 # outside
119 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)
120 # outside
121 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)
122 # inside
123 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)
124 # inside
125 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)
126 # outside (not fully inside)
127 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)
128 # outside
129 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)
130 # outside
131 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)
133 refresh_materialized_views_rapid(empty_pb2.Empty())
134 refresh_materialized_views(empty_pb2.Empty())
136 with search_session(token5) as api:
137 res = api.UserSearch(
138 search_pb2.UserSearchReq(
139 search_in_rectangle=search_pb2.RectArea(
140 lat_min=0,
141 lat_max=2,
142 lng_min=0,
143 lng_max=1,
144 )
145 )
146 )
147 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]
149 res = api.UserSearchV2(
150 search_pb2.UserSearchReq(
151 search_in_rectangle=search_pb2.RectArea(
152 lat_min=0,
153 lat_max=2,
154 lng_min=0,
155 lng_max=1,
156 )
157 )
158 )
159 assert [result.user_id for result in res.results] == [user3.id, user4.id]
162def test_user_filter_complete_profile(db):
163 """
164 Make sure the completed profile flag returns only completed user profile
165 """
166 user_complete_profile, token6 = generate_user(complete_profile=True)
168 user_incomplete_profile, token7 = generate_user(complete_profile=False)
170 refresh_materialized_views_rapid(empty_pb2.Empty())
171 refresh_materialized_views(empty_pb2.Empty())
173 with search_session(token7) as api:
174 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
175 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]
177 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))
178 assert user_incomplete_profile.id in [result.user_id for result in res.results]
180 with search_session(token6) as api:
181 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
182 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]
184 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))
185 assert [result.user_id for result in res.results] == [user_complete_profile.id]
188def test_user_filter_meetup_status(db):
189 """
190 Make sure the completed profile flag returns only completed user profile
191 """
192 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
194 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)
196 refresh_materialized_views_rapid(empty_pb2.Empty())
197 refresh_materialized_views(empty_pb2.Empty())
199 with search_session(token8) as api:
200 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
201 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]
203 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))
204 assert user_wants_to_meetup.id in [result.user_id for result in res.results]
206 with search_session(token9) as api:
207 res = api.UserSearch(
208 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
209 )
210 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]
212 res = api.UserSearchV2(
213 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])
214 )
215 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]
218def test_user_filter_language(db):
219 """
220 Test filtering users by language ability.
221 """
222 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)
223 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)
224 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)
226 with session_scope() as session:
227 session.add(
228 LanguageAbility(
229 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner
230 ),
231 )
232 session.add(
233 LanguageAbility(
234 user_id=user_with_japanese_conversational.id,
235 language_code="jpn",
236 fluency=LanguageFluency.fluent,
237 )
238 )
239 session.add(
240 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)
241 )
243 refresh_materialized_views_rapid(empty_pb2.Empty())
244 refresh_materialized_views(empty_pb2.Empty())
246 with search_session(token11) as api:
247 res = api.UserSearch(
248 search_pb2.UserSearchReq(
249 language_ability_filter=[
250 api_pb2.LanguageAbility(
251 code="deu",
252 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
253 )
254 ]
255 )
256 )
257 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]
259 res = api.UserSearchV2(
260 search_pb2.UserSearchReq(
261 language_ability_filter=[
262 api_pb2.LanguageAbility(
263 code="deu",
264 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,
265 )
266 ]
267 )
268 )
269 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]
271 res = api.UserSearch(
272 search_pb2.UserSearchReq(
273 language_ability_filter=[
274 api_pb2.LanguageAbility(
275 code="jpn",
276 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
277 )
278 ]
279 )
280 )
281 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]
283 res = api.UserSearchV2(
284 search_pb2.UserSearchReq(
285 language_ability_filter=[
286 api_pb2.LanguageAbility(
287 code="jpn",
288 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,
289 )
290 ]
291 )
292 )
293 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]
296def test_user_filter_strong_verification(db):
297 user1, token1 = generate_user()
298 user2, _ = generate_user(strong_verification=True)
299 user3, _ = generate_user()
300 user4, _ = generate_user(strong_verification=True)
301 user5, _ = generate_user(strong_verification=True)
303 refresh_materialized_views_rapid(empty_pb2.Empty())
304 refresh_materialized_views(empty_pb2.Empty())
306 with search_session(token1) as api:
307 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))
308 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
310 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))
311 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]
313 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))
314 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]
316 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))
317 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]
320def test_regression_search_only_with_references(db):
321 user1, token1 = generate_user()
322 user2, _ = generate_user()
323 user3, _ = generate_user()
324 user4, _ = generate_user(delete_user=True)
326 refresh_materialized_views_rapid(empty_pb2.Empty())
327 refresh_materialized_views(empty_pb2.Empty())
329 with session_scope() as session:
330 # user 2 has references
331 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))
332 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))
333 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))
335 # user 3 only has reference from a deleted user
336 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))
338 with search_session(token1) as api:
339 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=False))
340 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id]
342 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=False))
343 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id]
345 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))
346 assert [result.user.user_id for result in res.results] == [user2.id]
348 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
349 assert [result.user_id for result in res.results] == [user2.id]
352def test_user_search_exactly_user_ids(db):
353 """
354 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.
355 """
356 # Create users with different properties
357 user1, token1 = generate_user()
358 user2, _ = generate_user(strong_verification=True)
359 user3, _ = generate_user(complete_profile=True)
360 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)
361 user5, _ = generate_user(delete_user=True) # Deleted user
363 refresh_materialized_views_rapid(empty_pb2.Empty())
364 refresh_materialized_views(empty_pb2.Empty())
366 with search_session(token1) as api:
367 # Test that exactly_user_ids returns only the specified users
368 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
369 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
371 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))
372 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
374 # Test that exactly_user_ids ignores other filters
375 res = api.UserSearch(
376 search_pb2.UserSearchReq(
377 exactly_user_ids=[user2.id, user3.id, user4.id],
378 only_with_strong_verification=True, # This would normally filter out user3 and user4
379 )
380 )
381 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
383 res = api.UserSearchV2(
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_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])
391 # Test with non-existent user IDs (should be ignored)
392 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
393 assert [result.user.user_id for result in res.results] == [user1.id]
395 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))
396 assert [result.user_id for result in res.results] == [user1.id]
398 # Test with deleted user ID (should be ignored due to visibility filter)
399 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
400 assert [result.user.user_id for result in res.results] == [user1.id]
402 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))
403 assert [result.user_id for result in res.results] == [user1.id]
406@pytest.fixture
407def sample_event_data() -> dict[str, Any]:
408 """Dummy data for creating events."""
409 start_time = now() + timedelta(hours=2)
410 end_time = start_time + timedelta(hours=3)
411 return {
412 "title": "Dummy Title",
413 "content": "Dummy content.",
414 "photo_key": None,
415 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),
416 "start_time": Timestamp_from_datetime(start_time),
417 "end_time": Timestamp_from_datetime(end_time),
418 "timezone": "UTC",
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_time=Timestamp_from_datetime(now() + timedelta(hours=1)),
478 end_time=Timestamp_from_datetime(now() + timedelta(hours=2)),
479 )
480 event2 = create_event(
481 api,
482 start_time=Timestamp_from_datetime(now() + timedelta(hours=4)),
483 end_time=Timestamp_from_datetime(now() + timedelta(hours=5)),
484 )
485 event3 = create_event(
486 api,
487 start_time=Timestamp_from_datetime(now() + timedelta(hours=7)),
488 end_time=Timestamp_from_datetime(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 is correct
575 """
576 user, token = generate_user()
578 anchor_time = now()
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_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
585 end_time=Timestamp_from_datetime(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 == str(millis_from_dt(anchor_time + timedelta(hours=5, minutes=30)))
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 res = api.EventSearch(
600 search_pb2.EventSearchReq(
601 past=True, page_size=2, page_token=str(millis_from_dt(anchor_time + timedelta(hours=4, minutes=30)))
602 )
603 )
604 assert len(res.events) == 2
605 assert [event.title for event in res.events] == ["Event 4", "Event 3"]
606 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=2, minutes=30)))
608 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))
609 assert len(res.events) == 2
610 assert [event.title for event in res.events] == ["Event 2", "Event 1"]
611 assert res.next_page_token == ""
614def test_event_search_pagination_with_page_number(sample_community, create_event):
615 """Test that EventSearch paginates correctly with page number.
617 Check that
618 - <page_size> events are returned, if available
619 - sort order is applied (default: past=False)
620 - <page_number> is respected
621 - <total_items> is correct
622 """
623 user, token = generate_user()
625 anchor_time = now()
626 with events_session(token) as api:
627 for i in range(5):
628 create_event(
629 api,
630 title=f"Event {i + 1}",
631 start_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1)),
632 end_time=Timestamp_from_datetime(anchor_time + timedelta(hours=i + 1, minutes=30)),
633 )
635 with search_session(token) as api:
636 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))
637 assert len(res.events) == 2
638 assert [event.title for event in res.events] == ["Event 1", "Event 2"]
639 assert res.total_items == 5
641 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))
642 assert len(res.events) == 2
643 assert [event.title for event in res.events] == ["Event 3", "Event 4"]
644 assert res.total_items == 5
646 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))
647 assert len(res.events) == 1
648 assert [event.title for event in res.events] == ["Event 5"]
649 assert res.total_items == 5
651 # Verify no more pages
652 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))
653 assert not res.events
654 assert res.total_items == 5
657def test_event_search_filter_subscription_attendance_organizing_my_communities(
658 sample_community, create_event, moderator: Moderator
659):
660 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default
661 returns all events.
662 """
663 _, token = generate_user()
664 other_user, other_token = generate_user()
666 with communities_session(token) as api:
667 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
669 with session_scope() as session:
670 create_community(session, 55, 60, "Other community", [other_user], [], None)
672 with events_session(other_token) as api:
673 e_subscribed = create_event(api, title="Subscribed event")
674 e_attending = create_event(api, title="Attending event")
675 create_event(api, title="Community event")
676 create_event(
677 api,
678 title="Other community event",
679 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
680 )
682 # Approve all events so they're visible to other users
683 with session_scope() as session:
684 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
685 for oid in occurrence_ids:
686 moderator.approve_event_occurrence(oid)
688 with events_session(token) as api:
689 create_event(api, title="Organized event")
690 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))
691 api.SetEventAttendance(
692 events_pb2.SetEventAttendanceReq(
693 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
694 )
695 )
697 with search_session(token) as api:
698 res = api.EventSearch(search_pb2.EventSearchReq())
699 assert {event.title for event in res.events} == {
700 "Subscribed event",
701 "Attending event",
702 "Community event",
703 "Other community event",
704 "Organized event",
705 }
707 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))
708 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}
710 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))
711 assert {event.title for event in res.events} == {"Attending event", "Organized event"}
713 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))
714 assert {event.title for event in res.events} == {"Organized event"}
716 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
717 assert {event.title for event in res.events} == {
718 "Subscribed event",
719 "Attending event",
720 "Community event",
721 "Organized event",
722 }
724 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))
725 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}
728def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):
729 """Test that exclude_attending removes events the user is attending or organizing."""
730 user, token = generate_user()
731 other_user, other_token = generate_user()
733 with communities_session(token) as api:
734 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))
736 with session_scope() as session:
737 create_community(session, 55, 60, "Other community", [other_user], [], None)
739 with events_session(other_token) as api:
740 e_attending = create_event(api, title="Attending event")
741 e_community_only = create_event(api, title="Community only event")
742 create_event(
743 api,
744 title="Other community event",
745 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),
746 )
748 with session_scope() as session:
749 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()
750 for oid in occurrence_ids:
751 moderator.approve_event_occurrence(oid)
753 with events_session(token) as api:
754 e_organized = create_event(api, title="Organized event")
755 api.SetEventAttendance(
756 events_pb2.SetEventAttendanceReq(
757 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING
758 )
759 )
761 with search_session(token) as api:
762 # baseline: my_communities returns all community events including attended/organized
763 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))
764 assert {event.title for event in res.events} == {
765 "Attending event",
766 "Community only event",
767 "Organized event",
768 }
770 # my_communities + exclude_attending: drops attended and organized events
771 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))
772 assert {event.title for event in res.events} == {"Community only event"}
774 # exclude_attending alone (no other filter = all events): drops attended and organized
775 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))
776 assert {event.title for event in res.events} == {"Community only event", "Other community event"}
778 # attending + exclude_attending is invalid
779 with pytest.raises(grpc.RpcError) as e:
780 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))
781 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
784def test_regression_search_multiple_pages(db):
785 """
786 There was a bug when there are multiple pages of results
787 """
788 user, token = generate_user()
789 user_ids = [user.id]
790 for _ in range(10):
791 other_user, _ = generate_user()
792 user_ids.append(other_user.id)
794 refresh_materialized_views_rapid(empty_pb2.Empty())
795 refresh_materialized_views(empty_pb2.Empty())
797 with search_session(token) as api:
798 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))
799 assert [result.user_id for result in res.results] == user_ids[:5]
800 assert res.next_page_token
803def test_regression_search_no_results(db):
804 """
805 There was a bug when there were no results
806 """
807 # put us far away
808 user, token = generate_user()
810 refresh_materialized_views_rapid(empty_pb2.Empty())
811 refresh_materialized_views(empty_pb2.Empty())
813 with search_session(token) as api:
814 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))
815 assert len(res.results) == 0
818def test_user_filter_same_gender_only(db):
819 """Test that same_gender_only filter works correctly"""
820 # Create users with different genders and strong verification status
821 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")
822 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")
823 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")
824 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")
825 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")
827 refresh_materialized_views_rapid(empty_pb2.Empty())
828 refresh_materialized_views(empty_pb2.Empty())
830 # Test 1: Woman with strong verification should see only women when same_gender_only=True
831 with search_session(token_woman_with_sv) as api:
832 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
833 result_ids = [result.user.user_id for result in res.results]
834 assert woman_with_sv.id in result_ids
835 assert woman_without_sv.id in result_ids
836 assert other_woman_with_sv.id in result_ids
837 assert man_with_sv.id not in result_ids
838 assert man_without_sv.id not in result_ids
840 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
841 result_ids = [result.user_id for result in res.results]
842 assert woman_with_sv.id in result_ids
843 assert woman_without_sv.id in result_ids
844 assert other_woman_with_sv.id in result_ids
845 assert man_with_sv.id not in result_ids
846 assert man_without_sv.id not in result_ids
848 # Test 2: Man with strong verification should see only men when same_gender_only=True
849 with search_session(token_man_with_sv) as api:
850 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
851 result_ids = [result.user.user_id for result in res.results]
852 assert man_with_sv.id in result_ids
853 assert man_without_sv.id in result_ids
854 assert woman_with_sv.id not in result_ids
855 assert woman_without_sv.id not in result_ids
856 assert other_woman_with_sv.id not in result_ids
858 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
859 result_ids = [result.user_id for result in res.results]
860 assert man_with_sv.id in result_ids
861 assert man_without_sv.id in result_ids
862 assert woman_with_sv.id not in result_ids
863 assert woman_without_sv.id not in result_ids
864 assert other_woman_with_sv.id not in result_ids
866 # Test 3: Woman without strong verification should get an error
867 with search_session(token_woman_without_sv) as api:
868 with pytest.raises(Exception) as e:
869 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))
870 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
872 with pytest.raises(Exception) as e:
873 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))
874 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)
876 # Test 4: When same_gender_only=False, should see all users
877 with search_session(token_woman_with_sv) as api:
878 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))
879 result_ids = [result.user.user_id for result in res.results]
880 assert woman_with_sv.id in result_ids
881 assert woman_without_sv.id in result_ids
882 assert other_woman_with_sv.id in result_ids
883 assert man_with_sv.id in result_ids
884 assert man_without_sv.id in result_ids
886 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))
887 result_ids = [result.user_id for result in res.results]
888 assert woman_with_sv.id in result_ids
889 assert woman_without_sv.id in result_ids
890 assert other_woman_with_sv.id in result_ids
891 assert man_with_sv.id in result_ids
892 assert man_without_sv.id in result_ids
895def test_user_filter_same_gender_only_with_other_filters(db):
896 """Test that same_gender_only filter works correctly combined with other filters"""
897 # Create users with different properties
898 woman_host, token_woman = generate_user(
899 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host
900 )
901 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)
902 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)
904 refresh_materialized_views_rapid(empty_pb2.Empty())
905 refresh_materialized_views(empty_pb2.Empty())
907 # Test: Combine same_gender_only with hosting_status filter
908 with search_session(token_woman) as api:
909 res = api.UserSearch(
910 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
911 )
912 result_ids = [result.user.user_id for result in res.results]
913 # Should only see woman who can host
914 assert woman_host.id in result_ids
915 assert woman_cant_host.id not in result_ids
916 assert man_host.id not in result_ids
918 res = api.UserSearchV2(
919 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])
920 )
921 result_ids = [result.user_id for result in res.results]
922 assert woman_host.id in result_ids
923 assert woman_cant_host.id not in result_ids
924 assert man_host.id not in result_ids