from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import patch from fastapi import HTTPException from pydantic import ValidationError from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.modules import ModuleContext from govoplan_core.core.poll import PollCapabilityError from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.base import Base from govoplan_poll.backend.db.models import ( Poll, PollInvitation, PollOption, PollParticipationSubmission, PollResponse, ) from govoplan_poll.backend.manifest import get_manifest as get_poll_manifest from govoplan_scheduling.backend import service as scheduling_service from govoplan_scheduling.backend.db.models import ( SchedulingCandidateSlot, SchedulingNotification, SchedulingParticipant, SchedulingRequest, ) from govoplan_scheduling.backend.manifest import ADMIN_SCOPE, RESPOND_SCOPE, WRITE_SCOPE from govoplan_scheduling.backend.router import ( api_get_my_scheduling_availability, api_submit_scheduling_availability, api_update_scheduling_candidate_slot, api_update_scheduling_request, ) from govoplan_scheduling.backend.runtime import configure_runtime from govoplan_scheduling.backend.schemas import ( SchedulingAvailabilityAnswerInput, SchedulingAvailabilityResponseRequest, SchedulingCalendarPreferences, SchedulingCandidateSlotInput, SchedulingCandidateSlotReconcileInput, SchedulingCandidateSlotUpdateRequest, SchedulingParticipantInput, SchedulingParticipantReconcileInput, SchedulingPublicParticipationAccessRequest, SchedulingPublicParticipationSubmitRequest, SchedulingRequestCreateRequest, SchedulingRequestUpdateRequest, ) from govoplan_scheduling.backend.security import verify_participant_password from govoplan_scheduling.backend.service import ( SchedulingError, SchedulingPublicParticipationError, cancel_scheduling_request, create_scheduling_request, get_public_scheduling_participation, scheduling_request_summary, scheduling_slot_revision, submit_scheduling_availability, submit_public_scheduling_participation, update_scheduling_candidate_slot, ) class SchedulingResponseEditingTests(unittest.TestCase): def setUp(self) -> None: registry = PlatformRegistry() registry.register(get_poll_manifest()) registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) configure_runtime(registry=registry) self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__, PollParticipationSubmission.__table__, SchedulingRequest.__table__, SchedulingCandidateSlot.__table__, SchedulingParticipant.__table__, SchedulingNotification.__table__, ], ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all( self.engine, tables=[ SchedulingNotification.__table__, SchedulingParticipant.__table__, SchedulingCandidateSlot.__table__, SchedulingRequest.__table__, PollParticipationSubmission.__table__, PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__, ], ) self.engine.dispose() @staticmethod def _principal(account_id: str, *, email: str | None, scopes: set[str]) -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=f"membership-{account_id}", tenant_id="tenant-1", email=email, display_name=account_id, scopes=frozenset(scopes), ), account=SimpleNamespace(id=account_id), user=SimpleNamespace(id=f"membership-{account_id}"), ) def _request( self, *, status: str = "collecting", allow_participant_updates: bool = True, participants: list[SchedulingParticipantInput] | None = None, **settings, ) -> SchedulingRequest: request, _tokens = self._request_and_tokens( status=status, allow_participant_updates=allow_participant_updates, participants=participants, **settings, ) return request def _request_and_tokens( self, *, status: str = "collecting", allow_participant_updates: bool = True, participants: list[SchedulingParticipantInput] | None = None, **settings, ) -> tuple[SchedulingRequest, dict[str, str]]: start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc) return create_scheduling_request( self.session, tenant_id="tenant-1", user_id="organizer-1", payload=SchedulingRequestCreateRequest( title="Response editing", status=status, allow_participant_updates=allow_participant_updates, calendar=SchedulingCalendarPreferences(), slots=[ SchedulingCandidateSlotInput( label="Monday", start_at=start, end_at=start + timedelta(hours=1), ), SchedulingCandidateSlotInput( label="Tuesday", start_at=start + timedelta(days=1), end_at=start + timedelta(days=1, hours=1), ), ], participants=( participants if participants is not None else [ SchedulingParticipantInput( display_name="Alice", email="alice@example.test", ) ] ), **settings, ), ) def _answer( self, request: SchedulingRequest, *, account_id: str, email: str, values: tuple[str, ...], comment: str | None = None, ) -> None: api_submit_scheduling_availability( request.id, SchedulingAvailabilityResponseRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=slot.id, value=value, option_revision=scheduling_slot_revision(slot), ) for slot, value in zip(request.slots, values, strict=True) ], comment=comment, ), session=self.session, principal=self._principal(account_id, email=email, scopes={RESPOND_SCOPE}), ) def test_response_settings_are_persisted_and_password_is_write_only(self) -> None: request = self._request( notify_on_answers=False, single_choice=True, max_participants_per_option=3, allow_maybe=False, allow_comments=True, participant_email_required=True, anonymous_password_protection_enabled=True, anonymous_password="correct horse battery staple", create_participant_invitations=False, ) self.assertFalse(request.notify_on_answers) self.assertTrue(request.single_choice) self.assertEqual(request.max_participants_per_option, 3) self.assertFalse(request.allow_maybe) self.assertTrue(request.allow_comments) self.assertTrue(request.participant_email_required) self.assertTrue(request.anonymous_password_protection_enabled) self.assertNotIn("correct horse", request.anonymous_password_hash or "") self.assertTrue( verify_participant_password( "correct horse battery staple", request.anonymous_password_hash, ) ) response = api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest(title="Response settings, updated"), session=self.session, principal=self._principal("organizer-1", email=None, scopes={WRITE_SCOPE}), ) serialized = response.model_dump() self.assertTrue(serialized["anonymous_password_protection_enabled"]) self.assertTrue(serialized["public_participation_policy_enforcement_available"]) self.assertIsNone(serialized["public_participation_policy_enforcement_reason"]) self.assertNotIn("anonymous_password", serialized) self.assertNotIn("anonymous_password_hash", serialized) api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest(anonymous_password_protection_enabled=False), session=self.session, principal=self._principal("admin-1", email=None, scopes={ADMIN_SCOPE}), ) self.assertFalse(request.anonymous_password_protection_enabled) self.assertIsNone(request.anonymous_password_hash) def test_only_organizer_or_scheduling_admin_can_edit(self) -> None: request = self._request() with self.assertRaises(HTTPException) as raised: api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest(title="Unauthorized change"), session=self.session, principal=self._principal("other-writer", email=None, scopes={WRITE_SCOPE}), ) self.assertEqual(raised.exception.status_code, 403) self.assertEqual(request.title, "Response editing") def test_request_edit_preserves_answers_when_slots_are_unchanged(self) -> None: request = self._request() self._submit_both(request) response = api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest( description=None, location=None, notify_on_answers=False, ), 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=self._principal( "alice-account", email="alice@example.test", scopes={RESPOND_SCOPE}, ), ) self.assertIsNone(response.description) self.assertIsNone(response.location) self.assertFalse(response.notify_on_answers) self.assertEqual(len(current.answers), 2) def test_single_choice_maybe_and_comment_settings_are_enforced(self) -> None: request = self._request( single_choice=True, allow_maybe=False, create_participant_invitations=False, ) with self.assertRaises(HTTPException) as multiple: self._answer( request, account_id="alice-account", email="alice@example.test", values=("available", "available"), ) self.assertEqual(multiple.exception.status_code, 400) with self.assertRaises(HTTPException) as maybe: self._answer( request, account_id="alice-account", email="alice@example.test", values=("maybe", "unavailable"), ) self.assertEqual(maybe.exception.status_code, 400) with self.assertRaises(HTTPException) as comment: self._answer( request, account_id="alice-account", email="alice@example.test", values=("available", "unavailable"), comment="I prefer Monday", ) self.assertEqual(comment.exception.status_code, 400) def test_comments_round_trip_and_notifications_follow_setting(self) -> None: request = self._request( allow_comments=True, notify_on_answers=False, create_participant_invitations=False, ) with patch("govoplan_scheduling.backend.service._emit_scheduling_center_notification") as emit: self._answer( request, account_id="alice-account", email="alice@example.test", values=("available", "unavailable"), comment=" Monday works best. ", ) emit.assert_not_called() response = api_get_my_scheduling_availability( request.id, session=self.session, principal=self._principal( "alice-account", email="alice@example.test", scopes={RESPOND_SCOPE}, ), ) self.assertEqual(response.comment, "Monday works best.") poll_response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one() self.assertEqual(poll_response.metadata_["comment"], "Monday works best.") def test_capacity_is_enforced_but_own_response_can_be_edited(self) -> None: request = self._request( max_participants_per_option=1, create_participant_invitations=False, participants=[ SchedulingParticipantInput(display_name="Alice", email="alice@example.test"), SchedulingParticipantInput(display_name="Bob", email="bob@example.test"), ], ) self._answer( request, account_id="alice-account", email="alice@example.test", values=("available", "unavailable"), ) with self.assertRaises(HTTPException) as full: self._answer( request, account_id="bob-account", email="bob@example.test", values=("available", "unavailable"), ) self.assertEqual(full.exception.status_code, 409) self._answer( request, account_id="alice-account", email="alice@example.test", values=("unavailable", "available"), ) self._answer( request, account_id="bob-account", email="bob@example.test", values=("available", "unavailable"), ) summary = scheduling_request_summary( self.session, tenant_id="tenant-1", request_id=request.id, ) values = { item["option_id"]: item["values"] for item in summary["option_results"] } self.assertEqual(values[request.slots[0].poll_option_id]["available"], 1) self.assertEqual(values[request.slots[1].poll_option_id]["available"], 1) def test_restricted_signed_participation_uses_governed_gateway(self) -> None: request = self._request( participant_email_required=True, create_participant_invitations=True, ) self.assertEqual(request.participants[0].participation_gateway, "scheduling") with self.assertRaises(HTTPException) as update: api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest(allow_comments=True), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) self.assertEqual(update.exception.status_code, 400) self.assertFalse(request.allow_comments) def test_public_governed_response_is_prefilled_and_idempotent(self) -> None: start = datetime(2026, 8, 1, 9, tzinfo=timezone.utc) public_request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="organizer-1", payload=SchedulingRequestCreateRequest( title="Public governed response", status="collecting", single_choice=True, allow_maybe=False, allow_comments=True, participant_email_required=True, anonymous_password_protection_enabled=True, anonymous_password="correct horse battery staple", calendar=SchedulingCalendarPreferences(), slots=[ SchedulingCandidateSlotInput( label="First", start_at=start, end_at=start + timedelta(hours=1), ), SchedulingCandidateSlotInput( label="Second", start_at=start + timedelta(days=1), end_at=start + timedelta(days=1, hours=1), ), ], participants=[ SchedulingParticipantInput( display_name="Alice", email="alice@example.test", ) ], ), ) token = tokens[public_request.participants[0].id] invitation = self.session.query(PollInvitation).filter( PollInvitation.id == public_request.participants[0].poll_invitation_id ).one() self.assertIsNotNone(invitation.token_hash) self.assertNotEqual(invitation.token_hash, token) access = SchedulingPublicParticipationAccessRequest( participant_email="ALICE@EXAMPLE.TEST", password="correct horse battery staple", ) initial = get_public_scheduling_participation( self.session, request_id=public_request.id, token=token, payload=access, client_address="192.0.2.10", ) self.assertFalse(initial["has_response"]) self.assertNotIn("anonymous_password_hash", initial) submit_payload = SchedulingPublicParticipationSubmitRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=public_request.slots[0].id, value="available", option_revision=scheduling_slot_revision(public_request.slots[0]), ), SchedulingAvailabilityAnswerInput( slot_id=public_request.slots[1].id, value="unavailable", option_revision=scheduling_slot_revision(public_request.slots[1]), ), ], participant_email="alice@example.test", password="correct horse battery staple", comment=" First works. ", idempotency_key="public-response-1", ) submitted = submit_public_scheduling_participation( self.session, request_id=public_request.id, token=token, payload=submit_payload, client_address="192.0.2.10", ) replayed = submit_public_scheduling_participation( self.session, request_id=public_request.id, token=token, payload=submit_payload, client_address="192.0.2.10", ) prefilled = get_public_scheduling_participation( self.session, request_id=public_request.id, token=token, payload=access, client_address="192.0.2.10", ) self.assertTrue(submitted["has_response"]) self.assertFalse(submitted["replayed"]) self.assertTrue(replayed["replayed"]) self.assertEqual(prefilled["comment"], "First works.") self.assertEqual(len(prefilled["answers"]), 2) self.assertEqual( self.session.query(PollResponse).filter( PollResponse.poll_id == public_request.poll_id ).count(), 1, ) def test_public_password_failures_are_generic_and_throttled_before_verify(self) -> None: start = datetime(2026, 8, 10, 9, tzinfo=timezone.utc) request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="organizer-1", payload=SchedulingRequestCreateRequest( title="Protected response", status="collecting", anonymous_password_protection_enabled=True, anonymous_password="correct horse battery staple", calendar=SchedulingCalendarPreferences(), slots=[ SchedulingCandidateSlotInput( start_at=start, end_at=start + timedelta(hours=1), ) ], participants=[SchedulingParticipantInput(display_name="Guest")], ), ) token = tokens[request.participants[0].id] wrong = SchedulingPublicParticipationAccessRequest(password="wrong password") for attempt in range(10): with self.assertRaises(SchedulingPublicParticipationError) as raised: get_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=wrong, client_address="192.0.2.20", ) self.assertEqual( str(raised.exception), "Scheduling participation link or credentials are invalid", ) if attempt < 9: self.assertEqual(raised.exception.retry_after_seconds, 0) self.assertGreater(raised.exception.retry_after_seconds, 0) with patch( "govoplan_scheduling.backend.service.verify_participant_password" ) as verify: with self.assertRaises(SchedulingPublicParticipationError) as blocked: get_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=wrong, client_address="192.0.2.20", ) verify.assert_not_called() self.assertGreater(blocked.exception.retry_after_seconds, 0) def test_public_email_is_bound_after_first_anonymous_response(self) -> None: start = datetime(2026, 8, 12, 9, tzinfo=timezone.utc) request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="organizer-1", payload=SchedulingRequestCreateRequest( title="Email-bound response", status="collecting", participant_email_required=True, calendar=SchedulingCalendarPreferences(), slots=[ SchedulingCandidateSlotInput( start_at=start, end_at=start + timedelta(hours=1), ) ], participants=[SchedulingParticipantInput(display_name="Guest")], ), ) participant = request.participants[0] token = tokens[participant.id] answer = SchedulingAvailabilityAnswerInput( slot_id=request.slots[0].id, value="available", option_revision=scheduling_slot_revision(request.slots[0]), ) submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationSubmitRequest( answers=[answer], participant_email="first@example.test", ), client_address="192.0.2.30", ) with self.assertRaises(SchedulingPublicParticipationError): submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationSubmitRequest( answers=[answer], participant_email="second@example.test", ), client_address="192.0.2.30", ) self.assertEqual(participant.email, "first@example.test") self.assertTrue( participant.respondent_id.startswith("scheduling-participant:") ) self.assertEqual( self.session.query(PollResponse).filter( PollResponse.poll_id == request.poll_id ).count(), 1, ) def test_public_submission_locks_scheduling_before_poll_submission(self) -> None: request, tokens = self._request_and_tokens() token = tokens[request.participants[0].id] provider = scheduling_service._poll_participation_provider() self.assertIsNotNone(provider) events: list[str] = [] original_lock = scheduling_service._lock_scheduling_request original_submit = provider.submit_governed_response def recording_lock(*args, **kwargs): self.assertTrue(kwargs["lock_participants"]) events.append("scheduling") return original_lock(*args, **kwargs) def recording_submit(*args, **kwargs): self.assertEqual(events, ["scheduling"]) events.append("poll") return original_submit(*args, **kwargs) with ( patch.object( scheduling_service, "_lock_scheduling_request", side_effect=recording_lock, ), patch.object( provider, "submit_governed_response", side_effect=recording_submit, ), ): submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationSubmitRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=request.slots[0].id, value="available", option_revision=scheduling_slot_revision( request.slots[0] ), ) ], participant_email="alice@example.test", ), client_address="192.0.2.40", ) self.assertEqual(events, ["scheduling", "poll"]) def test_scheduling_status_and_deadline_guard_all_governed_submissions(self) -> None: request, tokens = self._request_and_tokens() token = tokens[request.participants[0].id] answer = SchedulingAvailabilityAnswerInput( slot_id=request.slots[0].id, value="available", option_revision=scheduling_slot_revision(request.slots[0]), ) authenticated_payload = SchedulingAvailabilityResponseRequest( answers=[answer] ) public_payload = SchedulingPublicParticipationSubmitRequest( answers=[answer], participant_email="alice@example.test", ) # Simulate a stale Poll projection that remains open after Scheduling # has already closed response collection. request.status = "closed" with self.assertRaisesRegex( SchedulingError, "not collecting availability", ): submit_scheduling_availability( self.session, tenant_id=request.tenant_id, request_id=request.id, actor_ids=("alice@example.test",), respondent_id="alice-account", respondent_label="Alice", payload=authenticated_payload, ) with self.assertRaises(SchedulingPublicParticipationError): submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=public_payload, client_address="192.0.2.42", ) # Likewise, invitation expiry drift must not override Scheduling's # authoritative deadline. request.status = "collecting" request.deadline_at = datetime.now(timezone.utc) - timedelta(minutes=1) invitation = ( self.session.query(PollInvitation) .filter(PollInvitation.id == request.participants[0].poll_invitation_id) .one() ) invitation.expires_at = None with self.assertRaisesRegex(SchedulingError, "deadline has passed"): submit_scheduling_availability( self.session, tenant_id=request.tenant_id, request_id=request.id, actor_ids=("alice@example.test",), respondent_id="alice-account", respondent_label="Alice", payload=authenticated_payload, ) with self.assertRaises(SchedulingPublicParticipationError): submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=public_payload, client_address="192.0.2.42", ) self.assertEqual( self.session.query(PollResponse) .filter(PollResponse.poll_id == request.poll_id) .count(), 0, ) def test_deadline_changes_update_same_link_and_preserve_prior_response(self) -> None: initial_deadline = datetime(2030, 8, 10, 12, tzinfo=timezone.utc) request, tokens = self._request_and_tokens( deadline_at=initial_deadline, allow_comments=True, ) participant = request.participants[0] token = tokens[participant.id] submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationSubmitRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=request.slots[0].id, value="available", option_revision=scheduling_slot_revision(request.slots[0]), ) ], participant_email="alice@example.test", comment="Keep this response.", ), client_address="192.0.2.41", ) self.session.commit() invitation_id = participant.poll_invitation_id invitation = ( self.session.query(PollInvitation) .filter(PollInvitation.id == invitation_id) .one() ) token_hash = invitation.token_hash responded_at = participant.responded_at deadlines = ( datetime(2030, 8, 20, 12, tzinfo=timezone.utc), datetime.now(timezone.utc) - timedelta(minutes=1), datetime(2030, 8, 15, 12, tzinfo=timezone.utc), None, ) for deadline in deadlines: response = api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest(deadline_at=deadline), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) projected = response.participants[0] self.assertEqual(projected.status, "responded") self.assertIsNone(projected.invitation_token) self.assertEqual(projected.poll_invitation_id, invitation_id) participant = ( self.session.query(SchedulingParticipant) .filter(SchedulingParticipant.id == participant.id) .populate_existing() .one() ) invitation = ( self.session.query(PollInvitation) .filter(PollInvitation.id == invitation_id) .populate_existing() .one() ) self.assertEqual(participant.poll_invitation_id, invitation_id) self.assertEqual(participant.responded_at, responded_at) self.assertEqual(participant.status, "responded") self.assertEqual( scheduling_service.response_datetime(invitation.expires_at), deadline, ) self.assertEqual(invitation.token_hash, token_hash) self.assertIsNone(invitation.revoked_at) if deadline is not None and deadline <= datetime.now(timezone.utc): with self.assertRaises(SchedulingPublicParticipationError): get_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationAccessRequest( participant_email="alice@example.test" ), client_address="192.0.2.41", ) continue prefilled = get_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationAccessRequest( participant_email="alice@example.test" ), client_address="192.0.2.41", ) self.assertTrue(prefilled["has_response"]) self.assertEqual(prefilled["comment"], "Keep this response.") self.assertEqual( self.session.query(PollResponse) .filter(PollResponse.poll_id == request.poll_id) .count(), 1, ) def test_deadline_expiry_update_rolls_back_when_poll_rejects_it(self) -> None: original_deadline = datetime(2030, 9, 10, 12, tzinfo=timezone.utc) request, tokens = self._request_and_tokens(deadline_at=original_deadline) participant = request.participants[0] token = tokens[participant.id] invitation_id = participant.poll_invitation_id request_id = request.id participant_id = participant.id self.session.commit() provider = scheduling_service._poll_participation_provider() self.assertIsNotNone(provider) with patch.object( provider, "update_invitation_expiry", side_effect=PollCapabilityError("Expiry update failed"), ): with self.assertRaises(HTTPException): api_update_scheduling_request( request_id, SchedulingRequestUpdateRequest( deadline_at=datetime( 2030, 9, 20, 12, tzinfo=timezone.utc, ) ), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) self.session.rollback() persisted_request = ( self.session.query(SchedulingRequest) .filter(SchedulingRequest.id == request_id) .populate_existing() .one() ) persisted_participant = ( self.session.query(SchedulingParticipant) .filter(SchedulingParticipant.id == participant_id) .populate_existing() .one() ) persisted_invitation = ( self.session.query(PollInvitation) .filter(PollInvitation.id == invitation_id) .populate_existing() .one() ) self.assertEqual( scheduling_service.response_datetime(persisted_request.deadline_at), original_deadline, ) self.assertEqual(persisted_participant.poll_invitation_id, invitation_id) self.assertIsNone(persisted_invitation.revoked_at) self.assertEqual( scheduling_service.response_datetime(persisted_invitation.expires_at), original_deadline, ) self.assertEqual( get_public_scheduling_participation( self.session, request_id=request_id, token=token, payload=SchedulingPublicParticipationAccessRequest( participant_email="alice@example.test" ), client_address="192.0.2.41", )["request_id"], request_id, ) def test_full_edit_reconciles_slots_participants_and_revokes_removed_link(self) -> None: request = self._request(create_participant_invitations=False) self._submit_both(request) first_slot, removed_slot = request.slots alice = request.participants[0] new_start = datetime(2026, 7, 23, 9, tzinfo=timezone.utc) response = api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest( slots=[ SchedulingCandidateSlotReconcileInput( id=first_slot.id, revision=scheduling_slot_revision(first_slot), label=first_slot.label, description=first_slot.description, start_at=first_slot.start_at.replace(tzinfo=timezone.utc), end_at=first_slot.end_at.replace(tzinfo=timezone.utc), timezone=first_slot.timezone, location=first_slot.location, metadata=first_slot.metadata_ or {}, ), SchedulingCandidateSlotReconcileInput( label="Wednesday", start_at=new_start, end_at=new_start + timedelta(hours=1), ), ], participants=[ SchedulingParticipantReconcileInput( id=alice.id, respondent_id=alice.respondent_id, display_name=alice.display_name, email=alice.email, participant_type=alice.participant_type, required=alice.required, metadata=alice.metadata_ or {}, ), SchedulingParticipantReconcileInput( display_name="Bob", email="BOB@EXAMPLE.TEST", ), ], create_participant_invitations=True, ), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) self.assertEqual([slot.label for slot in response.slots], ["Monday", "Wednesday"]) self.assertIsNotNone(removed_slot.deleted_at) bob = next(item for item in response.participants if item.display_name == "Bob") self.assertEqual(bob.email, "bob@example.test") self.assertIsNotNone(bob.poll_invitation_id) self.assertIsNotNone(bob.invitation_token) current = api_get_my_scheduling_availability( request.id, session=self.session, principal=self._principal( "alice-account", email="ALICE@EXAMPLE.TEST", scopes={RESPOND_SCOPE}, ), ) self.assertEqual( [(answer.slot_id, answer.value) for answer in current.answers], [(first_slot.id, "available")], ) retained_slots = [slot for slot in request.slots if slot.deleted_at is None] api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest( slots=[ SchedulingCandidateSlotReconcileInput( id=slot.id, revision=scheduling_slot_revision(slot), label=slot.label, description=slot.description, start_at=slot.start_at.replace(tzinfo=timezone.utc), end_at=slot.end_at.replace(tzinfo=timezone.utc), timezone=slot.timezone, location=slot.location, metadata=slot.metadata_ or {}, ) for slot in retained_slots ], participants=[ SchedulingParticipantReconcileInput( id=alice.id, respondent_id=alice.respondent_id, display_name=alice.display_name, email=alice.email, participant_type=alice.participant_type, required=alice.required, metadata=alice.metadata_ or {}, ) ], create_participant_invitations=False, ), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) invitation = self.session.query(PollInvitation).filter( PollInvitation.id == bob.poll_invitation_id ).one() self.assertIsNotNone(invitation.revoked_at) def test_full_edit_rejects_stale_slot_revision(self) -> None: request = self._request(create_participant_invitations=False) slot = request.slots[0] with self.assertRaises(HTTPException) as raised: api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest( slots=[ SchedulingCandidateSlotReconcileInput( id=current.id, revision=("0" * 64 if current.id == slot.id else scheduling_slot_revision(current)), label=current.label, description=current.description, start_at=current.start_at.replace(tzinfo=timezone.utc), end_at=current.end_at.replace(tzinfo=timezone.utc), timezone=current.timezone, location=current.location, metadata=current.metadata_ or {}, ) for current in request.slots ] ), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) self.assertEqual(raised.exception.status_code, 409) def test_draft_edit_does_not_issue_link_for_added_participant(self) -> None: request = self._request( status="draft", participants=[], create_participant_invitations=False, ) response = api_update_scheduling_request( request.id, SchedulingRequestUpdateRequest( participants=[ SchedulingParticipantReconcileInput( display_name="Draft participant", email="draft@example.test", ) ], create_participant_invitations=True, ), session=self.session, principal=self._principal( "organizer-1", email=None, scopes={WRITE_SCOPE}, ), ) self.assertEqual(len(response.participants), 1) self.assertEqual(response.participants[0].status, "draft") self.assertIsNone(response.participants[0].poll_invitation_id) self.assertIsNone(response.participants[0].invitation_token) def _submit_both(self, request: SchedulingRequest) -> None: 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]), ), SchedulingAvailabilityAnswerInput( slot_id=request.slots[1].id, value="maybe", option_revision=scheduling_slot_revision(request.slots[1]), ), ] ), session=self.session, principal=self._principal( "alice-account", email="alice@example.test", scopes={RESPOND_SCOPE}, ), ) def test_current_participant_response_is_prefilled_and_changed_slot_is_invalidated(self) -> None: request = self._request() self._submit_both(request) alice = self._principal( "alice-account", email="alice@example.test", scopes={RESPOND_SCOPE}, ) before = api_get_my_scheduling_availability( request.id, session=self.session, principal=alice, ) self.assertTrue(before.has_response) self.assertIsNotNone(before.submitted_at) self.assertEqual(before.participant_id, request.participants[0].id) self.assertEqual( [(answer.slot_id, answer.value) for answer in before.answers], [(request.slots[0].id, "available"), (request.slots[1].id, "maybe")], ) 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}), ) after = api_get_my_scheduling_availability( request.id, session=self.session, principal=alice, ) self.assertTrue(after.has_response) self.assertEqual( [(answer.slot_id, answer.value) for answer in after.answers], [(request.slots[1].id, "maybe")], ) poll_response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one() self.assertEqual( [answer["option_id"] for answer in poll_response.answers], [request.slots[1].poll_option_id], ) # Re-submit and repeat the exact option snapshot: an idempotent no-op # must preserve both current answers. self._submit_both(request) 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}), ) unchanged = api_get_my_scheduling_availability( request.id, session=self.session, principal=alice, ) self.assertEqual(len(unchanged.answers), 2) def test_calendar_hold_rejection_leaves_slot_and_answers_unchanged(self) -> None: request = self._request() self._submit_both(request) slot = request.slots[0] original_start = slot.start_at slot.tentative_hold_event_id = "event-1" self.session.flush() with self.assertRaisesRegex(SchedulingError, "tentative calendar hold"): update_scheduling_candidate_slot( self.session, tenant_id="tenant-1", request_id=request.id, slot_id=slot.id, payload=SchedulingCandidateSlotUpdateRequest( start_at=original_start.replace(tzinfo=timezone.utc) + timedelta(minutes=15), ), ) self.assertEqual(slot.start_at, original_start) response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one() self.assertEqual(len(response.answers), 2) def test_cancelling_draft_request_preserves_draft_poll(self) -> None: request = self._request(status="draft") cancelled = cancel_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ) poll = self.session.query(Poll).filter(Poll.id == request.poll_id).one() self.assertEqual(cancelled.status, "cancelled") 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()