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, Response 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.calendar import CALENDAR_AVAILABILITY_READ_SCOPE, CALENDAR_EVENT_WRITE_SCOPE from govoplan_core.db.base import Base from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.core.modules import ModuleContext from govoplan_core.core.registry import PlatformRegistry from govoplan_access.backend.db.models import Account, User from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarOutboxOperation, CalendarSyncSource from govoplan_calendar.backend.manifest import get_manifest as get_calendar_manifest from govoplan_poll.backend.db.models import ( Poll, PollInvitation, PollLifecycleTransition, PollOption, PollParticipationSubmission, PollResponse, ) from govoplan_poll.backend.manifest import get_manifest as get_poll_manifest from govoplan_poll.backend.router import api_get_poll, api_list_polls, api_submit_poll_response from govoplan_poll.backend.schemas import PollAnswerInput, PollSubmitResponseRequest from govoplan_poll.backend.service import get_poll from govoplan_scheduling.backend.db.models import SchedulingCandidateSlot, SchedulingNotification, SchedulingParticipant, SchedulingRequest from govoplan_scheduling.backend.manifest import READ_SCOPE as SCHEDULING_READ_SCOPE from govoplan_scheduling.backend.manifest import ADMIN_SCOPE as SCHEDULING_ADMIN_SCOPE from govoplan_scheduling.backend.manifest import RESPOND_SCOPE as SCHEDULING_RESPOND_SCOPE from govoplan_scheduling.backend.manifest import WRITE_SCOPE as SCHEDULING_WRITE_SCOPE from govoplan_scheduling.backend.schemas import ( SchedulingAvailabilityAnswerInput, SchedulingAvailabilityResponseRequest, SchedulingCalendarPreferences, SchedulingCandidateSlotInput, SchedulingDecisionRequest, SchedulingInvitationActionRequest, SchedulingParticipantInput, SchedulingPublicParticipationAccessRequest, SchedulingPublicParticipationSubmitRequest, SchedulingRequestCreateRequest, SchedulingRequestUpdateRequest, ) from govoplan_scheduling.backend.router import ( api_create_final_calendar_event, api_create_scheduling_request, api_create_tentative_calendar_holds, api_decide_scheduling_request, api_evaluate_calendar_freebusy, api_get_my_scheduling_availability, api_get_scheduling_request, api_issue_scheduling_participant_invitation, api_list_scheduling_requests, api_scheduling_summary, api_submit_scheduling_availability, api_revoke_scheduling_participant_invitation, ) from govoplan_scheduling.backend.service import ( SchedulingError, cancel_scheduling_request, close_scheduling_request, create_final_calendar_event, create_scheduling_notification_jobs, create_scheduling_request, create_tentative_calendar_holds, decide_scheduling_request, evaluate_calendar_freebusy, get_visible_scheduling_request, get_public_scheduling_participation, issue_scheduling_participant_invitation, list_scheduling_notifications, list_visible_scheduling_notifications, list_visible_scheduling_requests, open_scheduling_request, require_visible_scheduling_results, scheduling_request_summary, scheduling_slot_revision, submit_public_scheduling_participation, update_scheduling_request, ) from govoplan_scheduling.backend.runtime import configure_runtime class SchedulingServiceTests(unittest.TestCase): def setUp(self) -> None: registry = PlatformRegistry() registry.register(get_poll_manifest()) registry.register(get_calendar_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__, PollLifecycleTransition.__table__, CalendarCollection.__table__, CalendarEvent.__table__, CalendarSyncSource.__table__, CalendarOutboxOperation.__table__, ChangeSequenceEntry.__table__, Account.__table__, User.__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=[ SchedulingParticipant.__table__, SchedulingNotification.__table__, SchedulingCandidateSlot.__table__, SchedulingRequest.__table__, User.__table__, Account.__table__, ChangeSequenceEntry.__table__, CalendarOutboxOperation.__table__, CalendarSyncSource.__table__, CalendarEvent.__table__, CalendarCollection.__table__, PollParticipationSubmission.__table__, PollInvitation.__table__, PollLifecycleTransition.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__, ], ) self.engine.dispose() def _calendar(self) -> CalendarCollection: calendar = CalendarCollection( id="calendar-1", tenant_id="tenant-1", slug="team", name="Team", timezone="Europe/Berlin", owner_type="tenant", visibility="tenant", is_default=True, metadata_={}, ) self.session.add(calendar) self.session.flush() return calendar @staticmethod def _principal( account_id: str, *, email: str | None = None, membership_id: str | None = None, scopes: set[str] | None = None, ) -> ApiPrincipal: membership_id = membership_id or f"membership-{account_id}" return ApiPrincipal( principal=PrincipalRef( account_id=account_id, membership_id=membership_id, tenant_id="tenant-1", email=email, display_name=account_id, scopes=frozenset(scopes or {SCHEDULING_READ_SCOPE}), ), account=SimpleNamespace(id=account_id), user=SimpleNamespace(id=membership_id), ) def _payload(self) -> SchedulingRequestCreateRequest: start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc) return SchedulingRequestCreateRequest( title="Steering group", description="Find a slot", location="Room 1", timezone="Europe/Berlin", status="collecting", deadline_at=start + timedelta(days=3), calendar=SchedulingCalendarPreferences( enabled=True, calendar_id="calendar-1", freebusy_enabled=True, tentative_holds_enabled=True, create_event_on_decision=True, ), slots=[ SchedulingCandidateSlotInput(start_at=start, end_at=start + timedelta(hours=1), label="Monday 09:00"), SchedulingCandidateSlotInput(start_at=start + timedelta(days=1), end_at=start + timedelta(days=1, hours=1), label="Tuesday 09:00"), ], participants=[ SchedulingParticipantInput(display_name="Alice", email="alice@example.test"), SchedulingParticipantInput(display_name="Bob", email="bob@example.test", required=False), ], ) def _issue_copy( self, request: SchedulingRequest, participant: SchedulingParticipant, ) -> str: result = issue_scheduling_participant_invitation( self.session, tenant_id=request.tenant_id, request_id=request.id, participant_id=participant.id, action="copy", ) self.assertIsNotNone(result.action_url) return str(result.action_url).rsplit("/", 1)[-1] def test_create_request_creates_poll_slots_without_implicit_invitations(self) -> None: request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) poll = get_poll(self.session, tenant_id="tenant-1", poll_id=request.poll_id) self.assertEqual(request.status, "collecting") self.assertEqual(request.calendar_id, "calendar-1") self.assertEqual(poll.kind, "availability") self.assertEqual(poll.status, "open") self.assertEqual(poll.visibility, "private") self.assertEqual(poll.context_module, "scheduling") self.assertEqual(poll.context_resource_id, request.id) self.assertEqual(len(request.slots), 2) self.assertTrue(all(slot.poll_option_id for slot in request.slots)) self.assertEqual(tokens, {}) self.assertTrue( all( participant.status == "draft" and participant.poll_invitation_id is None for participant in request.participants ) ) self.assertEqual( self.session.query(SchedulingNotification).filter( SchedulingNotification.request_id == request.id ).count(), 0, ) def test_draft_save_does_not_issue_or_deliver_public_invitations(self) -> None: class RejectingNotificationProvider: def enqueue_notification(self, *_args, **_kwargs): raise AssertionError("A draft must not enqueue invitation delivery") with patch( "govoplan_scheduling.backend.service.notification_dispatch_provider", return_value=RejectingNotificationProvider(), ): request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload().model_copy(update={"status": "draft"}), ) self.assertEqual(tokens, {}) self.assertEqual( self.session.query(PollInvitation).filter( PollInvitation.poll_id == request.poll_id ).count(), 0, ) self.assertTrue( all( participant.status == "draft" and participant.poll_invitation_id is None for participant in request.participants ) ) self.assertEqual( self.session.query(SchedulingNotification).filter( SchedulingNotification.request_id == request.id ).count(), 0, ) open_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ) 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=self._principal( "alice-account", email="alice@example.test", scopes={SCHEDULING_RESPOND_SCOPE}, ), ) alice = next( participant for participant in request.participants if participant.email == "alice@example.test" ) self.assertEqual(alice.status, "responded") self.assertEqual(alice.participation_gateway, "scheduling") self.assertIsNotNone(alice.poll_invitation_id) self.assertEqual( self.session.query(SchedulingNotification).filter( SchedulingNotification.request_id == request.id, SchedulingNotification.event_kind == "invitation", ).count(), 0, ) def test_create_route_persists_after_request_session_closes(self) -> None: response = api_create_scheduling_request( self._payload().model_copy(update={"calendar": SchedulingCalendarPreferences()}), session=self.session, principal=self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE}), ) self.session.close() self.session = self.Session() request = self.session.query(SchedulingRequest).filter(SchedulingRequest.id == response.id).one_or_none() poll = self.session.query(Poll).filter(Poll.id == response.poll_id).one_or_none() self.assertIsNotNone(request) self.assertIsNotNone(poll) self.assertEqual(request.title, "Steering group") self.assertEqual(poll.visibility, "private") def test_signed_response_summary_and_decision_handoff(self) -> None: payload = self._payload().model_copy(update={"calendar": SchedulingCalendarPreferences()}) request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) first_participant = request.participants[0] first_slot = request.slots[0] second_slot = request.slots[1] self.assertEqual(tokens, {}) token = self._issue_copy(request, first_participant) submit_public_scheduling_participation( self.session, request_id=request.id, token=token, payload=SchedulingPublicParticipationSubmitRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=first_slot.id, value="available", option_revision=scheduling_slot_revision(first_slot), ), SchedulingAvailabilityAnswerInput( slot_id=second_slot.id, value="maybe", option_revision=scheduling_slot_revision(second_slot), ), ] ), client_address="127.0.0.1", ) summary = scheduling_request_summary(self.session, tenant_id="tenant-1", request_id=request.id) self.assertEqual(summary["response_count"], 1) self.assertEqual(first_participant.status, "responded") self.assertIsNotNone(first_participant.responded_at) values = {result["option_id"]: result["values"] for result in summary["option_results"]} self.assertEqual(values[first_slot.poll_option_id]["available"], 1) self.assertEqual(values[second_slot.poll_option_id]["maybe"], 1) close_scheduling_request(self.session, tenant_id="tenant-1", request_id=request.id) decided = decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingDecisionRequest(slot_id=first_slot.id, handoff_to_calendar=True), ) poll = get_poll(self.session, tenant_id="tenant-1", poll_id=request.poll_id) self.assertEqual(decided.status, "handed_off") self.assertEqual(decided.selected_slot_id, first_slot.id) self.assertEqual(decided.metadata_["calendar_handoff"]["status"], "pending") self.assertEqual(poll.status, "decided") self.assertEqual(poll.workflow_state, "handed_off") def test_calendar_freebusy_holds_and_final_event_creation(self) -> None: self._calendar() self.session.add( CalendarSyncSource( id="source-1", tenant_id="tenant-1", calendar_id="calendar-1", source_kind="caldav", collection_url="https://dav.example.test/cal/", auth_type="none", sync_enabled=True, sync_interval_seconds=900, sync_direction="two_way", conflict_policy="etag", metadata_={}, ) ) payload = self._payload().model_copy(update={"title": "x" * 500}) request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) first_slot = request.slots[0] self.session.add( CalendarEvent( tenant_id="tenant-1", calendar_id="calendar-1", uid="busy@example.test", sequence=0, summary="Busy", status="CONFIRMED", transparency="OPAQUE", classification="PUBLIC", start_at=first_slot.start_at, end_at=first_slot.end_at, all_day=False, attendees=[], categories=[], rdate=[], exdate=[], reminders=[], attachments=[], related_to=[], source_kind="local", icalendar={}, metadata_={}, ) ) self.session.flush() request, warnings = evaluate_calendar_freebusy(self.session, tenant_id="tenant-1", request_id=request.id) self.assertEqual(warnings, []) self.assertEqual(request.slots[0].freebusy_status, "busy") self.assertEqual(request.slots[1].freebusy_status, "free") with patch("govoplan_calendar.backend.service.caldav_client_for_source") as caldav_client: request, hold_event_ids, warnings = create_tentative_calendar_holds( self.session, tenant_id="tenant-1", user_id="user-1", request_id=request.id, ) caldav_client.assert_not_called() self.assertEqual(warnings, []) self.assertEqual(len(hold_event_ids), 2) self.assertTrue(all(slot.tentative_hold_event_id for slot in request.slots)) request, duplicate_hold_ids, warnings = create_tentative_calendar_holds( self.session, tenant_id="tenant-1", user_id="user-1", request_id=request.id, ) self.assertEqual(warnings, []) self.assertEqual(duplicate_hold_ids, []) with self.assertRaisesRegex(SchedulingError, "only be created for a decided"): create_final_calendar_event( self.session, tenant_id="tenant-1", user_id="user-1", request_id=request.id, ) close_scheduling_request(self.session, tenant_id="tenant-1", request_id=request.id) decided = decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingDecisionRequest(slot_id=request.slots[1].id, handoff_to_calendar=False), user_id="user-1", ) with self.assertRaisesRegex(SchedulingError, "only available before a scheduling decision"): evaluate_calendar_freebusy( self.session, tenant_id="tenant-1", request_id=decided.id, ) with self.assertRaisesRegex(SchedulingError, "only available before a scheduling decision"): create_tentative_calendar_holds( self.session, tenant_id="tenant-1", user_id="user-1", request_id=decided.id, ) with patch("govoplan_calendar.backend.service.caldav_client_for_source") as caldav_client: decided, event_id, warnings = create_final_calendar_event( self.session, tenant_id="tenant-1", user_id="user-1", request_id=decided.id, ) caldav_client.assert_not_called() self.assertEqual(warnings, []) self.assertIsNotNone(event_id) self.assertEqual(decided.status, "handed_off") self.assertEqual(decided.calendar_event_id, event_id) self.assertEqual(decided.metadata_["calendar_handoff"]["status"], "accepted") self.assertEqual( decided.metadata_["calendar_handoff"]["last_known_external_state"], "queued", ) self.assertIsNotNone(decided.metadata_["calendar_handoff"]["outbox_operation_id"]) with self.assertRaisesRegex(SchedulingError, "only available before a scheduling decision"): evaluate_calendar_freebusy( self.session, tenant_id="tenant-1", request_id=decided.id, ) with self.assertRaisesRegex(SchedulingError, "only available before a scheduling decision"): create_tentative_calendar_holds( self.session, tenant_id="tenant-1", user_id="user-1", request_id=decided.id, ) self.assertTrue( all( slot.metadata_["calendar_hold"]["last_known_external_state"] == "queued" for slot in request.slots ) ) generated_events = ( self.session.query(CalendarEvent) .filter(CalendarEvent.metadata_["scheduling_request_id"].as_string() == request.id) .all() ) self.assertTrue(generated_events) self.assertTrue(all(event.classification == "PRIVATE" for event in generated_events)) self.assertTrue(all(len(event.summary) <= 500 for event in generated_events)) event_count = len(generated_events) decided_again, same_event_id, warnings = create_final_calendar_event( self.session, tenant_id="tenant-1", user_id="user-1", request_id=decided.id, ) self.assertEqual(warnings, []) self.assertEqual(same_event_id, event_id) self.assertEqual(decided_again.calendar_event_id, event_id) self.assertEqual( self.session.query(CalendarEvent) .filter(CalendarEvent.metadata_["scheduling_request_id"].as_string() == request.id) .count(), event_count, ) exact_repeat = decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=decided.id, payload=SchedulingDecisionRequest( slot_id=request.slots[1].id, handoff_to_calendar=True, ), user_id="user-1", ) self.assertEqual(exact_repeat.calendar_event_id, event_id) with self.assertRaisesRegex(SchedulingError, "different slot"): decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=decided.id, payload=SchedulingDecisionRequest( slot_id=request.slots[0].id, handoff_to_calendar=True, ), user_id="user-1", ) def test_notification_outbox_jobs_are_created_and_listed(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) reminder_jobs = create_scheduling_notification_jobs( self.session, tenant_id="tenant-1", request_id=request.id, event_kind="reminder", ) all_jobs = list_scheduling_notifications(self.session, tenant_id="tenant-1", request_id=request.id) self.assertEqual(len(reminder_jobs), 2) self.assertEqual(len(all_jobs), 2) self.assertTrue(all(job.status == "pending" for job in reminder_jobs)) organizer_jobs = list_visible_scheduling_notifications( self.session, tenant_id="tenant-1", request_id=request.id, actor_ids=("user-1",), ) alice_jobs = list_visible_scheduling_notifications( self.session, tenant_id="tenant-1", request_id=request.id, actor_ids=("alice@example.test",), ) outsider_jobs = list_visible_scheduling_notifications( self.session, tenant_id="tenant-1", request_id=request.id, actor_ids=("outsider@example.test",), ) self.assertEqual(len(all_jobs), len(organizer_jobs)) self.assertTrue(alice_jobs) self.assertTrue(all(job.recipient == "alice@example.test" for job in alice_jobs)) self.assertEqual([], outsider_jobs) def test_request_visibility_is_limited_to_organizer_participants_and_managers(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) self.assertEqual( [request.id], [ item.id for item in list_visible_scheduling_requests( self.session, tenant_id="tenant-1", actor_ids=("alice@example.test",), ) ], ) with self.assertRaisesRegex(SchedulingError, "Scheduling request not found"): get_visible_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, actor_ids=("outsider@example.test",), ) self.assertEqual( request.id, get_visible_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, actor_ids=("outsider@example.test",), can_manage=True, ).id, ) ordinary_writer = self._principal( "unrelated-writer", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_WRITE_SCOPE}, ) administrator = self._principal( "scheduling-admin", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_ADMIN_SCOPE}, ) self.assertEqual( [], api_list_scheduling_requests( status_filter=None, session=self.session, principal=ordinary_writer, ).requests, ) self.assertEqual( [request.id], [ item.id for item in api_list_scheduling_requests( status_filter=None, session=self.session, principal=administrator, ).requests ], ) def test_participant_list_and_get_redact_management_and_connector_internals(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) request.calendar_event_id = "calendar-event-internal" request.metadata_ = {"connector_ref": "connector-internal"} slot = request.slots[0] slot.freebusy_status = "busy" slot.freebusy_conflicts = [ { "calendar_id": "calendar-1", "event_id": "busy-event-internal", "uid": "busy-uid-internal", "recurrence_id": "busy-recurrence-internal", "start_at": "2026-07-20T09:15:00+00:00", "end_at": "2026-07-20T09:45:00+00:00", "status": "CONFIRMED", } ] slot.tentative_hold_event_id = "hold-event-internal" slot.metadata_ = {"provider_ref": "provider-internal"} participant = request.participants[0] participant.respondent_id = "alice-identity-internal" participant.metadata_ = {"directory_ref": "directory-internal"} self.session.flush() principal = self._principal( "alice-account", email="alice@example.test", scopes={SCHEDULING_READ_SCOPE}, ) listed = api_list_scheduling_requests( status_filter=None, session=self.session, principal=principal, ).requests fetched = api_get_scheduling_request( request.id, session=self.session, principal=principal, ) self.assertEqual(len(listed), 1) for response in (listed[0], fetched): self.assertIsNone(response.tenant_id) self.assertIsNone(response.poll_id) self.assertIsNone(response.organizer_user_id) self.assertIsNone(response.calendar_integration_enabled) self.assertIsNone(response.calendar_id) self.assertIsNone(response.calendar_freebusy_enabled) self.assertIsNone(response.calendar_hold_enabled) self.assertIsNone(response.create_calendar_event_on_decision) self.assertIsNone(response.calendar_event_id) self.assertIsNone( response.public_participation_policy_enforcement_available ) self.assertEqual(response.metadata, {}) self.assertEqual(response.participant_aggregate.total, 2) self.assertEqual(len(response.participants), 1) own = response.participants[0] self.assertTrue(own.is_current_participant) self.assertEqual(own.email, "alice@example.test") self.assertIsNone(own.respondent_id) self.assertIsNone(own.poll_invitation_id) self.assertEqual(own.metadata, {}) projected_slot = response.slots[0] self.assertIsNone(projected_slot.poll_option_id) self.assertEqual(projected_slot.freebusy_status, "busy") self.assertEqual( projected_slot.freebusy_conflicts, [ { "start_at": "2026-07-20T09:15:00+00:00", "end_at": "2026-07-20T09:45:00+00:00", "status": "CONFIRMED", } ], ) self.assertIsNone(projected_slot.tentative_hold_event_id) self.assertEqual(projected_slot.metadata, {}) organizer = self._principal( "user-1", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_WRITE_SCOPE}, ) management = api_get_scheduling_request( request.id, session=self.session, principal=organizer, ) self.assertEqual(management.tenant_id, "tenant-1") self.assertEqual(management.poll_id, request.poll_id) self.assertEqual(management.calendar_id, "calendar-1") self.assertEqual(management.calendar_event_id, "calendar-event-internal") self.assertEqual(management.metadata["connector_ref"], "connector-internal") self.assertEqual(management.slots[0].poll_option_id, slot.poll_option_id) self.assertEqual( management.slots[0].freebusy_conflicts[0]["uid"], "busy-uid-internal", ) self.assertEqual( management.slots[0].tentative_hold_event_id, "hold-event-internal", ) def test_result_visibility_follows_existing_after_close_policy(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) with self.assertRaisesRegex(SchedulingError, "Scheduling results are not visible"): require_visible_scheduling_results( self.session, request=request, actor_ids=("alice@example.test",), ) require_visible_scheduling_results( self.session, request=request, actor_ids=("user-1",), ) close_scheduling_request(self.session, tenant_id="tenant-1", request_id=request.id) require_visible_scheduling_results( self.session, request=request, actor_ids=("alice@example.test",), ) def test_update_synchronizes_backing_poll_policy(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) new_deadline = datetime(2026, 7, 30, 12, tzinfo=timezone.utc) update_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingRequestUpdateRequest( title="Restricted steering group", description="Updated private context", deadline_at=new_deadline, allow_participant_updates=False, result_visibility="organizer", ), ) poll = get_poll(self.session, tenant_id="tenant-1", poll_id=request.poll_id) self.assertEqual(poll.title, "Restricted steering group") self.assertEqual(poll.description, "Updated private context") self.assertEqual(poll.visibility, "private") self.assertEqual(poll.result_visibility, "organizer") self.assertFalse(poll.allow_anonymous) self.assertFalse(poll.allow_response_update) self.assertEqual(poll.closes_at.replace(tzinfo=timezone.utc), new_deadline) def test_lifecycle_guards_and_terminal_actions_are_idempotent(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload().model_copy( update={"calendar": SchedulingCalendarPreferences()} ), ) self.assertIs(open_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ), request) close_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ) closed_at = get_poll( self.session, tenant_id="tenant-1", poll_id=request.poll_id, ).closed_at self.assertIs(close_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ), request) self.assertEqual( get_poll(self.session, tenant_id="tenant-1", poll_id=request.poll_id).closed_at, closed_at, ) with self.assertRaisesRegex(SchedulingError, "Only draft"): open_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, ) decided = decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingDecisionRequest( slot_id=request.slots[0].id, handoff_to_calendar=False, ), ) decision_job_count = ( self.session.query(SchedulingNotification) .filter( SchedulingNotification.request_id == request.id, SchedulingNotification.event_kind == "decision", ) .count() ) self.assertIs( decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingDecisionRequest( slot_id=request.slots[0].id, handoff_to_calendar=False, ), ), decided, ) self.assertEqual( self.session.query(SchedulingNotification) .filter( SchedulingNotification.request_id == request.id, SchedulingNotification.event_kind == "decision", ) .count(), decision_job_count, ) with self.assertRaisesRegex(SchedulingError, "different slot"): decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=request.id, payload=SchedulingDecisionRequest( slot_id=request.slots[1].id, handoff_to_calendar=False, ), ) for callback in (open_scheduling_request, close_scheduling_request, cancel_scheduling_request): with self.assertRaises(SchedulingError): callback( self.session, tenant_id="tenant-1", request_id=request.id, ) cancelled_request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload().model_copy( update={ "title": "Cancelled request", "calendar": SchedulingCalendarPreferences(), } ), ) cancelled = cancel_scheduling_request( self.session, tenant_id="tenant-1", request_id=cancelled_request.id, ) cancellation_job_count = ( self.session.query(SchedulingNotification) .filter( SchedulingNotification.request_id == cancelled_request.id, SchedulingNotification.event_kind == "cancellation", ) .count() ) self.assertIs( cancel_scheduling_request( self.session, tenant_id="tenant-1", request_id=cancelled_request.id, ), cancelled, ) self.assertEqual( self.session.query(SchedulingNotification) .filter( SchedulingNotification.request_id == cancelled_request.id, SchedulingNotification.event_kind == "cancellation", ) .count(), cancellation_job_count, ) with self.assertRaisesRegex(SchedulingError, "Only draft"): open_scheduling_request( self.session, tenant_id="tenant-1", request_id=cancelled_request.id, ) with self.assertRaisesRegex(SchedulingError, "Only collecting"): close_scheduling_request( self.session, tenant_id="tenant-1", request_id=cancelled_request.id, ) with self.assertRaisesRegex(SchedulingError, "Only closed"): decide_scheduling_request( self.session, tenant_id="tenant-1", request_id=cancelled_request.id, payload=SchedulingDecisionRequest( slot_id=cancelled_request.slots[0].id, handoff_to_calendar=False, ), ) def test_backing_poll_is_hidden_from_unrelated_poll_reader(self) -> None: request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) outsider = self._principal( "outsider", scopes={"poll:poll:read", "poll:response:write"}, ) self.assertEqual( api_list_polls(status_filter=None, kind=None, session=self.session, principal=outsider).polls, [], ) with self.assertRaises(HTTPException) as hidden: api_get_poll(request.poll_id, session=self.session, principal=outsider) self.assertEqual(hidden.exception.status_code, 404) with self.assertRaises(HTTPException) as blocked_vote: api_submit_poll_response( request.poll_id, PollSubmitResponseRequest( answers=[PollAnswerInput(option_id=request.slots[0].poll_option_id, value="available")] ), session=self.session, principal=outsider, ) self.assertEqual(blocked_vote.exception.status_code, 404) organizer = self._principal("user-1", scopes={"poll:poll:read"}) self.assertEqual(api_get_poll(request.poll_id, session=self.session, principal=organizer).id, request.poll_id) def test_module_owned_poll_rejects_direct_response_and_uses_current_participant(self) -> None: participants = [ SchedulingParticipantInput( respondent_id="alice-membership", display_name="Alice", email="alice@example.test", ), SchedulingParticipantInput( respondent_id="bob-membership", display_name="Bob", email="bob@example.test", ), ] payload = self._payload().model_copy( update={"calendar": SchedulingCalendarPreferences(), "participants": participants} ) request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) target = request.participants[1] self._issue_copy(request, target) attacker = self._principal( "attacker", email="alice@example.test", membership_id="alice-membership", scopes={ SCHEDULING_READ_SCOPE, SCHEDULING_RESPOND_SCOPE, "poll:poll:read", "poll:response:write", }, ) with self.assertRaises(HTTPException) as direct_response: api_submit_poll_response( request.poll_id, PollSubmitResponseRequest( answers=[ PollAnswerInput( option_id=request.slots[0].poll_option_id, value="available", ) ], metadata={"invitation_id": target.poll_invitation_id}, ), session=self.session, principal=attacker, ) listed = api_list_scheduling_requests( status_filter=None, session=self.session, principal=attacker, ) current = api_get_my_scheduling_availability( request.id, session=self.session, principal=attacker, ) self.assertEqual(direct_response.exception.status_code, 404) self.assertEqual([request.id], [item.id for item in listed.requests]) self.assertFalse(current.has_response) 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=attacker, ) current = api_get_my_scheduling_availability( request.id, session=self.session, principal=attacker, ) 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") self.assertIsNotNone(alice.responded_at) self.assertEqual(bob.status, "invited") def test_scheduling_participant_scope_can_respond_without_poll_scope(self) -> None: payload = self._payload().model_copy( update={ "calendar": SchedulingCalendarPreferences(), "participants": [ SchedulingParticipantInput( respondent_id="alice-account", display_name="Alice", email="alice@example.test", ) ], } ) request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) alice = self._principal( "alice-account", email="alice@example.test", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_RESPOND_SCOPE}, ) result = 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=alice, ) self.assertEqual(result.request.id, request.id) self.assertEqual(request.participants[0].status, "responded") self.assertIsNotNone(request.participants[0].responded_at) summary = scheduling_request_summary( self.session, tenant_id="tenant-1", request_id=request.id, ) self.assertEqual(summary["response_count"], 1) outsider = self._principal( "outsider-account", email="outsider@example.test", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_RESPOND_SCOPE}, ) with self.assertRaises(HTTPException) as hidden: api_submit_scheduling_availability( request.id, SchedulingAvailabilityResponseRequest( answers=[ SchedulingAvailabilityAnswerInput( slot_id=request.slots[0].id, value="unavailable", option_revision=scheduling_slot_revision(request.slots[0]), ) ] ), session=self.session, principal=outsider, ) self.assertEqual(hidden.exception.status_code, 404) def test_signed_link_and_in_module_response_update_the_same_poll_response(self) -> None: payload = self._payload().model_copy( update={ "calendar": SchedulingCalendarPreferences(), "participants": [ SchedulingParticipantInput( display_name="Alice", email="alice@example.test", ) ], } ) request, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) participant = request.participants[0] self.assertEqual(tokens, {}) token = self._issue_copy(request, participant) 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]), ) ] ), client_address="127.0.0.1", ) alice = self._principal( "alice-account", email="alice@example.test", scopes={SCHEDULING_READ_SCOPE, SCHEDULING_RESPOND_SCOPE}, ) api_submit_scheduling_availability( request.id, SchedulingAvailabilityResponseRequest( answers=[ 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]), ), ] ), session=self.session, principal=alice, ) responses = ( self.session.query(PollResponse) .filter(PollResponse.poll_id == request.poll_id) .all() ) self.assertEqual(len(responses), 1) self.assertEqual( responses[0].respondent_id, participant.respondent_id, ) self.assertEqual( responses[0].metadata_["invitation_id"], participant.poll_invitation_id, ) self.assertEqual( [answer["value"] for answer in responses[0].answers], ["unavailable", "maybe"], ) summary = scheduling_request_summary( self.session, tenant_id="tenant-1", request_id=request.id, ) self.assertEqual(summary["response_count"], 1) def test_participant_request_projection_redacts_other_participant_details(self) -> None: participants = [ SchedulingParticipantInput( respondent_id="alice-id", display_name="Alice", email="alice@example.test", metadata={"private": "alice"}, ), SchedulingParticipantInput( respondent_id="bob-id", display_name="Bob", email="bob@example.test", metadata={"private": "bob"}, ), ] payload = self._payload().model_copy( update={ "calendar": SchedulingCalendarPreferences(), "participants": participants, "result_visibility": "public", "participant_visibility": "names_and_statuses", } ) request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) alice = self._principal( "alice-account", email="alice@example.test", membership_id="alice-id", scopes={SCHEDULING_READ_SCOPE}, ) responses = [ api_list_scheduling_requests(status_filter=None, session=self.session, principal=alice).requests[0], api_get_scheduling_request(request.id, session=self.session, principal=alice), api_scheduling_summary(request.id, session=self.session, principal=alice).request, ] for response in responses: own = next(item for item in response.participants if item.display_name == "Alice") other = next(item for item in response.participants if item.display_name == "Bob") self.assertTrue(own.is_current_participant) self.assertIsNone(own.respondent_id) self.assertEqual(own.email, "alice@example.test") self.assertEqual(own.metadata, {}) self.assertIsNone(own.poll_invitation_id) self.assertIsNone(own.invitation_token) self.assertFalse(other.is_current_participant) self.assertIsNone(other.respondent_id) self.assertIsNone(other.email) self.assertIsNone(other.participant_type) self.assertIsNone(other.required) self.assertIsNone(other.poll_invitation_id) self.assertIsNone(other.last_invited_at) self.assertIsNone(other.responded_at) self.assertEqual(other.metadata, {}) self.assertIsNone(other.invitation_token) def test_calendar_side_effect_routes_require_calendar_scopes(self) -> None: self._calendar() request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) scheduling_writer = self._principal("writer", scopes={SCHEDULING_WRITE_SCOPE}) for callback in ( api_evaluate_calendar_freebusy, api_create_tentative_calendar_holds, api_create_final_calendar_event, ): with self.assertRaises(HTTPException) as denied: callback(request.id, session=self.session, principal=scheduling_writer) self.assertEqual(denied.exception.status_code, 403) close_scheduling_request(self.session, tenant_id="tenant-1", request_id=request.id) with self.assertRaises(HTTPException) as denied_decision: api_decide_scheduling_request( request.id, SchedulingDecisionRequest(slot_id=request.slots[0].id), session=self.session, principal=scheduling_writer, ) self.assertEqual(denied_decision.exception.status_code, 403) self.assertIsNone(request.selected_slot_id) availability_reader = self._principal( "user-1", scopes={SCHEDULING_WRITE_SCOPE, CALENDAR_AVAILABILITY_READ_SCOPE}, ) freebusy = api_evaluate_calendar_freebusy(request.id, session=self.session, principal=availability_reader) self.assertTrue(freebusy.updated_slot_ids) event_writer = self._principal( "user-1", scopes={SCHEDULING_WRITE_SCOPE, CALENDAR_EVENT_WRITE_SCOPE}, ) holds = api_create_tentative_calendar_holds(request.id, session=self.session, principal=event_writer) self.assertTrue(holds.created_event_ids) def test_decision_rechecks_calendar_scope_after_locked_state_refresh(self) -> None: self._calendar() request, _tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload(), ) request.calendar_integration_enabled = False request.create_calendar_event_on_decision = False self.session.flush() close_scheduling_request(self.session, tenant_id="tenant-1", request_id=request.id) scheduling_writer = self._principal( "user-1", scopes={SCHEDULING_WRITE_SCOPE}, ) def refreshed_locked_request(*_args, **_kwargs): # Simulate a concurrent preference update committed after an old # router preflight but before the service obtains its row lock. request.calendar_integration_enabled = True request.create_calendar_event_on_decision = True return request with ( patch( "govoplan_scheduling.backend.service._lock_scheduling_calendar_handoff", side_effect=refreshed_locked_request, ), self.assertRaises(HTTPException) as denied, ): api_decide_scheduling_request( request.id, SchedulingDecisionRequest(slot_id=request.slots[0].id), session=self.session, principal=scheduling_writer, ) self.assertEqual(denied.exception.status_code, 403) self.assertIsNone(request.selected_slot_id) self.assertIsNone(request.calendar_event_id) def test_explicit_invitation_send_uses_signed_link_and_verified_recipient_id(self) -> None: class CapturingNotificationProvider: def __init__(self) -> None: self.requests = [] def enqueue_notification(self, _session, request, *, enqueue_delivery): self.requests.append(request) return {"id": f"notification-{len(self.requests)}", "status": "queued"} provider = CapturingNotificationProvider() participants = [ SchedulingParticipantInput( respondent_id="alice-id", display_name="Alice", email="alice@example.test", ), SchedulingParticipantInput( respondent_id="bob-id", display_name="Bob", email="bob@example.test", ), ] payload = self._payload().model_copy( update={"calendar": SchedulingCalendarPreferences(), "participants": participants} ) with patch( "govoplan_scheduling.backend.service.notification_dispatch_provider", return_value=provider, ): request, automatic_tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) self.assertEqual(provider.requests, []) results = [ issue_scheduling_participant_invitation( self.session, tenant_id=request.tenant_id, request_id=request.id, participant_id=participant.id, action="send", ) for participant in request.participants ] self.assertEqual(automatic_tokens, {}) self.assertTrue(all(result.action_url is None for result in results)) self.assertTrue(all(result.status == "queued" for result in results)) self.assertEqual(len(provider.requests), 2) self.assertEqual({item.recipient_id for item in provider.requests}, {"alice-id", "bob-id"}) action_urls = {item.action_url for item in provider.requests} self.assertTrue( all( isinstance(action_url, str) and action_url.startswith(f"/scheduling/public/{request.id}/") for action_url in action_urls ) ) tokens = { str(action_url).rsplit("/", 1)[-1] for action_url in action_urls } local_notifications = list_scheduling_notifications( self.session, tenant_id="tenant-1", request_id=request.id, ) for notification in local_notifications: serialized = repr({"payload": notification.payload, "metadata": notification.metadata_}) self.assertTrue(all(token not in serialized for token in tokens)) def test_invitation_router_rotates_revokes_and_enforces_management_policy(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] organizer = self._principal( "user-1", scopes={SCHEDULING_WRITE_SCOPE}, ) response_headers = Response() with patch("govoplan_scheduling.backend.router.audit_event") as audit: first = api_issue_scheduling_participant_invitation( request.id, participant.id, SchedulingInvitationActionRequest(action="copy"), response_headers, session=self.session, principal=organizer, ) self.assertEqual(automatic_tokens, {}) self.assertEqual(first.action, "copy") self.assertEqual(first.status, "issued") self.assertIsNotNone(first.action_url) self.assertEqual( response_headers.headers["cache-control"], "no-store, private", ) first_token = str(first.action_url).rsplit("/", 1)[-1] self.assertNotIn(first_token, repr([call.kwargs for call in audit.call_args_list])) unrelated_writer = self._principal( "unrelated", scopes={SCHEDULING_WRITE_SCOPE}, ) with self.assertRaises(HTTPException) as denied: api_issue_scheduling_participant_invitation( request.id, participant.id, SchedulingInvitationActionRequest(action="copy"), Response(), session=self.session, principal=unrelated_writer, ) self.assertEqual(denied.exception.status_code, 403) self.assertEqual( get_public_scheduling_participation( self.session, request_id=request.id, token=first_token, payload=SchedulingPublicParticipationAccessRequest(), client_address="127.0.0.1", )["request_id"], request.id, ) administrator = self._principal( "administrator", scopes={SCHEDULING_ADMIN_SCOPE}, ) with patch("govoplan_scheduling.backend.router.audit_event") as audit: rotated = api_issue_scheduling_participant_invitation( request.id, participant.id, SchedulingInvitationActionRequest(action="copy"), Response(), session=self.session, principal=administrator, ) rotated_token = str(rotated.action_url).rsplit("/", 1)[-1] self.assertNotEqual(rotated_token, first_token) self.assertNotIn(rotated_token, repr([call.kwargs for call in audit.call_args_list])) with self.assertRaisesRegex( Exception, "Scheduling participation link or credentials are invalid", ): get_public_scheduling_participation( self.session, request_id=request.id, token=first_token, payload=SchedulingPublicParticipationAccessRequest(), client_address="127.0.0.1", ) with patch("govoplan_scheduling.backend.router.audit_event") as audit: revoked = api_revoke_scheduling_participant_invitation( request.id, participant.id, Response(), session=self.session, principal=organizer, ) replayed = api_revoke_scheduling_participant_invitation( request.id, participant.id, Response(), session=self.session, principal=organizer, ) self.assertFalse(revoked.replayed) self.assertTrue(replayed.replayed) self.assertNotIn(rotated_token, repr([call.kwargs for call in audit.call_args_list])) with self.assertRaisesRegex( Exception, "Scheduling participation link or credentials are invalid", ): get_public_scheduling_participation( self.session, request_id=request.id, token=rotated_token, payload=SchedulingPublicParticipationAccessRequest(), client_address="127.0.0.1", ) def test_links_can_be_issued_for_draft_and_closed_requests(self) -> None: organizer = self._principal( "user-1", scopes={SCHEDULING_WRITE_SCOPE}, ) for request_status in ("draft", "collecting"): request, _automatic_tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=self._payload().model_copy( update={ "status": request_status, "calendar": SchedulingCalendarPreferences(), } ), ) if request_status == "collecting": close_scheduling_request( self.session, tenant_id=request.tenant_id, request_id=request.id, ) expected_status = "closed" else: expected_status = "draft" with patch("govoplan_scheduling.backend.router.audit_event"): issued = api_issue_scheduling_participant_invitation( request.id, request.participants[0].id, SchedulingInvitationActionRequest(action="copy"), Response(), session=self.session, principal=organizer, ) self.assertEqual(request.status, expected_status) self.assertTrue( str(issued.action_url).startswith( f"/scheduling/public/{request.id}/" ) ) def test_send_response_audit_error_and_durable_job_never_store_token(self) -> None: class EchoingFailureNotificationProvider: def __init__(self) -> None: self.requests = [] def enqueue_notification(self, _session, request, *, enqueue_delivery): self.requests.append(request) raise RuntimeError(f"failed dispatch for {request.action_url}") provider = EchoingFailureNotificationProvider() 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] 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, ): result = api_issue_scheduling_participant_invitation( request.id, participant.id, SchedulingInvitationActionRequest(action="send"), Response(), session=self.session, principal=organizer, ) self.assertEqual(len(provider.requests), 1) action_url = provider.requests[0].action_url self.assertIsInstance(action_url, str) token = str(action_url).rsplit("/", 1)[-1] self.assertIsNone(result.action_url) self.assertEqual(result.status, "failed") self.assertEqual(result.notification.error, "Notification center enqueue failed") self.assertNotIn(token, repr(result.model_dump())) self.assertNotIn(token, repr([call.kwargs for call in audit.call_args_list])) notification = self.session.query(SchedulingNotification).filter( SchedulingNotification.id == result.notification.id ).one() durable_projection = repr( { "payload": notification.payload, "metadata": notification.metadata_, "error": notification.error, } ) self.assertNotIn(token, durable_projection) def test_external_participants_can_be_rejected(self) -> None: payload = self._payload().model_copy(update={"allow_external_participants": False}) with self.assertRaisesRegex(SchedulingError, "External participants are not allowed"): create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) if __name__ == "__main__": unittest.main()