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 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, 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, submit_poll_response_with_token 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 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, SchedulingParticipantInput, 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_list_scheduling_requests, api_scheduling_summary, api_submit_scheduling_availability, ) 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, list_scheduling_notifications, list_visible_scheduling_notifications, list_visible_scheduling_requests, open_scheduling_request, require_visible_scheduling_results, scheduling_request_summary, scheduling_slot_revision, 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__, 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__, 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 test_create_request_creates_poll_slots_and_signed_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(len(tokens), 2) self.assertTrue(all(participant.poll_invitation_id for participant in request.participants)) 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] submit_poll_response_with_token( self.session, token=tokens[first_participant.id], payload=PollSubmitResponseRequest( answers=[ PollAnswerInput(option_id=first_slot.poll_option_id, value="available"), PollAnswerInput(option_id=second_slot.poll_option_id, value="maybe"), ] ), ) 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.assertGreaterEqual(len(all_jobs), 4) 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, ) 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_authenticated_response_reconciles_only_the_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] 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", }, ) 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.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") 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] submit_poll_response_with_token( self.session, token=tokens[participant.id], payload=PollSubmitResponseRequest( answers=[ PollAnswerInput( option_id=request.slots[0].poll_option_id, value="available", ) ] ), ) 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, f"invitation:{participant.poll_invitation_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", } ) 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.assertEqual(own.respondent_id, "alice-id") self.assertEqual(own.email, "alice@example.test") self.assertEqual(own.metadata, {"private": "alice"}) self.assertIsNotNone(own.poll_invitation_id) self.assertIsNone(own.invitation_token) self.assertIsNone(other.respondent_id) self.assertIsNone(other.email) 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( "reader", 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( "event-writer", 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( "writer", 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_initial_invitation_notifications_use_signed_poll_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, tokens = create_scheduling_request( self.session, tenant_id="tenant-1", user_id="user-1", payload=payload, ) self.assertEqual(len(provider.requests), 2) self.assertEqual({item.recipient_id for item in provider.requests}, {"alice-id", "bob-id"}) self.assertEqual( {item.action_url for item in provider.requests}, {f"/poll/public/{token}" for token in tokens.values()}, ) 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.values())) 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()