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

fix(responses): reject stale scheduling answers

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-20 18:35:57 +02:00
parent 0f33e25c83
commit bbf503a7d7
5 changed files with 311 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
tests/test_response_editing.py
View File

@@ -4,6 +4,8 @@ import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
@@ -40,6 +42,8 @@ from govoplan_scheduling.backend.service import (
SchedulingError,
cancel_scheduling_request,
create_scheduling_request,
scheduling_request_summary,
scheduling_slot_revision,
update_scheduling_candidate_slot,
)
@@ -99,7 +103,12 @@ class SchedulingResponseEditingTests(unittest.TestCase):
user=SimpleNamespace(id=f"membership-{account_id}"),
)
def _request(self, *, status: str = "collecting") -> SchedulingRequest:
def _request(
self,
*,
status: str = "collecting",
allow_participant_updates: bool = True,
) -> SchedulingRequest:
start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc)
request, _tokens = create_scheduling_request(
self.session,
@@ -108,6 +117,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
payload=SchedulingRequestCreateRequest(
title="Response editing",
status=status,
allow_participant_updates=allow_participant_updates,
calendar=SchedulingCalendarPreferences(),
slots=[
SchedulingCandidateSlotInput(
@@ -136,8 +146,16 @@ class SchedulingResponseEditingTests(unittest.TestCase):
request.id,
SchedulingAvailabilityResponseRequest(
answers=[
SchedulingAvailabilityAnswerInput(slot_id=request.slots[0].id, value="available"),
SchedulingAvailabilityAnswerInput(slot_id=request.slots[1].id, value="maybe"),
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="available",
option_revision=scheduling_slot_revision(request.slots[0]),
),
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[1].id,
value="maybe",
option_revision=scheduling_slot_revision(request.slots[1]),
),
]
),
session=self.session,
@@ -227,7 +245,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
request_id=request.id,
slot_id=slot.id,
payload=SchedulingCandidateSlotUpdateRequest(
start_at=original_start + timedelta(minutes=15),
start_at=original_start.replace(tzinfo=timezone.utc) + timedelta(minutes=15),
),
)
@@ -249,6 +267,115 @@ class SchedulingResponseEditingTests(unittest.TestCase):
self.assertIsNotNone(cancelled.cancelled_at)
self.assertEqual(poll.status, "draft")
def test_fully_invalidated_response_becomes_unanswered(self) -> None:
request = self._request()
alice = self._principal(
"alice-account",
email="alice@example.test",
scopes={RESPOND_SCOPE},
)
api_submit_scheduling_availability(
request.id,
SchedulingAvailabilityResponseRequest(
answers=[
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="available",
option_revision=scheduling_slot_revision(request.slots[0]),
)
]
),
session=self.session,
principal=alice,
)
api_update_scheduling_candidate_slot(
request.id,
request.slots[0].id,
SchedulingCandidateSlotUpdateRequest(label="Monday, revised"),
session=self.session,
principal=self._principal("organizer-1", email=None, scopes={WRITE_SCOPE}),
)
current = api_get_my_scheduling_availability(
request.id,
session=self.session,
principal=alice,
)
self.assertFalse(current.has_response)
self.assertEqual(current.answers, [])
self.assertEqual(request.participants[0].status, "invited")
self.assertIsNone(request.participants[0].responded_at)
self.assertEqual(
scheduling_request_summary(self.session, tenant_id="tenant-1", request_id=request.id)["response_count"],
0,
)
def test_option_change_is_blocked_when_response_updates_are_disabled(self) -> None:
request = self._request(allow_participant_updates=False)
self._submit_both(request)
original_label = request.slots[0].label
with self.assertRaisesRegex(SchedulingError, "response updates are disabled"):
update_scheduling_candidate_slot(
self.session,
tenant_id="tenant-1",
request_id=request.id,
slot_id=request.slots[0].id,
payload=SchedulingCandidateSlotUpdateRequest(label="Monday, revised"),
)
self.assertEqual(request.slots[0].label, original_label)
response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one()
self.assertEqual(len(response.answers), 2)
def test_stale_option_revision_is_rejected_without_writing_a_response(self) -> None:
request = self._request()
stale_revision = scheduling_slot_revision(request.slots[0])
api_update_scheduling_candidate_slot(
request.id,
request.slots[0].id,
SchedulingCandidateSlotUpdateRequest(label="Monday, revised"),
session=self.session,
principal=self._principal("organizer-1", email=None, scopes={WRITE_SCOPE}),
)
with self.assertRaises(HTTPException) as stale:
api_submit_scheduling_availability(
request.id,
SchedulingAvailabilityResponseRequest(
answers=[
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="available",
option_revision=stale_revision,
)
]
),
session=self.session,
principal=self._principal(
"alice-account",
email="alice@example.test",
scopes={RESPOND_SCOPE},
),
)
self.assertEqual(stale.exception.status_code, 409)
self.assertEqual(self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).count(), 0)
def test_slot_inputs_require_aware_datetimes_and_known_timezones(self) -> None:
with self.assertRaises(ValidationError):
SchedulingCandidateSlotInput(
start_at=datetime(2026, 7, 20, 9),
end_at=datetime(2026, 7, 20, 10),
)
with self.assertRaisesRegex(ValidationError, "valid IANA timezone"):
SchedulingCandidateSlotInput(
start_at=datetime(2026, 7, 20, 9, tzinfo=timezone.utc),
end_at=datetime(2026, 7, 20, 10, tzinfo=timezone.utc),
timezone="Mars/Olympus_Mons",
)
if __name__ == "__main__":
unittest.main()

