Coverage for app/backend/src/tests/test_editor.py: 100%
463 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1from datetime import timedelta
3import grpc
4import pytest
5from google.protobuf import empty_pb2
6from google.protobuf.wrappers_pb2 import BoolValue, DoubleValue, StringValue
7from sqlalchemy import select, update
9from couchers.db import session_scope
10from couchers.materialized_views import refresh_materialized_views_rapid
11from couchers.models import (
12 Cluster,
13 Event,
14 Node,
15 User,
16 UserActivity,
17 Volunteer,
18)
19from couchers.proto import editor_pb2, events_pb2
20from couchers.tasks import enforce_community_memberships
21from couchers.utils import now
22from tests.fixtures.api import events as events_api
23from tests.fixtures.db import generate_user
24from tests.fixtures.sessions import real_editor_session
25from tests.test_communities import create_1d_point, create_community
26from tests.utils import local_date_time
28VALID_GEOJSON_MULTIPOLYGON = """
29 {
30 "type": "MultiPolygon",
31 "coordinates":
32 [
33 [
34 [
35 [
36 -73.98114904754641,
37 40.7470284264813
38 ],
39 [
40 -73.98314135177611,
41 40.73416844413217
42 ],
43 [
44 -74.00538969848634,
45 40.734314779027144
46 ],
47 [
48 -74.00479214294432,
49 40.75027851544338
50 ],
51 [
52 -73.98114904754641,
53 40.7470284264813
54 ]
55 ]
56 ]
57 ]
58 }
59"""
61POINT_GEOJSON = """
62{ "type": "Point", "coordinates": [100.0, 0.0] }
63"""
66def test_access_by_normal_user(db):
67 """Normal users should not be able to access editor APIs"""
68 normal_user, normal_token = generate_user()
70 with real_editor_session(normal_token) as api:
71 with pytest.raises(grpc.RpcError) as e:
72 api.CreateCommunity(
73 editor_pb2.CreateCommunityReq(
74 name="test community",
75 description="community for testing",
76 admin_ids=[],
77 geojson=VALID_GEOJSON_MULTIPOLYGON,
78 )
79 )
80 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
81 assert e.value.details() == "Permission denied"
84def test_access_by_editor_user(db):
85 """Editor users should be able to access editor APIs"""
86 editor_user, editor_token = generate_user(is_editor=True)
88 with session_scope() as session:
89 with real_editor_session(editor_token) as api:
90 api.CreateCommunity(
91 editor_pb2.CreateCommunityReq(
92 name="test community",
93 description="community for testing",
94 admin_ids=[],
95 geojson=VALID_GEOJSON_MULTIPOLYGON,
96 )
97 )
98 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
99 assert community.description == "community for testing"
100 assert community.slug == "test-community"
103def test_access_by_superuser(db):
104 """Superusers (who are also editors) should be able to access editor APIs"""
105 editor_user, editor_token = generate_user(is_editor=True)
107 with session_scope() as session:
108 with real_editor_session(editor_token) as api:
109 api.CreateCommunity(
110 editor_pb2.CreateCommunityReq(
111 name="test community",
112 description="community for testing",
113 admin_ids=[],
114 geojson=VALID_GEOJSON_MULTIPOLYGON,
115 )
116 )
117 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
118 assert community.description == "community for testing"
119 assert community.slug == "test-community"
122def test_CreateCommunity_invalid_geojson(db):
123 """CreateCommunity should reject invalid GeoJSON"""
124 editor_user, editor_token = generate_user(is_editor=True)
126 with real_editor_session(editor_token) as api:
127 with pytest.raises(grpc.RpcError) as e:
128 api.CreateCommunity(
129 editor_pb2.CreateCommunityReq(
130 name="test community",
131 description="community for testing",
132 admin_ids=[],
133 geojson=POINT_GEOJSON,
134 )
135 )
136 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
137 assert e.value.details() == "GeoJson was not of type MultiPolygon."
140def test_UpdateCommunity_invalid_geojson(db):
141 """UpdateCommunity should reject invalid GeoJSON"""
142 editor_user, editor_token = generate_user(is_editor=True)
144 with session_scope() as session:
145 with real_editor_session(editor_token) as api:
146 api.CreateCommunity(
147 editor_pb2.CreateCommunityReq(
148 name="test community",
149 description="community for testing",
150 admin_ids=[],
151 geojson=VALID_GEOJSON_MULTIPOLYGON,
152 )
153 )
154 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
156 with pytest.raises(grpc.RpcError) as e:
157 api.UpdateCommunity(
158 editor_pb2.UpdateCommunityReq(
159 community_id=community.parent_node_id,
160 name="test community 2",
161 description="community for testing 2",
162 geojson=POINT_GEOJSON,
163 )
164 )
165 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
166 assert e.value.details() == "GeoJson was not of type MultiPolygon."
169def test_UpdateCommunity_invalid_id(db):
170 """UpdateCommunity should reject invalid community IDs"""
171 editor_user, editor_token = generate_user(is_editor=True)
173 with real_editor_session(editor_token) as api:
174 api.CreateCommunity(
175 editor_pb2.CreateCommunityReq(
176 name="test community",
177 description="community for testing",
178 admin_ids=[],
179 geojson=VALID_GEOJSON_MULTIPOLYGON,
180 )
181 )
183 with pytest.raises(grpc.RpcError) as e:
184 api.UpdateCommunity(
185 editor_pb2.UpdateCommunityReq(
186 community_id=1000,
187 name="test community 1000",
188 description="community for testing 1000",
189 geojson=VALID_GEOJSON_MULTIPOLYGON,
190 )
191 )
192 assert e.value.code() == grpc.StatusCode.NOT_FOUND
193 assert e.value.details() == "Community not found."
196def test_UpdateCommunity(db):
197 """UpdateCommunity should successfully update a community"""
198 editor_user, editor_token = generate_user(is_editor=True)
200 with session_scope() as session:
201 with real_editor_session(editor_token) as api:
202 api.CreateCommunity(
203 editor_pb2.CreateCommunityReq(
204 name="test community",
205 description="community for testing",
206 admin_ids=[],
207 geojson=VALID_GEOJSON_MULTIPOLYGON,
208 )
209 )
210 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
212 api.UpdateCommunity(
213 editor_pb2.UpdateCommunityReq(
214 community_id=community.parent_node_id,
215 name="test community updated",
216 description="community for testing updated",
217 geojson=VALID_GEOJSON_MULTIPOLYGON,
218 )
219 )
220 session.commit()
222 community_updated = session.execute(select(Cluster).where(Cluster.id == community.id)).scalar_one()
223 assert community_updated.name == "test community updated"
224 assert community_updated.description == "community for testing updated"
225 assert community_updated.slug == "test-community-updated"
228def test_CreateCommunity(db):
229 with session_scope() as session:
230 editor_user, editor_token = generate_user(is_editor=True)
231 normal_user, normal_token = generate_user()
232 with real_editor_session(editor_token) as api:
233 api.CreateCommunity(
234 editor_pb2.CreateCommunityReq(
235 name="test community",
236 description="community for testing",
237 admin_ids=[],
238 geojson=VALID_GEOJSON_MULTIPOLYGON,
239 )
240 )
241 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
242 assert community.description == "community for testing"
243 assert community.slug == "test-community"
246def test_UpdateCommunity2(db):
247 editor_user, editor_token = generate_user(is_editor=True)
249 with session_scope() as session:
250 with real_editor_session(editor_token) as api:
251 api.CreateCommunity(
252 editor_pb2.CreateCommunityReq(
253 name="test community",
254 description="community for testing",
255 admin_ids=[],
256 geojson=VALID_GEOJSON_MULTIPOLYGON,
257 )
258 )
259 community = session.execute(select(Cluster).where(Cluster.name == "test community")).scalar_one()
260 assert community.description == "community for testing"
262 api.CreateCommunity(
263 editor_pb2.CreateCommunityReq(
264 name="test community 2",
265 description="community for testing 2",
266 admin_ids=[],
267 geojson=VALID_GEOJSON_MULTIPOLYGON,
268 )
269 )
270 community_2 = session.execute(select(Cluster).where(Cluster.name == "test community 2")).scalar_one()
272 api.UpdateCommunity(
273 editor_pb2.UpdateCommunityReq(
274 community_id=community.parent_node_id,
275 name="test community 2",
276 description="community for testing 2",
277 geojson=VALID_GEOJSON_MULTIPOLYGON,
278 parent_node_id=community_2.parent_node_id,
279 )
280 )
281 session.commit()
283 community_updated = session.execute(select(Cluster).where(Cluster.id == community.id)).scalar_one()
284 assert community_updated.description == "community for testing 2"
285 assert community_updated.slug == "test-community-2"
287 node_updated = session.execute(select(Node).where(Node.id == community_updated.parent_node_id)).scalar_one()
288 assert node_updated.parent_node_id == community_2.parent_node_id
291def test_MakeUserVolunteer(db):
292 """MakeUserVolunteer should successfully create a volunteer"""
293 editor_user, editor_token = generate_user(is_editor=True)
294 normal_user, normal_token = generate_user()
296 refresh_materialized_views_rapid(empty_pb2.Empty())
297 with session_scope() as session:
298 with real_editor_session(editor_token) as api:
299 res = api.MakeUserVolunteer(
300 editor_pb2.MakeUserVolunteerReq(
301 user_id=normal_user.id,
302 role="Test Volunteer",
303 started_volunteering="2024-01-15",
304 hide_on_team_page=False,
305 )
306 )
308 # Check response
309 assert res.user_id == normal_user.id
310 assert res.role == "Test Volunteer"
311 assert res.started_volunteering == "2024-01-15"
312 assert res.show_on_team_page is True
313 assert res.username == normal_user.username
314 assert res.name == normal_user.name
316 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
317 assert volunteer.role == "Test Volunteer"
318 assert volunteer.started_volunteering.isoformat() == "2024-01-15"
319 assert volunteer.show_on_team_page is True
322def test_MakeUserVolunteer_default_values(db):
323 """MakeUserVolunteer should use default values when not provided"""
324 editor_user, editor_token = generate_user(is_editor=True)
325 normal_user, normal_token = generate_user()
327 refresh_materialized_views_rapid(empty_pb2.Empty())
328 with session_scope() as session:
329 with real_editor_session(editor_token) as api:
330 api.MakeUserVolunteer(
331 editor_pb2.MakeUserVolunteerReq(
332 user_id=normal_user.id,
333 role="Test Volunteer",
334 )
335 )
337 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
338 assert volunteer.role == "Test Volunteer"
339 assert volunteer.started_volunteering # defaults to today
340 assert volunteer.show_on_team_page is True # hide_on_team_page defaults to False
343def test_MakeUserVolunteer_hide_on_team_page(db):
344 """MakeUserVolunteer should respect hide_on_team_page=True"""
345 editor_user, editor_token = generate_user(is_editor=True)
346 normal_user, normal_token = generate_user()
348 refresh_materialized_views_rapid(empty_pb2.Empty())
349 with session_scope() as session:
350 with real_editor_session(editor_token) as api:
351 api.MakeUserVolunteer(
352 editor_pb2.MakeUserVolunteerReq(
353 user_id=normal_user.id,
354 role="Test Volunteer",
355 hide_on_team_page=True,
356 )
357 )
359 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
360 assert volunteer.role == "Test Volunteer"
361 assert volunteer.show_on_team_page is False # hide_on_team_page=True means don't show
364def test_MakeUserVolunteer_user_not_found(db):
365 """MakeUserVolunteer should fail if user doesn't exist"""
366 editor_user, editor_token = generate_user(is_editor=True)
368 with real_editor_session(editor_token) as api:
369 with pytest.raises(grpc.RpcError) as e:
370 api.MakeUserVolunteer(
371 editor_pb2.MakeUserVolunteerReq(
372 user_id=999999,
373 role="Test Volunteer",
374 )
375 )
376 assert e.value.code() == grpc.StatusCode.NOT_FOUND
377 assert e.value.details() == "Couldn't find that user."
380def test_MakeUserVolunteer_already_volunteer(db):
381 """MakeUserVolunteer should fail if user is already a volunteer"""
382 editor_user, editor_token = generate_user(is_editor=True)
383 normal_user, normal_token = generate_user()
385 refresh_materialized_views_rapid(empty_pb2.Empty())
386 with real_editor_session(editor_token) as api:
387 # Create volunteer first time
388 api.MakeUserVolunteer(
389 editor_pb2.MakeUserVolunteerReq(
390 user_id=normal_user.id,
391 role="Test Volunteer",
392 )
393 )
395 # Try to create again
396 with pytest.raises(grpc.RpcError) as e:
397 api.MakeUserVolunteer(
398 editor_pb2.MakeUserVolunteerReq(
399 user_id=normal_user.id,
400 role="Test Volunteer 2",
401 )
402 )
403 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
404 assert e.value.details() == "This user is already a volunteer."
407def test_MakeUserVolunteer_invalid_date(db):
408 """MakeUserVolunteer should fail with invalid date format"""
409 editor_user, editor_token = generate_user(is_editor=True)
410 normal_user, normal_token = generate_user()
412 with real_editor_session(editor_token) as api:
413 with pytest.raises(grpc.RpcError) as e:
414 api.MakeUserVolunteer(
415 editor_pb2.MakeUserVolunteerReq(
416 user_id=normal_user.id,
417 role="Test Volunteer",
418 started_volunteering="invalid-date",
419 )
420 )
421 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
422 assert e.value.details() == "Invalid start date for volunteering."
425def test_UpdateVolunteer(db):
426 """UpdateVolunteer should successfully update volunteer fields"""
427 editor_user, editor_token = generate_user(is_editor=True)
428 normal_user, normal_token = generate_user()
430 refresh_materialized_views_rapid(empty_pb2.Empty())
431 with session_scope() as session:
432 with real_editor_session(editor_token) as api:
433 # Create volunteer first
434 api.MakeUserVolunteer(
435 editor_pb2.MakeUserVolunteerReq(
436 user_id=normal_user.id,
437 role="Test Volunteer",
438 )
439 )
441 # Update volunteer
442 res = api.UpdateVolunteer(
443 editor_pb2.UpdateVolunteerReq(
444 user_id=normal_user.id,
445 role=StringValue(value="Updated Volunteer"),
446 sort_key=DoubleValue(value=10.5),
447 started_volunteering=StringValue(value="2023-06-01"),
448 stopped_volunteering=StringValue(value="2024-12-31"),
449 show_on_team_page=BoolValue(value=False),
450 )
451 )
453 # Check response
454 assert res.user_id == normal_user.id
455 assert res.role == "Updated Volunteer"
456 assert res.sort_key == 10.5
457 assert res.started_volunteering == "2023-06-01"
458 assert res.stopped_volunteering == "2024-12-31"
459 assert res.show_on_team_page is False
460 assert res.username == normal_user.username
462 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
463 assert volunteer.role == "Updated Volunteer"
464 assert volunteer.sort_key == 10.5
465 assert volunteer.started_volunteering.isoformat() == "2023-06-01"
466 assert volunteer.stopped_volunteering
467 assert volunteer.stopped_volunteering.isoformat() == "2024-12-31"
468 assert volunteer.show_on_team_page is False
471def test_UpdateVolunteer_partial_update(db):
472 """UpdateVolunteer should only update provided fields"""
473 editor_user, editor_token = generate_user(is_editor=True)
474 normal_user, normal_token = generate_user()
476 refresh_materialized_views_rapid(empty_pb2.Empty())
477 with session_scope() as session:
478 with real_editor_session(editor_token) as api:
479 # Create volunteer first
480 api.MakeUserVolunteer(
481 editor_pb2.MakeUserVolunteerReq(
482 user_id=normal_user.id,
483 role="Test Volunteer",
484 started_volunteering="2024-01-01",
485 )
486 )
488 # Update only role
489 api.UpdateVolunteer(
490 editor_pb2.UpdateVolunteerReq(
491 user_id=normal_user.id,
492 role=StringValue(value="Updated Role"),
493 )
494 )
496 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
497 assert volunteer.role == "Updated Role"
498 assert volunteer.started_volunteering.isoformat() == "2024-01-01" # unchanged
499 assert volunteer.show_on_team_page is True # unchanged
502def test_UpdateVolunteer_not_found(db):
503 """UpdateVolunteer should fail if volunteer doesn't exist"""
504 editor_user, editor_token = generate_user(is_editor=True)
505 normal_user, normal_token = generate_user()
507 with real_editor_session(editor_token) as api:
508 with pytest.raises(grpc.RpcError) as e:
509 api.UpdateVolunteer(
510 editor_pb2.UpdateVolunteerReq(
511 user_id=normal_user.id,
512 role=StringValue(value="Updated Volunteer"),
513 )
514 )
515 assert e.value.code() == grpc.StatusCode.NOT_FOUND
516 assert e.value.details() == "Volunteer not found."
519def test_UpdateVolunteer_invalid_started_date(db):
520 """UpdateVolunteer should fail with invalid started_volunteering date"""
521 editor_user, editor_token = generate_user(is_editor=True)
522 normal_user, normal_token = generate_user()
524 refresh_materialized_views_rapid(empty_pb2.Empty())
525 with real_editor_session(editor_token) as api:
526 # Create volunteer first
527 api.MakeUserVolunteer(
528 editor_pb2.MakeUserVolunteerReq(
529 user_id=normal_user.id,
530 role="Test Volunteer",
531 )
532 )
534 # Try to update with invalid date
535 with pytest.raises(grpc.RpcError) as e:
536 api.UpdateVolunteer(
537 editor_pb2.UpdateVolunteerReq(
538 user_id=normal_user.id,
539 started_volunteering=StringValue(value="invalid-date"),
540 )
541 )
542 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
543 assert e.value.details() == "Invalid start date for volunteering."
546def test_UpdateVolunteer_invalid_stopped_date(db):
547 """UpdateVolunteer should fail with invalid stopped_volunteering date"""
548 editor_user, editor_token = generate_user(is_editor=True)
549 normal_user, normal_token = generate_user()
551 refresh_materialized_views_rapid(empty_pb2.Empty())
552 with real_editor_session(editor_token) as api:
553 # Create volunteer first
554 api.MakeUserVolunteer(
555 editor_pb2.MakeUserVolunteerReq(
556 user_id=normal_user.id,
557 role="Test Volunteer",
558 )
559 )
561 # Try to update with invalid date
562 with pytest.raises(grpc.RpcError) as e:
563 api.UpdateVolunteer(
564 editor_pb2.UpdateVolunteerReq(
565 user_id=normal_user.id,
566 stopped_volunteering=StringValue(value="not-a-date"),
567 )
568 )
569 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
570 assert e.value.details() == "Invalid end date for volunteering."
573def test_UpdateVolunteer_reinstate(db):
574 """UpdateVolunteer should clear stopped_volunteering when reinstate_volunteer=True"""
575 editor_user, editor_token = generate_user(is_editor=True)
576 normal_user, normal_token = generate_user()
578 refresh_materialized_views_rapid(empty_pb2.Empty())
579 with real_editor_session(editor_token) as api:
580 api.MakeUserVolunteer(
581 editor_pb2.MakeUserVolunteerReq(
582 user_id=normal_user.id,
583 role="Test Volunteer",
584 )
585 )
587 # Set a stopped date first (make them a former volunteer)
588 api.UpdateVolunteer(
589 editor_pb2.UpdateVolunteerReq(
590 user_id=normal_user.id,
591 stopped_volunteering=StringValue(value="2024-12-31"),
592 )
593 )
594 with session_scope() as session:
595 volunteer_before = session.execute(
596 select(Volunteer).where(Volunteer.user_id == normal_user.id)
597 ).scalar_one()
598 assert volunteer_before.stopped_volunteering is not None
600 # Reinstate them
601 res = api.UpdateVolunteer(
602 editor_pb2.UpdateVolunteerReq(
603 user_id=normal_user.id,
604 reinstate_volunteer=True,
605 )
606 )
607 assert not res.HasField("stopped_volunteering")
609 with session_scope() as session:
610 volunteer_after = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
611 assert volunteer_after.stopped_volunteering is None
614def test_UpdateVolunteer_reinstate_already_current(db):
615 """UpdateVolunteer with reinstate_volunteer=True on a current volunteer is a no-op"""
616 editor_user, editor_token = generate_user(is_editor=True)
617 normal_user, normal_token = generate_user()
619 refresh_materialized_views_rapid(empty_pb2.Empty())
620 with real_editor_session(editor_token) as api:
621 api.MakeUserVolunteer(
622 editor_pb2.MakeUserVolunteerReq(
623 user_id=normal_user.id,
624 role="Test Volunteer",
625 )
626 )
628 res = api.UpdateVolunteer(
629 editor_pb2.UpdateVolunteerReq(
630 user_id=normal_user.id,
631 reinstate_volunteer=True,
632 )
633 )
634 assert not res.HasField("stopped_volunteering")
636 with session_scope() as session:
637 volunteer = session.execute(select(Volunteer).where(Volunteer.user_id == normal_user.id)).scalar_one()
638 assert volunteer.stopped_volunteering is None
641def test_UpdateVolunteer_reinstate_conflict_with_stopped(db):
642 """UpdateVolunteer should fail if reinstate_volunteer=True and stopped_volunteering is also set"""
643 editor_user, editor_token = generate_user(is_editor=True)
644 normal_user, normal_token = generate_user()
646 refresh_materialized_views_rapid(empty_pb2.Empty())
647 with real_editor_session(editor_token) as api:
648 api.MakeUserVolunteer(
649 editor_pb2.MakeUserVolunteerReq(
650 user_id=normal_user.id,
651 role="Test Volunteer",
652 )
653 )
655 with pytest.raises(grpc.RpcError) as e:
656 api.UpdateVolunteer(
657 editor_pb2.UpdateVolunteerReq(
658 user_id=normal_user.id,
659 reinstate_volunteer=True,
660 stopped_volunteering=StringValue(value="2024-12-31"),
661 )
662 )
663 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
664 assert e.value.details() == "Cannot reinstate a volunteer and set a stopped date at the same time."
667def test_ListVolunteers(db):
668 """ListVolunteers should return all current volunteers"""
669 editor_user, editor_token = generate_user(is_editor=True)
670 user1, _ = generate_user()
671 user2, _ = generate_user()
672 user3, _ = generate_user()
674 refresh_materialized_views_rapid(empty_pb2.Empty())
675 with session_scope() as session:
676 with real_editor_session(editor_token) as api:
677 # Create three volunteers
678 api.MakeUserVolunteer(
679 editor_pb2.MakeUserVolunteerReq(
680 user_id=user1.id,
681 role="Volunteer 1",
682 started_volunteering="2024-01-15",
683 )
684 )
685 api.MakeUserVolunteer(
686 editor_pb2.MakeUserVolunteerReq(
687 user_id=user2.id,
688 role="Volunteer 2",
689 started_volunteering="2023-06-01",
690 )
691 )
692 api.MakeUserVolunteer(
693 editor_pb2.MakeUserVolunteerReq(
694 user_id=user3.id,
695 role="Volunteer 3",
696 started_volunteering="2024-03-20",
697 )
698 )
700 # List volunteers (only current ones by default)
701 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
703 assert len(res.volunteers) == 3
704 user_ids = {v.user_id for v in res.volunteers}
705 assert user_ids == {user1.id, user2.id, user3.id}
707 # Check that all fields are populated
708 for volunteer in res.volunteers:
709 assert volunteer.user_id > 0
710 assert volunteer.role != ""
711 assert volunteer.username != ""
712 assert volunteer.name != ""
713 assert volunteer.started_volunteering != ""
714 assert volunteer.show_on_team_page is True
717def test_ListVolunteers_with_past(db):
718 """ListVolunteers should include past volunteers when requested"""
719 editor_user, editor_token = generate_user(is_editor=True)
720 user1, _ = generate_user()
721 user2, _ = generate_user()
723 refresh_materialized_views_rapid(empty_pb2.Empty())
724 with session_scope() as session:
725 with real_editor_session(editor_token) as api:
726 # Create current volunteer
727 api.MakeUserVolunteer(
728 editor_pb2.MakeUserVolunteerReq(
729 user_id=user1.id,
730 role="Current Volunteer",
731 )
732 )
734 # Create past volunteer
735 api.MakeUserVolunteer(
736 editor_pb2.MakeUserVolunteerReq(
737 user_id=user2.id,
738 role="Past Volunteer",
739 )
740 )
741 api.UpdateVolunteer(
742 editor_pb2.UpdateVolunteerReq(
743 user_id=user2.id,
744 stopped_volunteering=StringValue(value="2024-06-30"),
745 )
746 )
748 # List only current volunteers
749 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
750 assert len(res.volunteers) == 1
751 assert res.volunteers[0].user_id == user1.id
752 assert not res.volunteers[0].HasField("stopped_volunteering")
754 # List all volunteers (including past)
755 res_with_past = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=True))
756 assert len(res_with_past.volunteers) == 2
757 user_ids = {v.user_id for v in res_with_past.volunteers}
758 assert user_ids == {user1.id, user2.id}
760 # Find the past volunteer and verify stopped_volunteering is set
761 past_volunteer = next(v for v in res_with_past.volunteers if v.user_id == user2.id)
762 assert past_volunteer.stopped_volunteering == "2024-06-30"
765def test_ListVolunteers_ordering(db):
766 """ListVolunteers should respect sort_key ordering"""
767 editor_user, editor_token = generate_user(is_editor=True)
768 user1, _ = generate_user()
769 user2, _ = generate_user()
770 user3, _ = generate_user()
772 refresh_materialized_views_rapid(empty_pb2.Empty())
773 with session_scope() as session:
774 with real_editor_session(editor_token) as api:
775 # Create volunteers with different sort keys
776 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user1.id, role="Volunteer 1"))
777 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user1.id, sort_key=DoubleValue(value=30.0)))
779 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user2.id, role="Volunteer 2"))
780 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user2.id, sort_key=DoubleValue(value=10.0)))
782 api.MakeUserVolunteer(editor_pb2.MakeUserVolunteerReq(user_id=user3.id, role="Volunteer 3"))
783 api.UpdateVolunteer(editor_pb2.UpdateVolunteerReq(user_id=user3.id, sort_key=DoubleValue(value=20.0)))
785 # List volunteers - should be ordered by sort_key ascending
786 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
787 assert len(res.volunteers) == 3
788 assert res.volunteers[0].user_id == user2.id # sort_key 10.0
789 assert res.volunteers[1].user_id == user3.id # sort_key 20.0
790 assert res.volunteers[2].user_id == user1.id # sort_key 30.0
793def test_ListVolunteers_empty(db):
794 """ListVolunteers should return empty list when no volunteers exist"""
795 editor_user, editor_token = generate_user(is_editor=True)
797 refresh_materialized_views_rapid(empty_pb2.Empty())
798 with real_editor_session(editor_token) as api:
799 res = api.ListVolunteers(editor_pb2.ListVolunteersReq(include_past=False))
800 assert len(res.volunteers) == 0
803def test_LookupUsersForDebug(db):
804 editor_user, editor_token = generate_user(is_editor=True)
805 user1, _ = generate_user()
806 user2, _ = generate_user()
807 user3, _ = generate_user()
808 user4, _ = generate_user()
809 user5, _ = generate_user()
811 shared_sofa = "abcdefghijklMNOPQR"
812 other_sofa = "zyxwvutsrqpoNMLKJI"
813 wildcard_sofa = "abcdefghijkXMNOPQR"
815 with session_scope() as session:
816 session.execute(update(User).where(User.id == user2.id).values(banned_at=now()))
817 session.execute(update(User).where(User.id == user3.id).values(deleted_at=now()))
818 session.execute(update(User).where(User.id == user4.id).values(shadowed_at=now()))
820 session.add(UserActivity(user_id=user1.id, period=now(), sofa=shared_sofa, api_calls=5))
821 session.add(UserActivity(user_id=user2.id, period=now(), sofa=shared_sofa, api_calls=3))
822 session.add(UserActivity(user_id=user2.id, period=now(), sofa=other_sofa, api_calls=100))
823 session.add(UserActivity(user_id=user3.id, period=now(), sofa=shared_sofa, api_calls=1))
824 session.add(UserActivity(user_id=user4.id, period=now(), sofa=shared_sofa, api_calls=1))
825 session.add(UserActivity(user_id=user5.id, period=now(), sofa=wildcard_sofa, api_calls=1))
826 session.add(UserActivity(user_id=user5.id, period=now(), sofa=None, api_calls=1))
828 with real_editor_session(editor_token) as api:
829 res = api.LookupUsersForDebug(editor_pb2.LookupUsersForDebugReq(sofa_prefix="abcdefghijkl"))
831 assert [u.user_id for u in res.users] == [user1.id, user2.id, user3.id, user4.id]
832 assert [(u.banned, u.deleted, u.shadowed) for u in res.users] == [
833 (False, False, False),
834 (True, False, False),
835 (False, True, False),
836 (False, False, True),
837 ]
839 with real_editor_session(editor_token) as api:
840 res = api.LookupUsersForDebug(editor_pb2.LookupUsersForDebugReq(sofa_prefix="abcdefghijk_"))
841 assert list(res.users) == []
843 with real_editor_session(editor_token) as api:
844 for bad_prefix in ["", "abcdefghijk", "abcdefghijklm", shared_sofa]:
845 with pytest.raises(grpc.RpcError) as e:
846 api.LookupUsersForDebug(editor_pb2.LookupUsersForDebugReq(sofa_prefix=bad_prefix))
847 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
848 assert e.value.details() == "The sofa prefix must be exactly 12 characters."
851@pytest.fixture
852def event_in_wrong_community(db):
853 """An event physically in "City" but parented to the far too large "Country" community."""
854 editor_user, editor_token = generate_user(is_editor=True)
855 creator, creator_token = generate_user(complete_profile=True, geom=create_1d_point(3), geom_radius=0.1)
856 member1, _ = generate_user(geom=create_1d_point(3), geom_radius=0.1)
857 member2, _ = generate_user(geom=create_1d_point(4), geom_radius=0.1)
859 with session_scope() as session:
860 w = create_community(session, 0, 100, "Global", [editor_user], [], None)
861 mr = create_community(session, 0, 100, "Macroregion", [editor_user], [], w)
862 country = create_community(session, 0, 50, "Country", [editor_user], [], mr)
863 city = create_community(session, 0, 10, "City", [creator], [member1, member2], country)
864 country_id = country.id
865 city_id = city.id
867 enforce_community_memberships()
869 event_id = events_api.create_event(
870 token=creator_token,
871 parent_community_id=country_id,
872 location=events_pb2.EventLocation(address="Somewhere", lat=3, lng=1),
873 ).event_id
875 yield editor_token, creator_token, event_id, country_id, city_id
878def test_GetEventCommunityInfo(event_in_wrong_community):
879 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
881 with real_editor_session(editor_token) as api:
882 res = api.GetEventCommunityInfo(editor_pb2.GetEventCommunityInfoReq(event_id=event_id))
884 assert res.event_id == event_id
885 assert res.title == "Dummy Title"
886 assert res.address == "Somewhere"
887 assert res.occurrence_count == 1
888 assert not res.is_in_smallest_local_community
890 assert res.current.community.community_id == country_id
891 assert res.current.community.name == "Country"
892 assert res.current.too_large_for_notifications
893 assert res.current.approx_users_to_notify == 0
895 assert res.smallest_local.community.community_id == city_id
896 assert res.smallest_local.community.name == "City"
897 assert not res.smallest_local.too_large_for_notifications
898 assert not res.smallest_local.notifies_by_proximity
899 # the two other members of the city; the creator is an organizer so doesn't need an invite
900 assert res.smallest_local.approx_users_to_notify == 2
903def test_GetEventCommunityInfo_already_in_smallest_community(event_in_wrong_community):
904 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
906 correctly_placed_event_id = events_api.create_event(
907 token=creator_token,
908 parent_community_id=city_id,
909 location=events_pb2.EventLocation(address="Somewhere", lat=3, lng=1),
910 ).event_id
912 with real_editor_session(editor_token) as api:
913 res = api.GetEventCommunityInfo(editor_pb2.GetEventCommunityInfoReq(event_id=correctly_placed_event_id))
915 assert res.is_in_smallest_local_community
916 assert res.current.community.community_id == city_id
917 assert res.smallest_local.community.community_id == city_id
920def test_GetEventCommunityInfo_event_not_found(db):
921 _, editor_token = generate_user(is_editor=True)
923 with real_editor_session(editor_token) as api:
924 with pytest.raises(grpc.RpcError) as e:
925 api.GetEventCommunityInfo(editor_pb2.GetEventCommunityInfoReq(event_id=42))
926 assert e.value.code() == grpc.StatusCode.NOT_FOUND
927 assert e.value.details() == "Event not found."
930def test_GetEventCommunityInfo_access_by_normal_user(event_in_wrong_community):
931 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
933 with real_editor_session(creator_token) as api:
934 with pytest.raises(grpc.RpcError) as e:
935 api.GetEventCommunityInfo(editor_pb2.GetEventCommunityInfoReq(event_id=event_id))
936 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
939def test_MoveEventToCommunity_smallest_local_community(event_in_wrong_community):
940 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
942 with real_editor_session(editor_token) as api:
943 res = api.MoveEventToCommunity(
944 editor_pb2.MoveEventToCommunityReq(event_id=event_id, smallest_local_community=True)
945 )
947 assert res.current.community.community_id == city_id
948 assert res.is_in_smallest_local_community
949 assert not res.current.too_large_for_notifications
950 assert res.current.approx_users_to_notify == 2
952 with session_scope() as session:
953 assert session.execute(select(Event.parent_node_id)).scalar_one() == city_id
956def test_MoveEventToCommunity_explicit_community(event_in_wrong_community):
957 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
959 with real_editor_session(editor_token) as api:
960 res = api.MoveEventToCommunity(editor_pb2.MoveEventToCommunityReq(event_id=event_id, community_id=city_id))
961 assert res.current.community.community_id == city_id
963 # and back again, editors aren't restricted to the smallest community
964 res = api.MoveEventToCommunity(editor_pb2.MoveEventToCommunityReq(event_id=event_id, community_id=country_id))
965 assert res.current.community.community_id == country_id
966 assert not res.is_in_smallest_local_community
968 with session_scope() as session:
969 assert session.execute(select(Event.parent_node_id)).scalar_one() == country_id
972def test_MoveEventToCommunity_moves_all_occurrences(event_in_wrong_community):
973 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
975 # the fixture's occurrence is tomorrow at noon, so schedule this one after it
976 other_occurrence_id = events_api.schedule_event(
977 token=creator_token,
978 event_id=event_id,
979 location=events_pb2.EventLocation(address="Somewhere", lat=3, lng=1),
980 start_date_time=local_date_time(now() + timedelta(days=2)),
981 end_date_time=local_date_time(now() + timedelta(days=2, hours=1)),
982 ).event_id
984 with real_editor_session(editor_token) as api:
985 res = api.MoveEventToCommunity(
986 editor_pb2.MoveEventToCommunityReq(event_id=event_id, smallest_local_community=True)
987 )
988 assert res.occurrence_count == 2
990 res = api.GetEventCommunityInfo(editor_pb2.GetEventCommunityInfoReq(event_id=other_occurrence_id))
991 assert res.current.community.community_id == city_id
994def test_MoveEventToCommunity_no_target(event_in_wrong_community):
995 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
997 with real_editor_session(editor_token) as api:
998 with pytest.raises(grpc.RpcError) as e:
999 api.MoveEventToCommunity(editor_pb2.MoveEventToCommunityReq(event_id=event_id))
1000 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1001 assert (
1002 e.value.details()
1003 == "Specify either a community to move the event to, or recompute it from the event's location."
1004 )
1007def test_MoveEventToCommunity_community_not_found(event_in_wrong_community):
1008 editor_token, creator_token, event_id, country_id, city_id = event_in_wrong_community
1010 with real_editor_session(editor_token) as api:
1011 with pytest.raises(grpc.RpcError) as e:
1012 api.MoveEventToCommunity(editor_pb2.MoveEventToCommunityReq(event_id=event_id, community_id=42))
1013 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1014 assert e.value.details() == "Community not found."