301 lines
12 KiB
Python
301 lines
12 KiB
Python
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()
|