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 (

14 Timestamp_from_datetime,

15 create_coordinate,

16 datetime_to_iso8601_local,

17 millis_from_dt,

18 now,

19)

20from tests.fixtures.db import generate_user

21from tests.fixtures.misc import Moderator

22from tests.fixtures.sessions import communities_session, events_session, search_session

23from tests.test_communities import create_community, testing_communities # noqa

24from tests.test_references import create_friend_reference

27@pytest.fixture(autouse=True)

28def _(testconfig):

29 pass

32def test_Search(testing_communities):

33 user, token = generate_user()

34 with search_session(token) as api:

35 res = api.Search(

36 search_pb2.SearchReq(

37 query="Country 1, Region 1",

38 include_users=True,

39 include_communities=True,

40 include_groups=True,

41 include_places=True,

42 include_guides=True,

43 )

44 )

45 res = api.Search(

46 search_pb2.SearchReq(

47 query="Country 1, Region 1, Attraction",

48 title_only=True,

49 include_users=True,

50 include_communities=True,

51 include_groups=True,

52 include_places=True,

53 include_guides=True,

54 )

55 )

58def test_UserSearch(testing_communities):

59 """Test that UserSearch returns all users if no filter is set."""

60 user, token = generate_user()

62 refresh_materialized_views_rapid(empty_pb2.Empty())

63 refresh_materialized_views(empty_pb2.Empty())

65 with search_session(token) as api:

66 res = api.UserSearch(search_pb2.UserSearchReq())

67 assert len(res.results) > 0

68 assert res.total_items == len(res.results)

69 res = api.UserSearchV2(search_pb2.UserSearchReq())

70 assert len(res.results) > 0

71 assert res.total_items == len(res.results)

74def test_regression_search_in_area(db):

75 """

76 Makes sure search_in_area works.

78 At the equator/prime meridian intersection (0,0), one degree is roughly 111 km.

79 """

81 # outside

82 user1, token1 = generate_user(geom=create_coordinate(1, 0), geom_radius=100)

83 # outside

84 user2, token2 = generate_user(geom=create_coordinate(0, 1), geom_radius=100)

85 # inside

86 user3, token3 = generate_user(geom=create_coordinate(0.1, 0), geom_radius=100)

87 # inside

88 user4, token4 = generate_user(geom=create_coordinate(0, 0.1), geom_radius=100)

89 # outside

90 user5, token5 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)

92 refresh_materialized_views_rapid(empty_pb2.Empty())

93 refresh_materialized_views(empty_pb2.Empty())

95 with search_session(token5) as api:

96 res = api.UserSearch(

97 search_pb2.UserSearchReq(

98 search_in_area=search_pb2.Area(

99 lat=0,

100 lng=0,

101 radius=100000,

102 )

103 )

104 )

105 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]

106

107 res = api.UserSearchV2(

108 search_pb2.UserSearchReq(

109 search_in_area=search_pb2.Area(

110 lat=0,

111 lng=0,

112 radius=100000,

113 )

114 )

115 )

116 assert [result.user_id for result in res.results] == [user3.id, user4.id]

117

118

119def test_user_search_in_rectangle(db):

120 """

121 Makes sure search_in_rectangle works as expected.

122 """

123

124 # outside

125 user1, token1 = generate_user(geom=create_coordinate(-1, 0), geom_radius=100)

126 # outside

127 user2, token2 = generate_user(geom=create_coordinate(0, -1), geom_radius=100)

128 # inside

129 user3, token3 = generate_user(geom=create_coordinate(0.1, 0.1), geom_radius=100)

130 # inside

131 user4, token4 = generate_user(geom=create_coordinate(1.2, 0.1), geom_radius=100)

132 # outside (not fully inside)

133 user5, token5 = generate_user(geom=create_coordinate(0, 0), geom_radius=100)

134 # outside

135 user6, token6 = generate_user(geom=create_coordinate(0.1, 1.2), geom_radius=100)

136 # outside

137 user7, token7 = generate_user(geom=create_coordinate(10, 10), geom_radius=100)

138

139 refresh_materialized_views_rapid(empty_pb2.Empty())