24
tests/test_service.py
View File

@@ -44,6 +44,7 @@ from govoplan_scheduling.backend.router import (
api_create_tentative_calendar_holds,
api_decide_scheduling_request,
api_evaluate_calendar_freebusy,
api_get_my_scheduling_availability,
api_get_scheduling_request,
api_list_scheduling_requests,
api_scheduling_summary,
@@ -66,6 +67,7 @@ from govoplan_scheduling.backend.service import (
open_scheduling_request,
require_visible_scheduling_results,
scheduling_request_summary,
scheduling_slot_revision,
update_scheduling_request,
)
from govoplan_scheduling.backend.runtime import configure_runtime
@@ -825,7 +827,12 @@ class SchedulingServiceTests(unittest.TestCase):
"attacker",
email="alice@example.test",
membership_id="alice-membership",
scopes={SCHEDULING_READ_SCOPE, "poll:poll:read", "poll:response:write"},
scopes={
SCHEDULING_READ_SCOPE,
SCHEDULING_RESPOND_SCOPE,
"poll:poll:read",
"poll:response:write",
},
)
response = api_submit_poll_response(
@@ -842,9 +849,19 @@ class SchedulingServiceTests(unittest.TestCase):
session=self.session,
principal=attacker,
)
current = api_get_my_scheduling_availability(
request.id,
session=self.session,
principal=attacker,
)
self.assertNotIn("invitation_id", response.metadata)
self.assertEqual([request.id], [item.id for item in listed.requests])
self.assertTrue(current.has_response)
self.assertEqual(
[(answer.slot_id, answer.value) for answer in current.answers],
[(request.slots[0].id, "available")],
)
alice = next(participant for participant in request.participants if participant.display_name == "Alice")
bob = next(participant for participant in request.participants if participant.display_name == "Bob")
self.assertEqual(alice.status, "responded")
@@ -883,10 +900,12 @@ class SchedulingServiceTests(unittest.TestCase):
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="available",
option_revision=scheduling_slot_revision(request.slots[0]),
),
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[1].id,
value="maybe",
option_revision=scheduling_slot_revision(request.slots[1]),
),
]
),
@@ -917,6 +936,7 @@ class SchedulingServiceTests(unittest.TestCase):
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="unavailable",
option_revision=scheduling_slot_revision(request.slots[0]),
)
]
),
@@ -969,10 +989,12 @@ class SchedulingServiceTests(unittest.TestCase):
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="unavailable",
option_revision=scheduling_slot_revision(request.slots[0]),
),
SchedulingAvailabilityAnswerInput(
slot_id=request.slots[1].id,
value="maybe",
option_revision=scheduling_slot_revision(request.slots[1]),
),
]
),