GovOPlaN/govoplan-scheduling
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(scheduling): reject stale invitation actions

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 04:16:59 +02:00
parent ed39f83688
commit f512784dd3
5 changed files with 225 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

175
tests/test_service.py
View File

@@ -43,6 +43,7 @@ from govoplan_scheduling.backend.schemas import (
SchedulingCandidateSlotInput,
SchedulingDecisionRequest,
SchedulingInvitationActionRequest,
SchedulingInvitationRevokeRequest,
SchedulingParticipantInput,
SchedulingPublicParticipationAccessRequest,
SchedulingPublicParticipationSubmitRequest,
@@ -82,6 +83,7 @@ from govoplan_scheduling.backend.service import (
open_scheduling_request,
require_visible_scheduling_results,
scheduling_request_summary,
scheduling_participant_revision,
scheduling_slot_revision,
submit_public_scheduling_participation,
update_scheduling_request,
@@ -222,6 +224,7 @@ class SchedulingServiceTests(unittest.TestCase):
tenant_id=request.tenant_id,
request_id=request.id,
participant_id=participant.id,
participant_revision=scheduling_participant_revision(participant),
action="copy",
)
self.assertIsNotNone(result.action_url)
@@ -1517,6 +1520,7 @@ class SchedulingServiceTests(unittest.TestCase):
tenant_id=request.tenant_id,
request_id=request.id,
participant_id=participant.id,
participant_revision=scheduling_participant_revision(participant),
action="send",
)
for participant in request.participants
@@ -1576,6 +1580,7 @@ class SchedulingServiceTests(unittest.TestCase):
tenant_id=request.tenant_id,
request_id=request.id,
participant_id=participant.id,
participant_revision=scheduling_participant_revision(participant),
action="send",
)
@@ -1625,6 +1630,7 @@ class SchedulingServiceTests(unittest.TestCase):
tenant_id=request.tenant_id,
request_id=request.id,
participant_id=participant.id,
participant_revision=scheduling_participant_revision(participant),
action="send",
)
@@ -1644,6 +1650,142 @@ class SchedulingServiceTests(unittest.TestCase):
request.id,
)
def test_stale_copy_is_rejected_before_rotating_invitation(self) -> None:
request, _automatic_tokens = create_scheduling_request(
self.session,
tenant_id="tenant-1",
user_id="user-1",
payload=self._payload().model_copy(
update={"calendar": SchedulingCalendarPreferences()}
),
)
participant = request.participants[0]
stale_revision = scheduling_participant_revision(participant)
self._issue_copy(request, participant)
invitation_id = participant.poll_invitation_id
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
with (
patch("govoplan_scheduling.backend.router.audit_event") as audit,
self.assertRaises(HTTPException) as conflict,
):
api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(
action="copy",
participant_revision=stale_revision,
),
Response(),
session=self.session,
principal=organizer,
)
self.assertEqual(conflict.exception.status_code, 409)
self.assertEqual(participant.poll_invitation_id, invitation_id)
invitation = self.session.query(PollInvitation).filter(
PollInvitation.id == invitation_id
).one()
self.assertIsNone(invitation.revoked_at)
audit.assert_not_called()
def test_stale_send_is_rejected_before_delivery_or_rotation(self) -> None:
class CapturingNotificationProvider:
def __init__(self) -> None:
self.requests = []
def enqueue_notification(self, _session, request, *, enqueue_delivery):
self.requests.append(request)
return {"id": "notification-1", "status": "queued"}
provider = CapturingNotificationProvider()
request, _automatic_tokens = create_scheduling_request(
self.session,
tenant_id="tenant-1",
user_id="user-1",
payload=self._payload().model_copy(
update={"calendar": SchedulingCalendarPreferences()}
),
)
participant = request.participants[0]
stale_revision = scheduling_participant_revision(participant)
self._issue_copy(request, participant)
invitation_id = participant.poll_invitation_id
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
with (
patch(
"govoplan_scheduling.backend.service.notification_dispatch_provider",
return_value=provider,
),
patch("govoplan_scheduling.backend.router.audit_event") as audit,
self.assertRaises(HTTPException) as conflict,
):
api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(
action="send",
participant_revision=stale_revision,
),
Response(),
session=self.session,
principal=organizer,
)
self.assertEqual(conflict.exception.status_code, 409)
self.assertEqual(participant.poll_invitation_id, invitation_id)
invitation = self.session.query(PollInvitation).filter(
PollInvitation.id == invitation_id
).one()
self.assertIsNone(invitation.revoked_at)
self.assertEqual(provider.requests, [])
self.assertEqual(
self.session.query(SchedulingNotification).filter(
SchedulingNotification.request_id == request.id
).count(),
0,
)
audit.assert_not_called()
def test_stale_revoke_is_rejected_before_changing_access(self) -> None:
request, _automatic_tokens = create_scheduling_request(
self.session,
tenant_id="tenant-1",
user_id="user-1",
payload=self._payload().model_copy(
update={"calendar": SchedulingCalendarPreferences()}
),
)
participant = request.participants[0]
stale_revision = scheduling_participant_revision(participant)
self._issue_copy(request, participant)
invitation_id = participant.poll_invitation_id
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
with (
patch("govoplan_scheduling.backend.router.audit_event") as audit,
self.assertRaises(HTTPException) as conflict,
):
api_revoke_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationRevokeRequest(
participant_revision=stale_revision,
),
Response(),
session=self.session,
principal=organizer,
)
self.assertEqual(conflict.exception.status_code, 409)
self.assertEqual(participant.poll_invitation_id, invitation_id)
invitation = self.session.query(PollInvitation).filter(
PollInvitation.id == invitation_id
).one()
self.assertIsNone(invitation.revoked_at)
audit.assert_not_called()
def test_invitation_router_rotates_revokes_and_enforces_management_policy(self) -> None:
request, automatic_tokens = create_scheduling_request(
self.session,
@@ -1663,7 +1805,10 @@ class SchedulingServiceTests(unittest.TestCase):
first = api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(action="copy"),
SchedulingInvitationActionRequest(
action="copy",
participant_revision=scheduling_participant_revision(participant),
),
response_headers,
session=self.session,
principal=organizer,
@@ -1688,7 +1833,10 @@ class SchedulingServiceTests(unittest.TestCase):
api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(action="copy"),
SchedulingInvitationActionRequest(
action="copy",
participant_revision=scheduling_participant_revision(participant),
),
Response(),
session=self.session,
principal=unrelated_writer,
@@ -1713,7 +1861,10 @@ class SchedulingServiceTests(unittest.TestCase):
rotated = api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(action="copy"),
SchedulingInvitationActionRequest(
action="copy",
participant_revision=scheduling_participant_revision(participant),
),
Response(),
session=self.session,
principal=administrator,
@@ -1737,6 +1888,9 @@ class SchedulingServiceTests(unittest.TestCase):
revoked = api_revoke_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationRevokeRequest(
participant_revision=scheduling_participant_revision(participant),
),
Response(),
session=self.session,
principal=organizer,
@@ -1744,6 +1898,9 @@ class SchedulingServiceTests(unittest.TestCase):
replayed = api_revoke_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationRevokeRequest(
participant_revision=scheduling_participant_revision(participant),
),
Response(),
session=self.session,
principal=organizer,
@@ -1793,7 +1950,12 @@ class SchedulingServiceTests(unittest.TestCase):
issued = api_issue_scheduling_participant_invitation(
request.id,
request.participants[0].id,
SchedulingInvitationActionRequest(action="copy"),
SchedulingInvitationActionRequest(
action="copy",
participant_revision=scheduling_participant_revision(
request.participants[0]
),
),
Response(),
session=self.session,
principal=organizer,
@@ -1838,7 +2000,10 @@ class SchedulingServiceTests(unittest.TestCase):
result = api_issue_scheduling_participant_invitation(
request.id,
participant.id,
SchedulingInvitationActionRequest(action="send"),
SchedulingInvitationActionRequest(
action="send",
participant_revision=scheduling_participant_revision(participant),
),
Response(),
session=self.session,
principal=organizer,