140 refresh_materialized_views(empty_pb2.Empty())

141

142 with search_session(token5) as api:

143 res = api.UserSearch(

144 search_pb2.UserSearchReq(

145 search_in_rectangle=search_pb2.RectArea(

146 lat_min=0,

147 lat_max=2,

148 lng_min=0,

149 lng_max=1,

150 )

151 )

152 )

153 assert [result.user.user_id for result in res.results] == [user3.id, user4.id]

154

155 res = api.UserSearchV2(

156 search_pb2.UserSearchReq(

157 search_in_rectangle=search_pb2.RectArea(

158 lat_min=0,

159 lat_max=2,

160 lng_min=0,

161 lng_max=1,

162 )

163 )

164 )

165 assert [result.user_id for result in res.results] == [user3.id, user4.id]

166

167

168def test_user_filter_complete_profile(db):

169 """

170 Make sure the completed profile flag returns only completed user profile

171 """

172 user_complete_profile, token6 = generate_user(complete_profile=True)

173

174 user_incomplete_profile, token7 = generate_user(complete_profile=False)

175

176 refresh_materialized_views_rapid(empty_pb2.Empty())

177 refresh_materialized_views(empty_pb2.Empty())

178

179 with search_session(token7) as api:

180 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))

181 assert user_incomplete_profile.id in [result.user.user_id for result in res.results]

182

183 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=False)))

184 assert user_incomplete_profile.id in [result.user_id for result in res.results]

185

186 with search_session(token6) as api:

187 res = api.UserSearch(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))

188 assert [result.user.user_id for result in res.results] == [user_complete_profile.id]

189

190 res = api.UserSearchV2(search_pb2.UserSearchReq(profile_completed=wrappers_pb2.BoolValue(value=True)))

191 assert [result.user_id for result in res.results] == [user_complete_profile.id]

192

193

194def test_user_filter_meetup_status(db):

195 """

196 Make sure the completed profile flag returns only completed user profile

197 """

198 user_wants_to_meetup, token8 = generate_user(meetup_status=MeetupStatus.wants_to_meetup)

199

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

201

202 refresh_materialized_views_rapid(empty_pb2.Empty())

203 refresh_materialized_views(empty_pb2.Empty())

204

205 with search_session(token8) as api:

206 res = api.UserSearch(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))

207 assert user_wants_to_meetup.id in [result.user.user_id for result in res.results]

208

209 res = api.UserSearchV2(search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_WANTS_TO_MEETUP]))

210 assert user_wants_to_meetup.id in [result.user_id for result in res.results]

211

212 with search_session(token9) as api:

213 res = api.UserSearch(

214 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])

215 )

216 assert [result.user.user_id for result in res.results] == [user_does_not_want_to_meet.id]

217

218 res = api.UserSearchV2(

219 search_pb2.UserSearchReq(meetup_status_filter=[api_pb2.MEETUP_STATUS_DOES_NOT_WANT_TO_MEETUP])

220 )

221 assert [result.user_id for result in res.results] == [user_does_not_want_to_meet.id]

222

223

224def test_user_filter_language(db):

225 """

226 Test filtering users by language ability.

227 """

228 user_with_german_beginner, token11 = generate_user(hosting_status=HostingStatus.can_host)

229 user_with_japanese_conversational, token12 = generate_user(hosting_status=HostingStatus.can_host)

230 user_with_german_fluent, token13 = generate_user(hosting_status=HostingStatus.can_host)

231

232 with session_scope() as session:

233 session.add(

234 LanguageAbility(

235 user_id=user_with_german_beginner.id, language_code="deu", fluency=LanguageFluency.beginner

236 ),

237 )

238 session.add(

239 LanguageAbility(

240 user_id=user_with_japanese_conversational.id,

241 language_code="jpn",

242 fluency=LanguageFluency.fluent,

243 )

244 )

245 session.add(

246 LanguageAbility(user_id=user_with_german_fluent.id, language_code="deu", fluency=LanguageFluency.fluent)

247 )

248

249 refresh_materialized_views_rapid(empty_pb2.Empty())

