Coverage for app/backend/src/tests/test

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]

100

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]

111

112

113def test_user_search_in_rectangle(db):

114 """

115 Makes sure search_in_rectangle works as expected.

116 """

117

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)

132

133 refresh_materialized_views_rapid(empty_pb2.Empty())

134 refresh_materialized_views(empty_pb2.Empty())

135

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]

148

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]

160

161

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)

167

168 user_incomplete_profile, token7 = generate_user(complete_profile=False)

169

170 refresh_materialized_views_rapid(empty_pb2.Empty())

171 refresh_materialized_views(empty_pb2.Empty())

172

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]

176

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]

179

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]

183

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]

186

187

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)

193

194 user_does_not_want_to_meet, token9 = generate_user(meetup_status=MeetupStatus.does_not_want_to_meetup)

195

196 refresh_materialized_views_rapid(empty_pb2.Empty())

197 refresh_materialized_views(empty_pb2.Empty())

198

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]

202

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]

205

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]

211

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]

216

217

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)

225

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 )

242

243 refresh_materialized_views_rapid(empty_pb2.Empty())

244 refresh_materialized_views(empty_pb2.Empty())

245

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]

258

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]

270

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]

282

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]

294

295

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)

302

303 refresh_materialized_views_rapid(empty_pb2.Empty())

304 refresh_materialized_views(empty_pb2.Empty())

305

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]

309

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]

312

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]

315

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]

318

319

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)

325

326 refresh_materialized_views_rapid(empty_pb2.Empty())

327 refresh_materialized_views(empty_pb2.Empty())

328

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))

334

335 # user 3 only has reference from a deleted user

336 create_friend_reference(session, user4.id, user3.id, timedelta(days=1))

337

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]

341

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]

344

345 res = api.UserSearch(search_pb2.UserSearchReq(only_with_references=True))

346 assert [result.user.user_id for result in res.results] == [user2.id]

347

348 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))

349 assert [result.user_id for result in res.results] == [user2.id]

350

351

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

362

363 refresh_materialized_views_rapid(empty_pb2.Empty())

364 refresh_materialized_views(empty_pb2.Empty())

365

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])

370

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])

373

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])

382

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])

390

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]

394

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]

397

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]

401

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]

404

405

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 "offline_information": events_pb2.OfflineEventInformation(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 }

420

421

422@pytest.fixture

423def create_event(sample_event_data):

424 """Factory for creating events."""

425

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

429

430 return _create_event

431

432

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

439

440

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

447

448

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()

452

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)

457

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}

462

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

468

469

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()

473

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 )

490

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}

495

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}

499

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

508

509

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()

513

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 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),

521 )

522

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 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),

529 )

530

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)

535

536

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()

540

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 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Inside area {i}"),

548 )

549

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 offline_information=events_pb2.OfflineEventInformation(lat=lat, lng=lng, address=f"Outside area {i}"),

556 )

557

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)

566

567

568def test_event_search_pagination(sample_community, create_event):

569 """Test that EventSearch paginates correctly.

570

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()

577

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 )

587

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)))

593

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 == ""

598

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)))

607

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 == ""

612

613

614def test_event_search_pagination_with_page_number(sample_community, create_event):

615 """Test that EventSearch paginates correctly with page number.

616

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()

624

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 )

634

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

640

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

645

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

650

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

655

656

657def test_event_search_online_status(sample_community, create_event):

658 """Test that EventSearch respects only_online and only_offline filters and by default returns both."""

659 user, token = generate_user()

660

661 with events_session(token) as api:

662 create_event(api, title="Offline event")

663

664 create_event(

665 api,

666 title="Online event",

667 online_information=events_pb2.OnlineEventInformation(link="https://couchers.org/meet/"),

668 parent_community_id=sample_community,

669 offline_information=events_pb2.OfflineEventInformation(),

670 )

671

672 with search_session(token) as api:

673 res = api.EventSearch(search_pb2.EventSearchReq())

674 assert len(res.events) == 2

675 assert {event.title for event in res.events} == {"Offline event", "Online event"}

676

677 res = api.EventSearch(search_pb2.EventSearchReq(only_online=True))

678 assert {event.title for event in res.events} == {"Online event"}

679

680 res = api.EventSearch(search_pb2.EventSearchReq(only_offline=True))

681 assert {event.title for event in res.events} == {"Offline event"}

682

683

684def test_event_search_filter_subscription_attendance_organizing_my_communities(

685 sample_community, create_event, moderator: Moderator

686):

687 """Test that EventSearch respects subscribed, attending, organizing and my_communities filters and by default

688 returns all events.

689 """

690 _, token = generate_user()

691 other_user, other_token = generate_user()

692

693 with communities_session(token) as api:

694 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))

695

696 with session_scope() as session:

697 create_community(session, 55, 60, "Other community", [other_user], [], None)

698

699 with events_session(other_token) as api:

700 e_subscribed = create_event(api, title="Subscribed event")

701 e_attending = create_event(api, title="Attending event")

702 create_event(api, title="Community event")

