from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.db.base import Base from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_access.backend.db.models import Account, User from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarSyncSource from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse 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.schemas import ( SchedulingCalendarPreferences, SchedulingCandidateSlotInput, SchedulingDecisionRequest, SchedulingParticipantInput, SchedulingRequestCreateRequest, ) from govoplan_scheduling.backend.service import ( SchedulingError, close_scheduling_request, create_final_calendar_event, create_scheduling_notification_jobs, create_scheduling_request, create_tentative_calendar_holds, decide_scheduling_request, evaluate_calendar_freebusy, list_scheduling_notifications, scheduling_request_summary, ) class SchedulingServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__, CalendarCollection.__table__, CalendarEvent.__table__, CalendarSyncSource.__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__, CalendarSyncSource.__table__, CalendarEvent.__table__, CalendarCollection.__table__, PollInvitation.__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 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.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_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() payload = self._payload() 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") request, hold_event_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(len(hold_event_ids), 2) self.assertTrue(all(slot.tentative_hold_event_id for slot in request.slots)) 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", ) decided, 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.assertIsNotNone(event_id) self.assertEqual(decided.status, "handed_off") self.assertEqual(decided.calendar_event_id, event_id) self.assertEqual(decided.metadata_["calendar_handoff"]["status"], "created") 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)) 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()