250 refresh_materialized_views(empty_pb2.Empty())

251

252 with search_session(token11) as api:

253 res = api.UserSearch(

254 search_pb2.UserSearchReq(

255 language_ability_filter=[

256 api_pb2.LanguageAbility(

257 code="deu",

258 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

259 )

260 ]

261 )

262 )

263 assert [result.user.user_id for result in res.results] == [user_with_german_fluent.id]

264

265 res = api.UserSearchV2(

266 search_pb2.UserSearchReq(

267 language_ability_filter=[

268 api_pb2.LanguageAbility(

269 code="deu",

270 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_FLUENT,

271 )

272 ]

273 )

274 )

275 assert [result.user_id for result in res.results] == [user_with_german_fluent.id]

276

277 res = api.UserSearch(

278 search_pb2.UserSearchReq(

279 language_ability_filter=[

280 api_pb2.LanguageAbility(

281 code="jpn",

282 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

283 )

284 ]

285 )

286 )

287 assert [result.user.user_id for result in res.results] == [user_with_japanese_conversational.id]

288

289 res = api.UserSearchV2(

290 search_pb2.UserSearchReq(

291 language_ability_filter=[

292 api_pb2.LanguageAbility(

293 code="jpn",

294 fluency=api_pb2.LanguageAbility.Fluency.FLUENCY_CONVERSATIONAL,

295 )

296 ]

297 )

298 )

299 assert [result.user_id for result in res.results] == [user_with_japanese_conversational.id]

300

301

302def test_user_filter_strong_verification(db):

303 user1, token1 = generate_user()

304 user2, _ = generate_user(strong_verification=True)

305 user3, _ = generate_user()

306 user4, _ = generate_user(strong_verification=True)

307 user5, _ = generate_user(strong_verification=True)

308

309 refresh_materialized_views_rapid(empty_pb2.Empty())

310 refresh_materialized_views(empty_pb2.Empty())

311

312 with search_session(token1) as api:

313 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=False))

314 assert [result.user.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]

315

316 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=False))

317 assert [result.user_id for result in res.results] == [user1.id, user2.id, user3.id, user4.id, user5.id]

318

319 res = api.UserSearch(search_pb2.UserSearchReq(only_with_strong_verification=True))

320 assert [result.user.user_id for result in res.results] == [user2.id, user4.id, user5.id]

321

322 res = api.UserSearchV2(search_pb2.UserSearchReq(only_with_strong_verification=True))

323 assert [result.user_id for result in res.results] == [user2.id, user4.id, user5.id]

324

325

326def test_regression_search_only_with_references(db):

327 user1, token1 = generate_user()

328 user2, _ = generate_user()

329 user3, _ = generate_user()

330 user4, _ = generate_user(delete_user=True)

331

332 refresh_materialized_views_rapid(empty_pb2.Empty())

333 refresh_materialized_views(empty_pb2.Empty())

334

335 with session_scope() as session:

336 # user 2 has references

337 create_friend_reference(session, user1.id, user2.id, timedelta(days=1))

338 create_friend_reference(session, user3.id, user2.id, timedelta(days=1))

339 create_friend_reference(session, user4.id, user2.id, timedelta(days=1))

340

341 # user 3 only has reference from a deleted user

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

343

344 with search_session(token1) as api:

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

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

347

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

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

350

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

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

353

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

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

356

357

358def test_user_search_exactly_user_ids(db):

359 """

360 Test that UserSearch with exactly_user_ids returns only those users and ignores other filters.

361 """

362 # Create users with different properties

363 user1, token1 = generate_user()

364 user2, _ = generate_user(strong_verification=True)

365 user3, _ = generate_user(complete_profile=True)

366 user4, _ = generate_user(meetup_status=MeetupStatus.wants_to_meetup)

367 user5, _ = generate_user(delete_user=True) # Deleted user

368

369 refresh_materialized_views_rapid(empty_pb2.Empty())

370 refresh_materialized_views(empty_pb2.Empty())

371

372 with search_session(token1) as api:

373 # Test that exactly_user_ids returns only the specified users

374 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))

375 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

376

