1237 lines
48 KiB
Python
1237 lines
48 KiB
Python
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",
|
|
"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.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()
|