2047 lines
79 KiB
Python
2047 lines
79 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, Response
|
|
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,
|
|
PollParticipationSubmission,
|
|
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
|
|
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 ADMIN_SCOPE as SCHEDULING_ADMIN_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,
|
|
SchedulingInvitationActionRequest,
|
|
SchedulingInvitationRevokeRequest,
|
|
SchedulingParticipantInput,
|
|
SchedulingPublicParticipationAccessRequest,
|
|
SchedulingPublicParticipationSubmitRequest,
|
|
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_issue_scheduling_participant_invitation,
|
|
api_list_scheduling_requests,
|
|
api_scheduling_summary,
|
|
api_submit_scheduling_availability,
|
|
api_revoke_scheduling_participant_invitation,
|
|
)
|
|
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,
|
|
get_public_scheduling_participation,
|
|
issue_scheduling_participant_invitation,
|
|
list_scheduling_notifications,
|
|
list_visible_scheduling_notifications,
|
|
list_visible_scheduling_requests,
|
|
open_scheduling_request,
|
|
require_visible_scheduling_results,
|
|
scheduling_request_summary,
|
|
scheduling_participant_revision,
|
|
scheduling_slot_revision,
|
|
submit_public_scheduling_participation,
|
|
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__,
|
|
PollParticipationSubmission.__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__,
|
|
PollParticipationSubmission.__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 _issue_copy(
|
|
self,
|
|
request: SchedulingRequest,
|
|
participant: SchedulingParticipant,
|
|
) -> str:
|
|
result = issue_scheduling_participant_invitation(
|
|
self.session,
|
|
tenant_id=request.tenant_id,
|
|
request_id=request.id,
|
|
participant_id=participant.id,
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
action="copy",
|
|
)
|
|
self.assertIsNotNone(result.action_url)
|
|
return str(result.action_url).rsplit("/", 1)[-1]
|
|
|
|
def test_create_request_creates_poll_slots_without_implicit_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(tokens, {})
|
|
self.assertTrue(
|
|
all(
|
|
participant.status == "draft"
|
|
and participant.poll_invitation_id is None
|
|
for participant in request.participants
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.session.query(SchedulingNotification).filter(
|
|
SchedulingNotification.request_id == request.id
|
|
).count(),
|
|
0,
|
|
)
|
|
|
|
def test_draft_save_does_not_issue_or_deliver_public_invitations(self) -> None:
|
|
class RejectingNotificationProvider:
|
|
def enqueue_notification(self, *_args, **_kwargs):
|
|
raise AssertionError("A draft must not enqueue invitation delivery")
|
|
|
|
with patch(
|
|
"govoplan_scheduling.backend.service.notification_dispatch_provider",
|
|
return_value=RejectingNotificationProvider(),
|
|
):
|
|
request, tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(update={"status": "draft"}),
|
|
)
|
|
|
|
self.assertEqual(tokens, {})
|
|
self.assertEqual(
|
|
self.session.query(PollInvitation).filter(
|
|
PollInvitation.poll_id == request.poll_id
|
|
).count(),
|
|
0,
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
participant.status == "draft"
|
|
and participant.poll_invitation_id is None
|
|
for participant in request.participants
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
self.session.query(SchedulingNotification).filter(
|
|
SchedulingNotification.request_id == request.id
|
|
).count(),
|
|
0,
|
|
)
|
|
|
|
open_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
request_id=request.id,
|
|
)
|
|
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]),
|
|
)
|
|
]
|
|
),
|
|
session=self.session,
|
|
principal=self._principal(
|
|
"alice-account",
|
|
email="alice@example.test",
|
|
scopes={SCHEDULING_RESPOND_SCOPE},
|
|
),
|
|
)
|
|
alice = next(
|
|
participant
|
|
for participant in request.participants
|
|
if participant.email == "alice@example.test"
|
|
)
|
|
self.assertEqual(alice.status, "responded")
|
|
self.assertEqual(alice.participation_gateway, "scheduling")
|
|
self.assertIsNotNone(alice.poll_invitation_id)
|
|
self.assertEqual(
|
|
self.session.query(SchedulingNotification).filter(
|
|
SchedulingNotification.request_id == request.id,
|
|
SchedulingNotification.event_kind == "invitation",
|
|
).count(),
|
|
0,
|
|
)
|
|
|
|
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]
|
|
self.assertEqual(tokens, {})
|
|
token = self._issue_copy(request, first_participant)
|
|
|
|
submit_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=token,
|
|
payload=SchedulingPublicParticipationSubmitRequest(
|
|
answers=[
|
|
SchedulingAvailabilityAnswerInput(
|
|
slot_id=first_slot.id,
|
|
value="available",
|
|
option_revision=scheduling_slot_revision(first_slot),
|
|
),
|
|
SchedulingAvailabilityAnswerInput(
|
|
slot_id=second_slot.id,
|
|
value="maybe",
|
|
option_revision=scheduling_slot_revision(second_slot),
|
|
),
|
|
]
|
|
),
|
|
client_address="127.0.0.1",
|
|
)
|
|
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.assertEqual(len(all_jobs), 2)
|
|
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,
|
|
)
|
|
|
|
ordinary_writer = self._principal(
|
|
"unrelated-writer",
|
|
scopes={SCHEDULING_READ_SCOPE, SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
administrator = self._principal(
|
|
"scheduling-admin",
|
|
scopes={SCHEDULING_READ_SCOPE, SCHEDULING_ADMIN_SCOPE},
|
|
)
|
|
self.assertEqual(
|
|
[],
|
|
api_list_scheduling_requests(
|
|
status_filter=None,
|
|
session=self.session,
|
|
principal=ordinary_writer,
|
|
).requests,
|
|
)
|
|
self.assertEqual(
|
|
[request.id],
|
|
[
|
|
item.id
|
|
for item in api_list_scheduling_requests(
|
|
status_filter=None,
|
|
session=self.session,
|
|
principal=administrator,
|
|
).requests
|
|
],
|
|
)
|
|
|
|
def test_participant_list_and_get_redact_management_and_connector_internals(self) -> None:
|
|
request, _tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload(),
|
|
)
|
|
request.calendar_event_id = "calendar-event-internal"
|
|
request.metadata_ = {"connector_ref": "connector-internal"}
|
|
slot = request.slots[0]
|
|
slot.freebusy_status = "busy"
|
|
slot.freebusy_conflicts = [
|
|
{
|
|
"calendar_id": "calendar-1",
|
|
"event_id": "busy-event-internal",
|
|
"uid": "busy-uid-internal",
|
|
"recurrence_id": "busy-recurrence-internal",
|
|
"start_at": "2026-07-20T09:15:00+00:00",
|
|
"end_at": "2026-07-20T09:45:00+00:00",
|
|
"status": "CONFIRMED",
|
|
}
|
|
]
|
|
slot.tentative_hold_event_id = "hold-event-internal"
|
|
slot.metadata_ = {"provider_ref": "provider-internal"}
|
|
participant = request.participants[0]
|
|
participant.respondent_id = "alice-identity-internal"
|
|
participant.metadata_ = {"directory_ref": "directory-internal"}
|
|
self.session.flush()
|
|
|
|
principal = self._principal(
|
|
"alice-account",
|
|
email="alice@example.test",
|
|
scopes={SCHEDULING_READ_SCOPE},
|
|
)
|
|
listed = api_list_scheduling_requests(
|
|
status_filter=None,
|
|
session=self.session,
|
|
principal=principal,
|
|
).requests
|
|
fetched = api_get_scheduling_request(
|
|
request.id,
|
|
session=self.session,
|
|
principal=principal,
|
|
)
|
|
|
|
self.assertEqual(len(listed), 1)
|
|
for response in (listed[0], fetched):
|
|
self.assertIsNone(response.tenant_id)
|
|
self.assertIsNone(response.poll_id)
|
|
self.assertIsNone(response.organizer_user_id)
|
|
self.assertIsNone(response.calendar_integration_enabled)
|
|
self.assertIsNone(response.calendar_id)
|
|
self.assertIsNone(response.calendar_freebusy_enabled)
|
|
self.assertIsNone(response.calendar_hold_enabled)
|
|
self.assertIsNone(response.create_calendar_event_on_decision)
|
|
self.assertIsNone(response.calendar_event_id)
|
|
self.assertIsNone(
|
|
response.public_participation_policy_enforcement_available
|
|
)
|
|
self.assertEqual(response.metadata, {})
|
|
self.assertEqual(response.participant_aggregate.total, 2)
|
|
self.assertEqual(len(response.participants), 1)
|
|
own = response.participants[0]
|
|
self.assertTrue(own.is_current_participant)
|
|
self.assertEqual(own.email, "alice@example.test")
|
|
self.assertIsNone(own.respondent_id)
|
|
self.assertIsNone(own.poll_invitation_id)
|
|
self.assertEqual(own.metadata, {})
|
|
projected_slot = response.slots[0]
|
|
self.assertIsNone(projected_slot.poll_option_id)
|
|
self.assertEqual(projected_slot.freebusy_status, "busy")
|
|
self.assertEqual(
|
|
projected_slot.freebusy_conflicts,
|
|
[
|
|
{
|
|
"start_at": "2026-07-20T09:15:00+00:00",
|
|
"end_at": "2026-07-20T09:45:00+00:00",
|
|
"status": "CONFIRMED",
|
|
}
|
|
],
|
|
)
|
|
self.assertIsNone(projected_slot.tentative_hold_event_id)
|
|
self.assertEqual(projected_slot.metadata, {})
|
|
|
|
organizer = self._principal(
|
|
"user-1",
|
|
scopes={SCHEDULING_READ_SCOPE, SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
management = api_get_scheduling_request(
|
|
request.id,
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
self.assertEqual(management.tenant_id, "tenant-1")
|
|
self.assertEqual(management.poll_id, request.poll_id)
|
|
self.assertEqual(management.calendar_id, "calendar-1")
|
|
self.assertEqual(management.calendar_event_id, "calendar-event-internal")
|
|
self.assertEqual(management.metadata["connector_ref"], "connector-internal")
|
|
self.assertEqual(management.slots[0].poll_option_id, slot.poll_option_id)
|
|
self.assertEqual(
|
|
management.slots[0].freebusy_conflicts[0]["uid"],
|
|
"busy-uid-internal",
|
|
)
|
|
self.assertEqual(
|
|
management.slots[0].tentative_hold_event_id,
|
|
"hold-event-internal",
|
|
)
|
|
|
|
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_module_owned_poll_rejects_direct_response_and_uses_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]
|
|
self._issue_copy(request, target)
|
|
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",
|
|
},
|
|
)
|
|
|
|
with self.assertRaises(HTTPException) as direct_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.assertEqual(direct_response.exception.status_code, 404)
|
|
self.assertEqual([request.id], [item.id for item in listed.requests])
|
|
self.assertFalse(current.has_response)
|
|
|
|
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]),
|
|
)
|
|
]
|
|
),
|
|
session=self.session,
|
|
principal=attacker,
|
|
)
|
|
current = api_get_my_scheduling_availability(
|
|
request.id,
|
|
session=self.session,
|
|
principal=attacker,
|
|
)
|
|
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]
|
|
self.assertEqual(tokens, {})
|
|
token = self._issue_copy(request, participant)
|
|
submit_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=token,
|
|
payload=SchedulingPublicParticipationSubmitRequest(
|
|
answers=[
|
|
SchedulingAvailabilityAnswerInput(
|
|
slot_id=request.slots[0].id,
|
|
value="available",
|
|
option_revision=scheduling_slot_revision(request.slots[0]),
|
|
)
|
|
]
|
|
),
|
|
client_address="127.0.0.1",
|
|
)
|
|
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,
|
|
participant.respondent_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.assertTrue(own.is_current_participant)
|
|
self.assertIsNone(own.respondent_id)
|
|
self.assertEqual(own.email, "alice@example.test")
|
|
self.assertEqual(own.metadata, {})
|
|
self.assertIsNone(own.poll_invitation_id)
|
|
self.assertIsNone(own.invitation_token)
|
|
self.assertFalse(other.is_current_participant)
|
|
self.assertIsNone(other.respondent_id)
|
|
self.assertIsNone(other.email)
|
|
self.assertIsNone(other.participant_type)
|
|
self.assertIsNone(other.required)
|
|
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(
|
|
"user-1",
|
|
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(
|
|
"user-1",
|
|
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(
|
|
"user-1",
|
|
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_explicit_invitation_send_uses_signed_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, automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=payload,
|
|
)
|
|
self.assertEqual(provider.requests, [])
|
|
results = [
|
|
issue_scheduling_participant_invitation(
|
|
self.session,
|
|
tenant_id=request.tenant_id,
|
|
request_id=request.id,
|
|
participant_id=participant.id,
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
action="send",
|
|
)
|
|
for participant in request.participants
|
|
]
|
|
|
|
self.assertEqual(automatic_tokens, {})
|
|
self.assertTrue(all(result.action_url is None for result in results))
|
|
self.assertTrue(all(result.status == "queued" for result in results))
|
|
self.assertEqual(len(provider.requests), 2)
|
|
self.assertEqual({item.recipient_id for item in provider.requests}, {"alice-id", "bob-id"})
|
|
action_urls = {item.action_url for item in provider.requests}
|
|
self.assertTrue(
|
|
all(
|
|
isinstance(action_url, str)
|
|
and action_url.startswith(f"/scheduling/public/{request.id}/")
|
|
for action_url in action_urls
|
|
)
|
|
)
|
|
tokens = {
|
|
str(action_url).rsplit("/", 1)[-1]
|
|
for action_url in action_urls
|
|
}
|
|
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))
|
|
|
|
def test_send_without_delivery_provider_preserves_existing_invitation(self) -> None:
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
token = self._issue_copy(request, participant)
|
|
invitation_id = participant.poll_invitation_id
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_scheduling.backend.service.notification_dispatch_provider",
|
|
return_value=None,
|
|
),
|
|
self.assertRaisesRegex(
|
|
SchedulingError,
|
|
"Notification delivery is unavailable; copy the link instead",
|
|
),
|
|
):
|
|
issue_scheduling_participant_invitation(
|
|
self.session,
|
|
tenant_id=request.tenant_id,
|
|
request_id=request.id,
|
|
participant_id=participant.id,
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
action="send",
|
|
)
|
|
|
|
self.assertEqual(participant.poll_invitation_id, invitation_id)
|
|
invitation = self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == invitation_id
|
|
).one()
|
|
self.assertIsNone(invitation.revoked_at)
|
|
self.assertEqual(
|
|
get_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=token,
|
|
payload=SchedulingPublicParticipationAccessRequest(),
|
|
client_address="127.0.0.1",
|
|
)["request_id"],
|
|
request.id,
|
|
)
|
|
|
|
def test_send_without_delivery_target_preserves_existing_invitation(self) -> None:
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
token = self._issue_copy(request, participant)
|
|
invitation_id = participant.poll_invitation_id
|
|
participant.email = None
|
|
self.session.flush()
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_scheduling.backend.service.notification_dispatch_provider",
|
|
return_value=object(),
|
|
),
|
|
self.assertRaisesRegex(
|
|
SchedulingError,
|
|
"Participant has no deliverable email address or account",
|
|
),
|
|
):
|
|
issue_scheduling_participant_invitation(
|
|
self.session,
|
|
tenant_id=request.tenant_id,
|
|
request_id=request.id,
|
|
participant_id=participant.id,
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
action="send",
|
|
)
|
|
|
|
self.assertEqual(participant.poll_invitation_id, invitation_id)
|
|
invitation = self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == invitation_id
|
|
).one()
|
|
self.assertIsNone(invitation.revoked_at)
|
|
self.assertEqual(
|
|
get_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=token,
|
|
payload=SchedulingPublicParticipationAccessRequest(),
|
|
client_address="127.0.0.1",
|
|
)["request_id"],
|
|
request.id,
|
|
)
|
|
|
|
def test_stale_copy_is_rejected_before_rotating_invitation(self) -> None:
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
stale_revision = scheduling_participant_revision(participant)
|
|
self._issue_copy(request, participant)
|
|
invitation_id = participant.poll_invitation_id
|
|
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
|
|
|
|
with (
|
|
patch("govoplan_scheduling.backend.router.audit_event") as audit,
|
|
self.assertRaises(HTTPException) as conflict,
|
|
):
|
|
api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="copy",
|
|
participant_revision=stale_revision,
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
|
|
self.assertEqual(conflict.exception.status_code, 409)
|
|
self.assertEqual(participant.poll_invitation_id, invitation_id)
|
|
invitation = self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == invitation_id
|
|
).one()
|
|
self.assertIsNone(invitation.revoked_at)
|
|
audit.assert_not_called()
|
|
|
|
def test_stale_send_is_rejected_before_delivery_or_rotation(self) -> None:
|
|
class CapturingNotificationProvider:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def enqueue_notification(self, _session, request, *, enqueue_delivery):
|
|
self.requests.append(request)
|
|
return {"id": "notification-1", "status": "queued"}
|
|
|
|
provider = CapturingNotificationProvider()
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
stale_revision = scheduling_participant_revision(participant)
|
|
self._issue_copy(request, participant)
|
|
invitation_id = participant.poll_invitation_id
|
|
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_scheduling.backend.service.notification_dispatch_provider",
|
|
return_value=provider,
|
|
),
|
|
patch("govoplan_scheduling.backend.router.audit_event") as audit,
|
|
self.assertRaises(HTTPException) as conflict,
|
|
):
|
|
api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="send",
|
|
participant_revision=stale_revision,
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
|
|
self.assertEqual(conflict.exception.status_code, 409)
|
|
self.assertEqual(participant.poll_invitation_id, invitation_id)
|
|
invitation = self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == invitation_id
|
|
).one()
|
|
self.assertIsNone(invitation.revoked_at)
|
|
self.assertEqual(provider.requests, [])
|
|
self.assertEqual(
|
|
self.session.query(SchedulingNotification).filter(
|
|
SchedulingNotification.request_id == request.id
|
|
).count(),
|
|
0,
|
|
)
|
|
audit.assert_not_called()
|
|
|
|
def test_stale_revoke_is_rejected_before_changing_access(self) -> None:
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
stale_revision = scheduling_participant_revision(participant)
|
|
self._issue_copy(request, participant)
|
|
invitation_id = participant.poll_invitation_id
|
|
organizer = self._principal("user-1", scopes={SCHEDULING_WRITE_SCOPE})
|
|
|
|
with (
|
|
patch("govoplan_scheduling.backend.router.audit_event") as audit,
|
|
self.assertRaises(HTTPException) as conflict,
|
|
):
|
|
api_revoke_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationRevokeRequest(
|
|
participant_revision=stale_revision,
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
|
|
self.assertEqual(conflict.exception.status_code, 409)
|
|
self.assertEqual(participant.poll_invitation_id, invitation_id)
|
|
invitation = self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == invitation_id
|
|
).one()
|
|
self.assertIsNone(invitation.revoked_at)
|
|
audit.assert_not_called()
|
|
|
|
def test_invitation_router_rotates_revokes_and_enforces_management_policy(self) -> None:
|
|
request, automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
organizer = self._principal(
|
|
"user-1",
|
|
scopes={SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
response_headers = Response()
|
|
with patch("govoplan_scheduling.backend.router.audit_event") as audit:
|
|
first = api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="copy",
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
response_headers,
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
|
|
self.assertEqual(automatic_tokens, {})
|
|
self.assertEqual(first.action, "copy")
|
|
self.assertEqual(first.status, "issued")
|
|
self.assertIsNotNone(first.action_url)
|
|
self.assertEqual(
|
|
response_headers.headers["cache-control"],
|
|
"no-store, private",
|
|
)
|
|
first_token = str(first.action_url).rsplit("/", 1)[-1]
|
|
self.assertNotIn(first_token, repr([call.kwargs for call in audit.call_args_list]))
|
|
|
|
unrelated_writer = self._principal(
|
|
"unrelated",
|
|
scopes={SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
with self.assertRaises(HTTPException) as denied:
|
|
api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="copy",
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=unrelated_writer,
|
|
)
|
|
self.assertEqual(denied.exception.status_code, 403)
|
|
self.assertEqual(
|
|
get_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=first_token,
|
|
payload=SchedulingPublicParticipationAccessRequest(),
|
|
client_address="127.0.0.1",
|
|
)["request_id"],
|
|
request.id,
|
|
)
|
|
|
|
administrator = self._principal(
|
|
"administrator",
|
|
scopes={SCHEDULING_ADMIN_SCOPE},
|
|
)
|
|
with patch("govoplan_scheduling.backend.router.audit_event") as audit:
|
|
rotated = api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="copy",
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=administrator,
|
|
)
|
|
rotated_token = str(rotated.action_url).rsplit("/", 1)[-1]
|
|
self.assertNotEqual(rotated_token, first_token)
|
|
self.assertNotIn(rotated_token, repr([call.kwargs for call in audit.call_args_list]))
|
|
with self.assertRaisesRegex(
|
|
Exception,
|
|
"Scheduling participation link or credentials are invalid",
|
|
):
|
|
get_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=first_token,
|
|
payload=SchedulingPublicParticipationAccessRequest(),
|
|
client_address="127.0.0.1",
|
|
)
|
|
|
|
with patch("govoplan_scheduling.backend.router.audit_event") as audit:
|
|
revoked = api_revoke_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationRevokeRequest(
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
replayed = api_revoke_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationRevokeRequest(
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
self.assertFalse(revoked.replayed)
|
|
self.assertTrue(replayed.replayed)
|
|
self.assertNotIn(rotated_token, repr([call.kwargs for call in audit.call_args_list]))
|
|
with self.assertRaisesRegex(
|
|
Exception,
|
|
"Scheduling participation link or credentials are invalid",
|
|
):
|
|
get_public_scheduling_participation(
|
|
self.session,
|
|
request_id=request.id,
|
|
token=rotated_token,
|
|
payload=SchedulingPublicParticipationAccessRequest(),
|
|
client_address="127.0.0.1",
|
|
)
|
|
|
|
def test_links_can_be_issued_for_draft_and_closed_requests(self) -> None:
|
|
organizer = self._principal(
|
|
"user-1",
|
|
scopes={SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
for request_status in ("draft", "collecting"):
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={
|
|
"status": request_status,
|
|
"calendar": SchedulingCalendarPreferences(),
|
|
}
|
|
),
|
|
)
|
|
if request_status == "collecting":
|
|
close_scheduling_request(
|
|
self.session,
|
|
tenant_id=request.tenant_id,
|
|
request_id=request.id,
|
|
)
|
|
expected_status = "closed"
|
|
else:
|
|
expected_status = "draft"
|
|
with patch("govoplan_scheduling.backend.router.audit_event"):
|
|
issued = api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
request.participants[0].id,
|
|
SchedulingInvitationActionRequest(
|
|
action="copy",
|
|
participant_revision=scheduling_participant_revision(
|
|
request.participants[0]
|
|
),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
self.assertEqual(request.status, expected_status)
|
|
self.assertTrue(
|
|
str(issued.action_url).startswith(
|
|
f"/scheduling/public/{request.id}/"
|
|
)
|
|
)
|
|
|
|
def test_send_response_audit_error_and_durable_job_never_store_token(self) -> None:
|
|
class EchoingFailureNotificationProvider:
|
|
def __init__(self) -> None:
|
|
self.requests = []
|
|
|
|
def enqueue_notification(self, _session, request, *, enqueue_delivery):
|
|
self.requests.append(request)
|
|
raise RuntimeError(f"failed dispatch for {request.action_url}")
|
|
|
|
provider = EchoingFailureNotificationProvider()
|
|
request, _automatic_tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
payload=self._payload().model_copy(
|
|
update={"calendar": SchedulingCalendarPreferences()}
|
|
),
|
|
)
|
|
participant = request.participants[0]
|
|
organizer = self._principal(
|
|
"user-1",
|
|
scopes={SCHEDULING_WRITE_SCOPE},
|
|
)
|
|
with (
|
|
patch(
|
|
"govoplan_scheduling.backend.service.notification_dispatch_provider",
|
|
return_value=provider,
|
|
),
|
|
patch("govoplan_scheduling.backend.router.audit_event") as audit,
|
|
):
|
|
result = api_issue_scheduling_participant_invitation(
|
|
request.id,
|
|
participant.id,
|
|
SchedulingInvitationActionRequest(
|
|
action="send",
|
|
participant_revision=scheduling_participant_revision(participant),
|
|
),
|
|
Response(),
|
|
session=self.session,
|
|
principal=organizer,
|
|
)
|
|
|
|
self.assertEqual(len(provider.requests), 1)
|
|
action_url = provider.requests[0].action_url
|
|
self.assertIsInstance(action_url, str)
|
|
token = str(action_url).rsplit("/", 1)[-1]
|
|
self.assertIsNone(result.action_url)
|
|
self.assertEqual(result.status, "failed")
|
|
self.assertEqual(result.notification.error, "Notification center enqueue failed")
|
|
self.assertNotIn(token, repr(result.model_dump()))
|
|
self.assertNotIn(token, repr([call.kwargs for call in audit.call_args_list]))
|
|
notification = self.session.query(SchedulingNotification).filter(
|
|
SchedulingNotification.id == result.notification.id
|
|
).one()
|
|
durable_projection = repr(
|
|
{
|
|
"payload": notification.payload,
|
|
"metadata": notification.metadata_,
|
|
"error": notification.error,
|
|
}
|
|
)
|
|
self.assertNotIn(token, durable_projection)
|
|
|
|
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()
|