377 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user2.id, user3.id, user4.id]))

378 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

379

380 # Test that exactly_user_ids ignores other filters

381 res = api.UserSearch(

382 search_pb2.UserSearchReq(

383 exactly_user_ids=[user2.id, user3.id, user4.id],

384 only_with_strong_verification=True, # This would normally filter out user3 and user4

385 )

386 )

387 assert sorted([result.user.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

388

389 res = api.UserSearchV2(

390 search_pb2.UserSearchReq(

391 exactly_user_ids=[user2.id, user3.id, user4.id],

392 only_with_strong_verification=True, # This would normally filter out user3 and user4

393 )

394 )

395 assert sorted([result.user_id for result in res.results]) == sorted([user2.id, user3.id, user4.id])

396

397 # Test with non-existent user IDs (should be ignored)

398 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))

399 assert [result.user.user_id for result in res.results] == [user1.id]

400

401 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, 99999]))

402 assert [result.user_id for result in res.results] == [user1.id]

403

404 # Test with deleted user ID (should be ignored due to visibility filter)

405 res = api.UserSearch(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))

406 assert [result.user.user_id for result in res.results] == [user1.id]

407

408 res = api.UserSearchV2(search_pb2.UserSearchReq(exactly_user_ids=[user1.id, user5.id]))

409 assert [result.user_id for result in res.results] == [user1.id]

410

411

412@pytest.fixture

413def sample_event_data() -> dict[str, Any]:

414 """Dummy data for creating events."""

415 start_time = now() + timedelta(hours=2)

416 end_time = start_time + timedelta(hours=3)

417 return {

418 "title": "Dummy Title",

419 "content": "Dummy content.",

420 "photo_key": None,

421 "location": events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),

422 "start_datetime_iso8601_local": datetime_to_iso8601_local(start_time),

423 "end_datetime_iso8601_local": datetime_to_iso8601_local(end_time),

424 }

425

426

427@pytest.fixture

428def create_event(sample_event_data):

429 """Factory for creating events."""

430

431 def _create_event(event_api, **kwargs) -> EventOccurrence:

432 """Create an event with default values, unless overridden by kwargs."""

433 return event_api.CreateEvent(events_pb2.CreateEventReq(**{**sample_event_data, **kwargs})) # type: ignore

434

435 return _create_event

436

437

438@pytest.fixture

439def sample_community(db) -> int:

440 """Create large community spanning from (-50, 0) to (50, 2) as events can only be created within communities."""

441 user, _ = generate_user()

442 with session_scope() as session:

443 return create_community(session, -50, 50, "Community", [user], [], None).id

444

445

446def test_EventSearch_no_filters(testing_communities):

447 """Test that EventSearch returns all events if no filter is set."""

448 user, token = generate_user()

449 with search_session(token) as api:

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

451 assert len(res.events) > 0

452

453

454def test_event_search_by_query(sample_community, create_event):

455 """Test that EventSearch finds events by title (and content if query_title_only=False)."""

456 user, token = generate_user()

457

458 with events_session(token) as api:

459 event1 = create_event(api, title="Lorem Ipsum")

460 event2 = create_event(api, content="Lorem Ipsum")

461 create_event(api)

462

463 with search_session(token) as api:

464 res = api.EventSearch(search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum")))

465 assert len(res.events) == 2

466 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}

467

468 res = api.EventSearch(

469 search_pb2.EventSearchReq(query=wrappers_pb2.StringValue(value="Ipsum"), query_title_only=True)

470 )

471 assert len(res.events) == 1

472 assert res.events[0].event_id == event1.event_id

473

474

475def test_event_search_by_time(sample_community, create_event):

476 """Test that EventSearch filters with the given time range."""

477 user, token = generate_user()

478

479 with events_session(token) as api:

480 event1 = create_event(

481 api,

482 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=1)),

483 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=2)),

484 )

485 event2 = create_event(

486 api,

487 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=4)),

488 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=5)),

489 )

490 event3 = create_event(

491 api,

492 start_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=7)),

493 end_datetime_iso8601_local=datetime_to_iso8601_local(now() + timedelta(hours=8)),