703 create_event(

704 api,

705 title="Other community event",

706 offline_information=events_pb2.OfflineEventInformation(lat=58, lng=1, address="Somewhere"),

707 )

708

709 # Approve all events so they're visible to other users

710 with session_scope() as session:

711 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()

712 for oid in occurrence_ids:

713 moderator.approve_event_occurrence(oid)

714

715 with events_session(token) as api:

716 create_event(api, title="Organized event")

717 api.SetEventSubscription(events_pb2.SetEventSubscriptionReq(event_id=e_subscribed.event_id, subscribe=True))

718 api.SetEventAttendance(

719 events_pb2.SetEventAttendanceReq(

720 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING

721 )

722 )

723

724 with search_session(token) as api:

725 res = api.EventSearch(search_pb2.EventSearchReq())

726 assert {event.title for event in res.events} == {

727 "Subscribed event",

728 "Attending event",

729 "Community event",

730 "Other community event",

731 "Organized event",

732 }

733

734 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True))

735 assert {event.title for event in res.events} == {"Subscribed event", "Organized event"}

736

737 res = api.EventSearch(search_pb2.EventSearchReq(attending=True))

738 assert {event.title for event in res.events} == {"Attending event", "Organized event"}

739

740 res = api.EventSearch(search_pb2.EventSearchReq(organizing=True))

741 assert {event.title for event in res.events} == {"Organized event"}

742

743 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))

744 assert {event.title for event in res.events} == {

745 "Subscribed event",

746 "Attending event",

747 "Community event",

748 "Organized event",

749 }

750

751 res = api.EventSearch(search_pb2.EventSearchReq(subscribed=True, attending=True))

752 assert {event.title for event in res.events} == {"Subscribed event", "Attending event", "Organized event"}

753

754

755def test_event_search_exclude_attending(sample_community, create_event, moderator: Moderator):

756 """Test that exclude_attending removes events the user is attending or organizing."""

757 user, token = generate_user()

758 other_user, other_token = generate_user()

759

760 with communities_session(token) as api:

761 api.JoinCommunity(communities_pb2.JoinCommunityReq(community_id=sample_community))

762

763 with session_scope() as session:

764 create_community(session, 55, 60, "Other community", [other_user], [], None)

765

766 with events_session(other_token) as api:

767 e_attending = create_event(api, title="Attending event")

768 e_community_only = create_event(api, title="Community only event")

769 create_event(

770 api,

771 title="Other community event",

772 offline_information=events_pb2.OfflineEventInformation(lat=58, lng=1, address="Somewhere"),

773 )

774

775 with session_scope() as session:

776 occurrence_ids = session.execute(select(EventOccurrence.id)).scalars().all()

777 for oid in occurrence_ids:

778 moderator.approve_event_occurrence(oid)

779

780 with events_session(token) as api:

781 e_organized = create_event(api, title="Organized event")

782 api.SetEventAttendance(

783 events_pb2.SetEventAttendanceReq(

784 event_id=e_attending.event_id, attendance_state=events_pb2.ATTENDANCE_STATE_GOING

785 )

786 )

787

788 with search_session(token) as api:

789 # baseline: my_communities returns all community events including attended/organized

790 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True))

791 assert {event.title for event in res.events} == {

792 "Attending event",

793 "Community only event",

794 "Organized event",

795 }

796

797 # my_communities + exclude_attending: drops attended and organized events

798 res = api.EventSearch(search_pb2.EventSearchReq(my_communities=True, exclude_attending=True))

799 assert {event.title for event in res.events} == {"Community only event"}

800

801 # exclude_attending alone (no other filter = all events): drops attended and organized

802 res = api.EventSearch(search_pb2.EventSearchReq(exclude_attending=True))

803 assert {event.title for event in res.events} == {"Community only event", "Other community event"}

804

805 # attending + exclude_attending is invalid

806 with pytest.raises(grpc.RpcError) as e:

807 api.EventSearch(search_pb2.EventSearchReq(attending=True, exclude_attending=True))

808 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT

809

810

811def test_regression_search_multiple_pages(db):

812 """

813 There was a bug when there are multiple pages of results

814 """

815 user, token = generate_user()

816 user_ids = [user.id]

817 for _ in range(10):

818 other_user, _ = generate_user()

819 user_ids.append(other_user.id)

820

821 refresh_materialized_views_rapid(empty_pb2.Empty())

822 refresh_materialized_views(empty_pb2.Empty())

823

824 with search_session(token) as api:

825 res = api.UserSearchV2(search_pb2.UserSearchReq(page_size=5))

826 assert [result.user_id for result in res.results] == user_ids[:5]

827 assert res.next_page_token

828

829

830def test_regression_search_no_results(db):

831 """

832 There was a bug when there were no results

833 """

834 # put us far away

835 user, token = generate_user()

836

837 refresh_materialized_views_rapid(empty_pb2.Empty())

838 refresh_materialized_views(empty_pb2.Empty())

839

840 with search_session(token) as api:

841 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_references=True))

842 assert len(res.results) == 0

843

844

845def test_user_filter_same_gender_only(db):

846 """Test that same_gender_only filter works correctly"""

