Coverage for app/backend/src/tests/test_editor.py: 100%
326 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import grpc
2import pytest
3from google.protobuf import empty_pb2
4from google.protobuf.wrappers_pb2 import BoolValue, DoubleValue, StringValue
5from sqlalchemy import select
7from couchers.db import session_scope
8from couchers.materialized_views import refresh_materialized_views_rapid
9from couchers.models import (
10 Cluster,
11 Node,
12 Volunteer,
13)
14from couchers.proto import editor_pb2
15from tests.fixtures.db import generate_user
16from tests.fixtures.sessions import real_editor_session
18VALID_GEOJSON_MULTIPOLYGON = """
19 {
20 "type": "MultiPolygon",
21 "coordinates":
22 [
23 [
24 [
25 [
26 -73.98114904754641,
27 40.7470284264813
28 ],
29 [
30 -73.98314135177611,
31 40.73416844413217
32 ],
33 [
34 -74.00538969848634,
35 40.734314779027144
36 ],
37 [
38 -74.00479214294432,
39 40.75027851544338
40 ],
41 [
42 -73.98114904754641,
43 40.7470284264813
44 ]
45 ]
46 ]
47 ]
48 }
49"""
51POINT_GEOJSON = """
52{ "type": "Point", "coordinates": [100.0, 0.0] }
53"""
56def test_access_by_normal_user(db):
57 """Normal users should not be able to access editor APIs"""
58 normal_user, normal_token = generate_user()
60 with real_editor_session(normal_token) as api:
61 with pytest.raises(grpc.RpcError) as e:
62 api.CreateCommunity(
63 editor_pb2.CreateCommunityReq(
64 name="test community",
65 description="community for testing",
66 admin_ids=[],
67 geojson=VALID_GEOJSON_MULTIPOLYGON,
68 )
69 )
70 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
71 assert e.value.details() == "Permission denied"
74def test_access_by_editor_user(db):
75 """Editor users should be able to access editor APIs"""
76 editor_user, editor_token = generate_user(is_editor=True)
78 with session_scope() as session:
79 with real_editor_session(editor_token) as api:
80 api.CreateCommunity(
81 editor_pb2.CreateCommunityReq(
82 name="test community",
83 description="community for testing",
84 admin_ids=[],
85 geojson=VALID_GEOJSON_MULTIPOLYGON,
86 )
87 )
88 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
89 assert community.description == "community for testing"
90 assert community.slug == "test-community"
93def test_access_by_superuser(db):
94 """Superusers (who are also editors) should be able to access editor APIs"""
95 editor_user, editor_token = generate_user(is_editor=True)
97 with session_scope() as session:
98 with real_editor_session(editor_token) as api:
99 api.CreateCommunity(
100 editor_pb2.CreateCommunityReq(
101 name="test community",
102 description="community for testing",
103 admin_ids=[],
104 geojson=VALID_GEOJSON_MULTIPOLYGON,
105 )
106 )
107 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
108 assert community.description == "community for testing"
109 assert community.slug == "test-community"
112def test_CreateCommunity_invalid_geojson(db):
113 """CreateCommunity should reject invalid GeoJSON"""
114 editor_user, editor_token = generate_user(is_editor=True)
116 with real_editor_session(editor_token) as api:
117 with pytest.raises(grpc.RpcError) as e:
118 api.CreateCommunity(
119 editor_pb2.CreateCommunityReq(
120 name="test community",
121 description="community for testing",
122 admin_ids=[],
123 geojson=POINT_GEOJSON,
124 )
125 )
126 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
127 assert e.value.details() == "GeoJson was not of type MultiPolygon."
130def test_UpdateCommunity_invalid_geojson(db):
131 """UpdateCommunity should reject invalid GeoJSON"""
132 editor_user, editor_token = generate_user(is_editor=True)
134 with session_scope() as session:
135 with real_editor_session(editor_token) as api:
136 api.CreateCommunity(
137 editor_pb2.CreateCommunityReq(
138 name="test community",
139 description="community for testing",
140 admin_ids=[],
141 geojson=VALID_GEOJSON_MULTIPOLYGON,
142 )
143 )
144 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
146 with pytest.raises(grpc.RpcError) as e:
147 api.UpdateCommunity(
148 editor_pb2.UpdateCommunityReq(
149 community_id=community.parent_node_id,
150 name="test community 2",
151 description="community for testing 2",
152 geojson=POINT_GEOJSON,
153 )
154 )
155 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
156 assert e.value.details() == "GeoJson was not of type MultiPolygon."
159def test_UpdateCommunity_invalid_id(db):
160 """UpdateCommunity should reject invalid community IDs"""
161 editor_user, editor_token = generate_user(is_editor=True)
163 with real_editor_session(editor_token) as api:
164 api.CreateCommunity(
165 editor_pb2.CreateCommunityReq(
166 name="test community",
167 description="community for testing",
168 admin_ids=[],
169 geojson=VALID_GEOJSON_MULTIPOLYGON,
170 )
171 )
173 with pytest.raises(grpc.RpcError) as e:
174 api.UpdateCommunity(
175 editor_pb2.UpdateCommunityReq(
176 community_id=1000,
177 name="test community 1000",
178 description="community for testing 1000",
179 geojson=VALID_GEOJSON_MULTIPOLYGON,
180 )
181 )
182 assert e.value.code() == grpc.StatusCode.NOT_FOUND
183 assert e.value.details() == "Community not found."
186def test_UpdateCommunity(db):
187 """UpdateCommunity should successfully update a community"""
188 editor_user, editor_token = generate_user(is_editor=True)
190 with session_scope() as session:
191 with real_editor_session(editor_token) as api:
192 api.CreateCommunity(
193 editor_pb2.CreateCommunityReq(
194 name="test community",
195 description="community for testing",
196 admin_ids=[],
197 geojson=VALID_GEOJSON_MULTIPOLYGON,
198 )
199 )
200 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
202 api.UpdateCommunity(
203 editor_pb2.UpdateCommunityReq(
204 community_id=community.parent_node_id,
205 name="test community updated",
206 description="community for testing updated",
207 geojson=VALID_GEOJSON_MULTIPOLYGON,
208 )
209 )
210 session.commit()
212 community_updated = session.execute(select(Cluster).where(Cluster.id == community.id)).scalar_one()
213 assert community_updated.name == "test community updated"
214 assert community_updated.description == "community for testing updated"
215 assert community_updated.slug == "test-community-updated"
218def test_CreateCommunity(db):
219 with session_scope() as session:
220 editor_user, editor_token = generate_user(is_editor=True)
221 normal_user, normal_token = generate_user()
222 with real_editor_session(editor_token) as api:
223 api.CreateCommunity(
224 editor_pb2.CreateCommunityReq(
225 name="test community",
226 description="community for testing",
227 admin_ids=[],
228 geojson=VALID_GEOJSON_MULTIPOLYGON,
229 )
230 )
231 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
232 assert community.description == "community for testing"
233 assert community.slug == "test-community"
236def test_UpdateCommunity2(db):
237 editor_user, editor_token = generate_user(is_editor=True)
239 with session_scope() as session:
240 with real_editor_session(editor_token) as api:
241 api.CreateCommunity(
242 editor_pb2.CreateCommunityReq(
243 name="test community",
244 description="community for testing",
245 admin_ids=[],
246 geojson=VALID_GEOJSON_MULTIPOLYGON,
247 )
248 )
249 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
250 assert community.description == "community for testing"
252 api.CreateCommunity(
253 editor_pb2.CreateCommunityReq(
254 name="test community 2",
255 description="community for testing 2",
256 admin_ids=[],
257 geojson=VALID_GEOJSON_MULTIPOLYGON,
258 )
259 )
260 community_2 = session.execute(select(Cluster).where(Cluster.name == "test community 2")).scalar_one()
262 api.UpdateCommunity(
263 editor_pb2.UpdateCommunityReq(
264 community_id=community.parent_node_id,
265 name="test community 2",
266 description="community for testing 2",
267 geojson=VALID_GEOJSON_MULTIPOLYGON,
268 parent_node_id=community_2.parent_node_id,
269 )
270 )
271 session.commit()
273 community_updated = session.execute(select(Cluster).where(Cluster.id == community.id)).scalar_one()
274 assert community_updated.description == "community for testing 2"
275 assert community_updated.slug == "test-community-2"
277 node_updated = session.execute(select(Node).where(Node.id == community_updated.parent_node_id)).scalar_one()
278 assert node_updated.parent_node_id == community_2.parent_node_id
281def test_MakeUserVolunteer(db):
282 """MakeUserVolunteer should successfully create a volunteer"""
283 editor_user, editor_token = generate_user(is_editor=True)
284 normal_user, normal_token = generate_user()
286 refresh_materialized_views_rapid(empty_pb2.Empty())
287 with session_scope() as session:
288 with real_editor_session(editor_token) as api:
289 res = api.MakeUserVolunteer(
290 editor_pb2.MakeUserVolunteerReq(
291 user_id=normal_user.id,
292 role="Test Volunteer",
293 started_volunteering="2024-01-15",
294 hide_on_team_page=False,
295 )
296 )
298 # Check response
299 assert res.user_id == normal_user.id
300 assert res.role == "Test Volunteer"
301 assert res.started_volunteering == "2024-01-15"
302 assert res.show_on_team_page is True
303 assert res.username == normal_user.username
304 assert res.name == normal_user.name
306 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
307 assert volunteer.role == "Test Volunteer"
308 assert volunteer.started_volunteering.isoformat() == "2024-01-15"
309 assert volunteer.show_on_team_page is True
312def test_MakeUserVolunteer_default_values(db):
313 """MakeUserVolunteer should use default values when not provided"""
314 editor_user, editor_token = generate_user(is_editor=True)
315 normal_user, normal_token = generate_user()
317 refresh_materialized_views_rapid(empty_pb2.Empty())
318 with session_scope() as session:
319 with real_editor_session(editor_token) as api:
320 api.MakeUserVolunteer(
321 editor_pb2.MakeUserVolunteerReq(
322 user_id=normal_user.id,
323 role="Test Volunteer",
324 )
325 )
327 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
328 assert volunteer.role == "Test Volunteer"
329 assert volunteer.started_volunteering # defaults to today
330 assert volunteer.show_on_team_page is True # hide_on_team_page defaults to False
333def test_MakeUserVolunteer_hide_on_team_page(db):
334 """MakeUserVolunteer should respect hide_on_team_page=True"""
335 editor_user, editor_token = generate_user(is_editor=True)
336 normal_user, normal_token = generate_user()
338 refresh_materialized_views_rapid(empty_pb2.Empty())
339 with session_scope() as session:
340 with real_editor_session(editor_token) as api:
341 api.MakeUserVolunteer(
342 editor_pb2.MakeUserVolunteerReq(
343 user_id=normal_user.id,
344 role="Test Volunteer",
345 hide_on_team_page=True,
346 )
347 )
349 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
350 assert volunteer.role == "Test Volunteer"
351 assert volunteer.show_on_team_page is False # hide_on_team_page=True means don't show
354def test_MakeUserVolunteer_user_not_found(db):
355 """MakeUserVolunteer should fail if user doesn't exist"""
356 editor_user, editor_token = generate_user(is_editor=True)
358 with real_editor_session(editor_token) as api:
359 with pytest.raises(grpc.RpcError) as e:
360 api.MakeUserVolunteer(
361 editor_pb2.MakeUserVolunteerReq(
362 user_id=999999,
363 role="Test Volunteer",
364 )
365 )
366 assert e.value.code() == grpc.StatusCode.NOT_FOUND
367 assert e.value.details() == "Couldn't find that user."
370def test_MakeUserVolunteer_already_volunteer(db):
371 """MakeUserVolunteer should fail if user is already a volunteer"""
372 editor_user, editor_token = generate_user(is_editor=True)
373 normal_user, normal_token = generate_user()
375 refresh_materialized_views_rapid(empty_pb2.Empty())
376 with real_editor_session(editor_token) as api:
377 # Create volunteer first time
378 api.MakeUserVolunteer(
379 editor_pb2.MakeUserVolunteerReq(
380 user_id=normal_user.id,
381 role="Test Volunteer",
382 )
383 )
385 # Try to create again
386 with pytest.raises(grpc.RpcError) as e:
387 api.MakeUserVolunteer(
388 editor_pb2.MakeUserVolunteerReq(
389 user_id=normal_user.id,
390 role="Test Volunteer 2",
391 )
392 )
393 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
394 assert e.value.details() == "This user is already a volunteer."
397def test_MakeUserVolunteer_invalid_date(db):
398 """MakeUserVolunteer should fail with invalid date format"""
399 editor_user, editor_token = generate_user(is_editor=True)
400 normal_user, normal_token = generate_user()
402 with real_editor_session(editor_token) as api:
403 with pytest.raises(grpc.RpcError) as e:
404 api.MakeUserVolunteer(
405 editor_pb2.MakeUserVolunteerReq(
406 user_id=normal_user.id,
407 role="Test Volunteer",
408 started_volunteering="invalid-date",
409 )
410 )
411 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
412 assert e.value.details() == "Invalid start date for volunteering."
415def test_UpdateVolunteer(db):
416 """UpdateVolunteer should successfully update volunteer fields"""
417 editor_user, editor_token = generate_user(is_editor=True)
418 normal_user, normal_token = generate_user()
420 refresh_materialized_views_rapid(empty_pb2.Empty())
421 with session_scope() as session:
422 with real_editor_session(editor_token) as api:
423 # Create volunteer first
424 api.MakeUserVolunteer(
425 editor_pb2.MakeUserVolunteerReq(
426 user_id=normal_user.id,
427 role="Test Volunteer",
428 )
429 )
431 # Update volunteer
432 res = api.UpdateVolunteer(
433 editor_pb2.UpdateVolunteerReq(
434 user_id=normal_user.id,
435 role=StringValue(value="Updated Volunteer"),
436 sort_key=DoubleValue(value=10.5),
437 started_volunteering=StringValue(value="2023-06-01"),
438 stopped_volunteering=StringValue(value="2024-12-31"),
439 show_on_team_page=BoolValue(value=False),
440 )
441 )
443 # Check response
444 assert res.user_id == normal_user.id
445 assert res.role == "Updated Volunteer"
446 assert res.sort_key == 10.5
447 assert res.started_volunteering == "2023-06-01"
448 assert res.stopped_volunteering == "2024-12-31"
449 assert res.show_on_team_page is False
450 assert res.username == normal_user.username
452 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
453 assert volunteer.role == "Updated Volunteer"
454 assert volunteer.sort_key == 10.5
455 assert volunteer.started_volunteering.isoformat() == "2023-06-01"
456 assert volunteer.stopped_volunteering
457 assert volunteer.stopped_volunteering.isoformat() == "2024-12-31"
458 assert volunteer.show_on_team_page is False
461def test_UpdateVolunteer_partial_update(db):
462 """UpdateVolunteer should only update provided fields"""
463 editor_user, editor_token = generate_user(is_editor=True)
464 normal_user, normal_token = generate_user()
466 refresh_materialized_views_rapid(empty_pb2.Empty())
467 with session_scope() as session:
468 with real_editor_session(editor_token) as api:
469 # Create volunteer first
470 api.MakeUserVolunteer(
471 editor_pb2.MakeUserVolunteerReq(
472 user_id=normal_user.id,
473 role="Test Volunteer",
474 started_volunteering="2024-01-01",
475 )
476 )
478 # Update only role
479 api.UpdateVolunteer(
480 editor_pb2.UpdateVolunteerReq(
481 user_id=normal_user.id,
482 role=StringValue(value="Updated Role"),
483 )
484 )
486 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
487 assert volunteer.role == "Updated Role"
488 assert volunteer.started_volunteering.isoformat() == "2024-01-01" # unchanged
489 assert volunteer.show_on_team_page is True # unchanged
492def test_UpdateVolunteer_not_found(db):
493 """UpdateVolunteer should fail if volunteer doesn't exist"""
494 editor_user, editor_token = generate_user(is_editor=True)
495 normal_user, normal_token = generate_user()
497 with real_editor_session(editor_token) as api:
498 with pytest.raises(grpc.RpcError) as e:
499 api.UpdateVolunteer(
500 editor_pb2.UpdateVolunteerReq(
501 user_id=normal_user.id,
502 role=StringValue(value="Updated Volunteer"),
503 )
504 )
505 assert e.value.code() == grpc.StatusCode.NOT_FOUND
506 assert e.value.details() == "Volunteer not found."
509def test_UpdateVolunteer_invalid_started_date(db):
510 """UpdateVolunteer should fail with invalid started_volunteering date"""
511 editor_user, editor_token = generate_user(is_editor=True)
512 normal_user, normal_token = generate_user()
514 refresh_materialized_views_rapid(empty_pb2.Empty())
515 with real_editor_session(editor_token) as api:
516 # Create volunteer first
517 api.MakeUserVolunteer(
518 editor_pb2.MakeUserVolunteerReq(
519 user_id=normal_user.id,
520 role="Test Volunteer",
521 )
522 )
524 # Try to update with invalid date
525 with pytest.raises(grpc.RpcError) as e:
526 api.UpdateVolunteer(
527 editor_pb2.UpdateVolunteerReq(
528 user_id=normal_user.id,
529 started_volunteering=StringValue(value="invalid-date"),
530 )
531 )
532 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
533 assert e.value.details() == "Invalid start date for volunteering."
536def test_UpdateVolunteer_invalid_stopped_date(db):
537 """UpdateVolunteer should fail with invalid stopped_volunteering date"""
538 editor_user, editor_token = generate_user(is_editor=True)
539 normal_user, normal_token = generate_user()
541 refresh_materialized_views_rapid(empty_pb2.Empty())
542 with real_editor_session(editor_token) as api:
543 # Create volunteer first
544 api.MakeUserVolunteer(
545 editor_pb2.MakeUserVolunteerReq(
546 user_id=normal_user.id,
547 role="Test Volunteer",
548 )
549 )
551 # Try to update with invalid date
552 with pytest.raises(grpc.RpcError) as e:
553 api.UpdateVolunteer(
554 editor_pb2.UpdateVolunteerReq(
555 user_id=normal_user.id,
556 stopped_volunteering=StringValue(value="not-a-date"),
557 )
558 )
559 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
560 assert e.value.details() == "Invalid end date for volunteering."
563def test_UpdateVolunteer_reinstate(db):
564 """UpdateVolunteer should clear stopped_volunteering when reinstate_volunteer=True"""
565 editor_user, editor_token = generate_user(is_editor=True)
566 normal_user, normal_token = generate_user()
568 refresh_materialized_views_rapid(empty_pb2.Empty())
569 with real_editor_session(editor_token) as api:
570 api.MakeUserVolunteer(
571 editor_pb2.MakeUserVolunteerReq(
572 user_id=normal_user.id,
573 role="Test Volunteer",
574 )
575 )
577 # Set a stopped date first (make them a former volunteer)
578 api.UpdateVolunteer(
579 editor_pb2.UpdateVolunteerReq(
580 user_id=normal_user.id,
581 stopped_volunteering=StringValue(value="2024-12-31"),
582 )
583 )
584 with session_scope() as session:
585 volunteer_before = session.execute(
586 select(Volunteer).where(Volunteer.user_id == normal_user.id)
587 ).scalar_one()
588 assert volunteer_before.stopped_volunteering is not None
590 # Reinstate them
591 res = api.UpdateVolunteer(
592 editor_pb2.UpdateVolunteerReq(
593 user_id=normal_user.id,
594 reinstate_volunteer=True,
595 )
596 )
597 assert not res.HasField("stopped_volunteering")
599 with session_scope() as session:
600 volunteer_after = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
601 assert volunteer_after.stopped_volunteering is None
604def test_UpdateVolunteer_reinstate_already_current(db):
605 """UpdateVolunteer with reinstate_volunteer=True on a current volunteer is a no-op"""
606 editor_user, editor_token = generate_user(is_editor=True)
607 normal_user, normal_token = generate_user()
609 refresh_materialized_views_rapid(empty_pb2.Empty())
610 with real_editor_session(editor_token) as api:
611 api.MakeUserVolunteer(
612 editor_pb2.MakeUserVolunteerReq(
613 user_id=normal_user.id,
614 role="Test Volunteer",
615 )
616 )
618 res = api.UpdateVolunteer(
619 editor_pb2.UpdateVolunteerReq(
620 user_id=normal_user.id,
621 reinstate_volunteer=True,
622 )
623 )
624 assert not res.HasField("stopped_volunteering")
626 with session_scope() as session:
627 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
628 assert volunteer.stopped_volunteering is None
631def test_UpdateVolunteer_reinstate_conflict_with_stopped(db):
632 """UpdateVolunteer should fail if reinstate_volunteer=True and stopped_volunteering is also set"""
633 editor_user, editor_token = generate_user(is_editor=True)
634 normal_user, normal_token = generate_user()
636 refresh_materialized_views_rapid(empty_pb2.Empty())
637 with real_editor_session(editor_token) as api:
638 api.MakeUserVolunteer(
639 editor_pb2.MakeUserVolunteerReq(
640 user_id=normal_user.id,
641 role="Test Volunteer",
642 )
643 )
645 with pytest.raises(grpc.RpcError) as e:
646 api.UpdateVolunteer(
647 editor_pb2.UpdateVolunteerReq(
648 user_id=normal_user.id,
649 reinstate_volunteer=True,
650 stopped_volunteering=StringValue(value="2024-12-31"),
651 )
652 )
653 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
654 assert e.value.details() == "Cannot reinstate a volunteer and set a stopped date at the same time."
657def test_ListVolunteers(db):
658 """ListVolunteers should return all current volunteers"""
659 editor_user, editor_token = generate_user(is_editor=True)
660 user1, _ = generate_user()
661 user2, _ = generate_user()
662 user3, _ = generate_user()
664 refresh_materialized_views_rapid(empty_pb2.Empty())
665 with session_scope() as session:
666 with real_editor_session(editor_token) as api:
667 # Create three volunteers
668 api.MakeUserVolunteer(
669 editor_pb2.MakeUserVolunteerReq(
670 user_id=user1.id,
671 role="Volunteer 1",
672 started_volunteering="2024-01-15",
673 )
674 )
675 api.MakeUserVolunteer(
676 editor_pb2.MakeUserVolunteerReq(
677 user_id=user2.id,
678 role="Volunteer 2",
679 started_volunteering="2023-06-01",
680 )
681 )
682 api.MakeUserVolunteer(
683 editor_pb2.MakeUserVolunteerReq(
684 user_id=user3.id,
685 role="Volunteer 3",
686 started_volunteering="2024-03-20",
687 )
688 )
690 # List volunteers (only current ones by default)
691 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
693 assert len(res.volunteers) == 3
694 user_ids = {v.user_id for v in res.volunteers}
695 assert user_ids == {user1.id, user2.id, user3.id}
697 # Check that all fields are populated
698 for volunteer in res.volunteers:
699 assert volunteer.user_id > 0
700 assert volunteer.role != ""
701 assert volunteer.username != ""
702 assert volunteer.name != ""
703 assert volunteer.started_volunteering != ""
704 assert volunteer.show_on_team_page is True
707def test_ListVolunteers_with_past(db):
708 """ListVolunteers should include past volunteers when requested"""
709 editor_user, editor_token = generate_user(is_editor=True)
710 user1, _ = generate_user()
711 user2, _ = generate_user()
713 refresh_materialized_views_rapid(empty_pb2.Empty())
714 with session_scope() as session:
715 with real_editor_session(editor_token) as api:
716 # Create current volunteer
717 api.MakeUserVolunteer(
718 editor_pb2.MakeUserVolunteerReq(
719 user_id=user1.id,
720 role="Current Volunteer",
721 )
722 )
724 # Create past volunteer
725 api.MakeUserVolunteer(
726 editor_pb2.MakeUserVolunteerReq(
727 user_id=user2.id,
728 role="Past Volunteer",
729 )
730 )
731 api.UpdateVolunteer(
732 editor_pb2.UpdateVolunteerReq(
733 user_id=user2.id,
734 stopped_volunteering=StringValue(value="2024-06-30"),
735 )
736 )
738 # List only current volunteers
739 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
740 assert len(res.volunteers) == 1
741 assert res.volunteers[0].user_id == user1.id
742 assert not res.volunteers[0].HasField("stopped_volunteering")
744 # List all volunteers (including past)
745 res_with_past = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=True))
746 assert len(res_with_past.volunteers) == 2
747 user_ids = {v.user_id for v in res_with_past.volunteers}
748 assert user_ids == {user1.id, user2.id}
750 # Find the past volunteer and verify stopped_volunteering is set
751 past_volunteer = next(v for v in res_with_past.volunteers if v.user_id == user2.id)
752 assert past_volunteer.stopped_volunteering == "2024-06-30"
755def test_ListVolunteers_ordering(db):
756 """ListVolunteers should respect sort_key ordering"""
757 editor_user, editor_token = generate_user(is_editor=True)
758 user1, _ = generate_user()
759 user2, _ = generate_user()
760 user3, _ = generate_user()
762 refresh_materialized_views_rapid(empty_pb2.Empty())
763 with session_scope() as session:
764 with real_editor_session(editor_token) as api:
765 # Create volunteers with different sort keys
766 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user1.id, role="Volunteer 1"))
767 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user1.id, sort_key=DoubleValue(value=30.0)))
769 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user2.id, role="Volunteer 2"))
770 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user2.id, sort_key=DoubleValue(value=10.0)))
772 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user3.id, role="Volunteer 3"))
773 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user3.id, sort_key=DoubleValue(value=20.0)))
775 # List volunteers - should be ordered by sort_key ascending
776 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
777 assert len(res.volunteers) == 3
778 assert res.volunteers[0].user_id == user2.id # sort_key 10.0
779 assert res.volunteers[1].user_id == user3.id # sort_key 20.0
780 assert res.volunteers[2].user_id == user1.id # sort_key 30.0
783def test_ListVolunteers_empty(db):
784 """ListVolunteers should return empty list when no volunteers exist"""
785 editor_user, editor_token = generate_user(is_editor=True)
787 refresh_materialized_views_rapid(empty_pb2.Empty())
788 with real_editor_session(editor_token) as api:
789 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
790 assert len(res.volunteers) == 0