494 )

495

496 with search_session(token) as api:

497 res = api.EventSearch(search_pb2.EventSearchReq(before=Timestamp_from_datetime(now() + timedelta(hours=6))))

498 assert len(res.events) == 2

499 assert {result.event_id for result in res.events} == {event1.event_id, event2.event_id}

500

501 res = api.EventSearch(search_pb2.EventSearchReq(after=Timestamp_from_datetime(now() + timedelta(hours=3))))

502 assert len(res.events) == 2

503 assert {result.event_id for result in res.events} == {event2.event_id, event3.event_id}

504

505 res = api.EventSearch(

506 search_pb2.EventSearchReq(

507 before=Timestamp_from_datetime(now() + timedelta(hours=6)),

508 after=Timestamp_from_datetime(now() + timedelta(hours=3)),

509 )

510 )

511 assert len(res.events) == 1

512 assert res.events[0].event_id == event2.event_id

513

514

515def test_event_search_by_circle(sample_community, create_event):

516 """Test that EventSearch only returns events within the given circle."""

517 user, token = generate_user()

518

519 with events_session(token) as api:

520 inside_pts = [(0.1, 0.01), (0.01, 0.1)]

521 for i, (lat, lng) in enumerate(inside_pts):

522 create_event(

523 api,

524 title=f"Inside area {i}",

525 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),

526 )

527

528 outside_pts = [(1, 0.1), (0.1, 1), (10, 1)]

529 for i, (lat, lng) in enumerate(outside_pts):

530 create_event(

531 api,

532 title=f"Outside area {i}",

533 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),

534 )

535

536 with search_session(token) as api:

537 res = api.EventSearch(search_pb2.EventSearchReq(search_in_area=search_pb2.Area(lat=0, lng=0, radius=100000)))

538 assert len(res.events) == len(inside_pts)

539 assert all(event.title.startswith("Inside area") for event in res.events)

540

541

542def test_event_search_by_rectangle(sample_community, create_event):

543 """Test that EventSearch only returns events within the given rectangular area."""

544 user, token = generate_user()

545

546 with events_session(token) as api:

547 inside_pts = [(0.1, 0.2), (1.2, 0.2)]

548 for i, (lat, lng) in enumerate(inside_pts):

549 create_event(

550 api,

551 title=f"Inside area {i}",

552 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Inside area {i}"),

553 )

554

555 outside_pts = [(-1, 0.1), (0.1, 0.01), (-0.01, 0.01), (0.1, 1.2), (10, 1)]

556 for i, (lat, lng) in enumerate(outside_pts):

557 create_event(

558 api,

559 title=f"Outside area {i}",

560 location=events_pb2.EventLocation(lat=lat, lng=lng, address=f"Outside area {i}"),

561 )

562

563 with search_session(token) as api:

564 res = api.EventSearch(

565 search_pb2.EventSearchReq(

566 search_in_rectangle=search_pb2.RectArea(lat_min=0, lat_max=2, lng_min=0.1, lng_max=1)

567 )

568 )

569 assert len(res.events) == len(inside_pts)

570 assert all(event.title.startswith("Inside area") for event in res.events)

571

572

573def test_event_search_pagination(sample_community, create_event):

574 """Test that EventSearch paginates correctly.

575

576 Check that

577 - <page_size> events are returned, if available

578 - sort order is applied (default: past=False)

579 - the next page token is correct

580 """

581 user, token = generate_user()

582

583 anchor_time = now().replace(second=0, microsecond=0) # Events are created at minute granularity

584 with events_session(token) as api:

585 for i in range(5):

586 create_event(

587 api,

588 title=f"Event {i + 1}",

589 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),

590 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),

591 )

592

593 with search_session(token) as api:

594 res = api.EventSearch(search_pb2.EventSearchReq(past=False, page_size=4))

595 assert len(res.events) == 4

596 assert [event.title for event in res.events] == ["Event 1", "Event 2", "Event 3", "Event 4"]

597 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=5, minutes=30)))

598

599 res = api.EventSearch(search_pb2.EventSearchReq(page_size=4, page_token=res.next_page_token))

