govoplan-scheduling/tests/test_service.py

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_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, SchedulingParticipant, SchedulingRequest
from govoplan_scheduling.backend.schemas import (
    SchedulingCalendarPreferences,
    SchedulingCandidateSlotInput,
    SchedulingDecisionRequest,
    SchedulingParticipantInput,
    SchedulingRequestCreateRequest,
)
from govoplan_scheduling.backend.service import (
    SchedulingError,
    close_scheduling_request,
    create_scheduling_request,
    decide_scheduling_request,
    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__,
                SchedulingRequest.__table__,
                SchedulingCandidateSlot.__table__,
                SchedulingParticipant.__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__,
                SchedulingCandidateSlot.__table__,
                SchedulingRequest.__table__,
                PollInvitation.__table__,
                PollResponse.__table__,
                PollOption.__table__,
                Poll.__table__,
            ],
        )
        self.engine.dispose()

    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,
                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:
        request, tokens = create_scheduling_request(
            self.session,
            tenant_id="tenant-1",
            user_id="user-1",
            payload=self._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_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()