847 # Create users with different genders and strong verification status

848 woman_with_sv, token_woman_with_sv = generate_user(strong_verification=True, gender="Woman")

849 woman_without_sv, token_woman_without_sv = generate_user(strong_verification=False, gender="Woman")

850 man_with_sv, token_man_with_sv = generate_user(strong_verification=True, gender="Man")

851 man_without_sv, _ = generate_user(strong_verification=False, gender="Man")

852 other_woman_with_sv, _ = generate_user(strong_verification=True, gender="Woman")

853

854 refresh_materialized_views_rapid(empty_pb2.Empty())

855 refresh_materialized_views(empty_pb2.Empty())

856

857 # Test 1: Woman with strong verification should see only women when same_gender_only=True

858 with search_session(token_woman_with_sv) as api:

859 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

860 result_ids = [result.user.user_id for result in res.results]

861 assert woman_with_sv.id in result_ids

862 assert woman_without_sv.id in result_ids

863 assert other_woman_with_sv.id in result_ids

864 assert man_with_sv.id not in result_ids

865 assert man_without_sv.id not in result_ids

866

867 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

868 result_ids = [result.user_id for result in res.results]

869 assert woman_with_sv.id in result_ids

870 assert woman_without_sv.id in result_ids

871 assert other_woman_with_sv.id in result_ids

872 assert man_with_sv.id not in result_ids

873 assert man_without_sv.id not in result_ids

874

875 # Test 2: Man with strong verification should see only men when same_gender_only=True

876 with search_session(token_man_with_sv) as api:

877 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

878 result_ids = [result.user.user_id for result in res.results]

879 assert man_with_sv.id in result_ids

880 assert man_without_sv.id in result_ids

881 assert woman_with_sv.id not in result_ids

882 assert woman_without_sv.id not in result_ids

883 assert other_woman_with_sv.id not in result_ids

884

885 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

886 result_ids = [result.user_id for result in res.results]

887 assert man_with_sv.id in result_ids

888 assert man_without_sv.id in result_ids

889 assert woman_with_sv.id not in result_ids

890 assert woman_without_sv.id not in result_ids

891 assert other_woman_with_sv.id not in result_ids

892

893 # Test 3: Woman without strong verification should get an error

894 with search_session(token_woman_without_sv) as api:

895 with pytest.raises(Exception) as e:

896 api.UserSearch(search_pb2.UserSearchReq(same_gender_only=True))

897 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)

898

899 with pytest.raises(Exception) as e:

900 api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=True))

901 assert "NEED_STRONG_VERIFICATION" in str(e.value) or "FAILED_PRECONDITION" in str(e.value)

902

903 # Test 4: When same_gender_only=False, should see all users

904 with search_session(token_woman_with_sv) as api:

905 res = api.UserSearch(search_pb2.UserSearchReq(same_gender_only=False))

906 result_ids = [result.user.user_id for result in res.results]

907 assert woman_with_sv.id in result_ids

908 assert woman_without_sv.id in result_ids

909 assert other_woman_with_sv.id in result_ids

910 assert man_with_sv.id in result_ids

911 assert man_without_sv.id in result_ids

912

913 res = api.UserSearchV2(search_pb2.UserSearchReq(same_gender_only=False))

914 result_ids = [result.user_id for result in res.results]

915 assert woman_with_sv.id in result_ids

916 assert woman_without_sv.id in result_ids

917 assert other_woman_with_sv.id in result_ids

918 assert man_with_sv.id in result_ids

919 assert man_without_sv.id in result_ids

920

921

922def test_user_filter_same_gender_only_with_other_filters(db):

923 """Test that same_gender_only filter works correctly combined with other filters"""

924 # Create users with different properties

925 woman_host, token_woman = generate_user(

926 strong_verification=True, gender="Woman", hosting_status=HostingStatus.can_host

927 )

928 woman_cant_host, _ = generate_user(strong_verification=True, gender="Woman", hosting_status=HostingStatus.cant_host)

929 man_host, _ = generate_user(strong_verification=True, gender="Man", hosting_status=HostingStatus.can_host)

930

931 refresh_materialized_views_rapid(empty_pb2.Empty())

932 refresh_materialized_views(empty_pb2.Empty())

933

934 # Test: Combine same_gender_only with hosting_status filter

935 with search_session(token_woman) as api:

936 res = api.UserSearch(

937 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])

938 )

939 result_ids = [result.user.user_id for result in res.results]

940 # Should only see woman who can host

941 assert woman_host.id in result_ids

942 assert woman_cant_host.id not in result_ids

943 assert man_host.id not in result_ids

944

945 res = api.UserSearchV2(

946 search_pb2.UserSearchReq(same_gender_only=True, hosting_status_filter=[api_pb2.HOSTING_STATUS_CAN_HOST])

947 )

948 result_ids = [result.user_id for result in res.results]

949 assert woman_host.id in result_ids

950 assert woman_cant_host.id not in result_ids

951 assert man_host.id not in result_ids

Coverage for app/backend/src/tests/test_search.py: 100%

465 statements