600 assert len(res.events) == 1

601 assert res.events[0].title == "Event 5"

602 assert res.next_page_token == ""

603

604 res = api.EventSearch(

605 search_pb2.EventSearchReq(

606 past=True, page_size=2, page_token=str(millis_from_dt(anchor_time + timedelta(hours=4, minutes=30)))

607 )

608 )

609 assert len(res.events) == 2

610 assert [event.title for event in res.events] == ["Event 4", "Event 3"]

611 assert res.next_page_token == str(millis_from_dt(anchor_time + timedelta(hours=2, minutes=30)))

612

613 res = api.EventSearch(search_pb2.EventSearchReq(past=True, page_size=2, page_token=res.next_page_token))

614 assert len(res.events) == 2

615 assert [event.title for event in res.events] == ["Event 2", "Event 1"]

616 assert res.next_page_token == ""

617

618

619def test_event_search_pagination_with_page_number(sample_community, create_event):

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

621

622 Check that

623 - <page_size> events are returned, if available

624 - sort order is applied (default: past=False)

625 - <page_number> is respected

626 - <total_items> is correct

627 """

628 user, token = generate_user()

629

630 anchor_time = now()

631 with events_session(token) as api:

632 for i in range(5):

633 create_event(

634 api,

635 title=f"Event {i + 1}",

636 start_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1)),

637 end_datetime_iso8601_local=datetime_to_iso8601_local(anchor_time + timedelta(hours=i + 1, minutes=30)),

638 )

639

640 with search_session(token) as api:

641 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=1))

642 assert len(res.events) == 2

643 assert [event.title for event in res.events] == ["Event 1", "Event 2"]

644 assert res.total_items == 5

645

646 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=2))

647 assert len(res.events) == 2

648 assert [event.title for event in res.events] == ["Event 3", "Event 4"]

649 assert res.total_items == 5

650

651 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=3))

652 assert len(res.events) == 1

653 assert [event.title for event in res.events] == ["Event 5"]

654 assert res.total_items == 5

655

656 # Verify no more pages

657 res = api.EventSearch(search_pb2.EventSearchReq(page_size=2, page_number=4))

658 assert not res.events

659 assert res.total_items == 5

660

661

662def test_event_search_filter_subscription_attendance_organizing_my_communities(

663 sample_community, create_event, moderator: Moderator

664):

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

666 returns all events.

667 """

668 _, token = generate_user()

669 other_user, other_token = generate_user()

670

671 with communities_session(token) as api:

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

673

674 with session_scope() as session:

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

676

677 with events_session(other_token) as api:

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

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

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

681 create_event(

682 api,

683 title="Other community event",

684 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),

685 )

686

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

688 with session_scope() as session:

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

690 for oid in occurrence_ids:

691 moderator.approve_event_occurrence(oid)

692

693 with events_session(token) as api:

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

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

696 api.SetEventAttendance(

697 events_pb2.SetEventAttendanceReq(

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

699 )

700 )

701

702 with search_session(token) as api:

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

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

705 "Subscribed event",

706 "Attending event",

707 "Community event",

708 "Other community event",

709 "Organized event",

710 }

711

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

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

714

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

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

717

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

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

720

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

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

723 "Subscribed event",

724 "Attending event",

725 "Community event",

726 "Organized event",

727 }

728

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

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

731

732

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

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

735 user, token = generate_user()

736 other_user, other_token = generate_user()

737

738 with communities_session(token) as api:

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

740

741 with session_scope() as session:

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

743

744 with events_session(other_token) as api:

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

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

747 create_event(

748 api,

749 title="Other community event",

750 location=events_pb2.EventLocation(lat=58, lng=1, address="Somewhere"),

751 )

752

753 with session_scope() as session:

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

755 for oid in occurrence_ids:

756 moderator.approve_event_occurrence(oid)

757

758 with events_session(token) as api:

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

760 api.SetEventAttendance(

761 events_pb2.SetEventAttendanceReq(

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

763 )

764 )

765

766 with search_session(token) as api:

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

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

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

770 "Attending event",

771 "Community only event",

772 "Organized event",

773 }

774

775 # my_communities + exclude_attending: drops attended and organized events

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

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

778

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

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

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

782

783 # attending + exclude_attending is invalid

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

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

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

787

788

789def test_regression_search_multiple_pages(db):

790 """

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

792 """

793 user, token = generate_user()

794 user_ids = [user.id]

795 for _ in range(10):

796 other_user, _ = generate_user()

797 user_ids.append(other_user.id)

798

799 refresh_materialized_views_rapid(empty_pb2.Empty())

800 refresh_materialized_views(empty_pb2.Empty())

801

802 with search_session(token) as api:

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

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

805 assert res.next_page_token

806

807

808def test_regression_search_no_results(db):

809 """

810 There was a bug when there were no results

811 """

812 # put us far away

813 user, token = generate_user()

814

815 refresh_materialized_views_rapid(empty_pb2.Empty())

816 refresh_materialized_views(empty_pb2.Empty())

817

818 with search_session(token) as api:

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

820 assert len(res.results) == 0

821

822

823def test_user_filter_same_gender_only(db):

824 """Test that same_gender_only filter works correctly"""

825 # Create users with different genders and strong verification status

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

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

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

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

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

831

832 refresh_materialized_views_rapid(empty_pb2.Empty())

833 refresh_materialized_views(empty_pb2.Empty())

834

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

836 with search_session(token_woman_with_sv) as api:

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

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

839 assert woman_with_sv.id in result_ids

840 assert woman_without_sv.id in result_ids

841 assert other_woman_with_sv.id in result_ids

842 assert man_with_sv.id not in result_ids

843 assert man_without_sv.id not in result_ids

844

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

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

847 assert woman_with_sv.id in result_ids

848 assert woman_without_sv.id in result_ids

849 assert other_woman_with_sv.id in result_ids

850 assert man_with_sv.id not in result_ids

851 assert man_without_sv.id not in result_ids

852

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

854 with search_session(token_man_with_sv) as api:

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

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

857 assert man_with_sv.id in result_ids

858 assert man_without_sv.id in result_ids

859 assert woman_with_sv.id not in result_ids

860 assert woman_without_sv.id not in result_ids

861 assert other_woman_with_sv.id not in result_ids

862

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

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

865 assert man_with_sv.id in result_ids

866 assert man_without_sv.id in result_ids

867 assert woman_with_sv.id not in result_ids

868 assert woman_without_sv.id not in result_ids

869 assert other_woman_with_sv.id not in result_ids

870

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

872 with search_session(token_woman_without_sv) as api:

873 with pytest.raises(Exception) as e:

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

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

876

877 with pytest.raises(Exception) as e:

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

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

880

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

882 with search_session(token_woman_with_sv) as api:

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

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

885 assert woman_with_sv.id in result_ids

886 assert woman_without_sv.id in result_ids

887 assert other_woman_with_sv.id in result_ids

888 assert man_with_sv.id in result_ids

889 assert man_without_sv.id in result_ids

890

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

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

893 assert woman_with_sv.id in result_ids

894 assert woman_without_sv.id in result_ids

895 assert other_woman_with_sv.id in result_ids

896 assert man_with_sv.id in result_ids

897 assert man_without_sv.id in result_ids

898

899

900def test_user_filter_same_gender_only_with_other_filters(db):

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

902 # Create users with different properties

903 woman_host, token_woman = generate_user(

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

905 )

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

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

908

909 refresh_materialized_views_rapid(empty_pb2.Empty())

910 refresh_materialized_views(empty_pb2.Empty())

911

912 # Test: Combine same_gender_only with hosting_status filter

913 with search_session(token_woman) as api:

914 res = api.UserSearch(

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

916 )

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

918 # Should only see woman who can host

919 assert woman_host.id in result_ids

920 assert woman_cant_host.id not in result_ids

921 assert man_host.id not in result_ids

922

923 res = api.UserSearchV2(

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

925 )

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

927 assert woman_host.id in result_ids

928 assert woman_cant_host.id not in result_ids

929 assert man_host.id not in result_ids

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

452 statements