255 lines
9.3 KiB
Python
255 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
|
|
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.modules import ModuleContext
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
|
|
from govoplan_poll.backend.manifest import get_manifest as get_poll_manifest
|
|
from govoplan_scheduling.backend.db.models import (
|
|
SchedulingCandidateSlot,
|
|
SchedulingNotification,
|
|
SchedulingParticipant,
|
|
SchedulingRequest,
|
|
)
|
|
from govoplan_scheduling.backend.manifest import RESPOND_SCOPE, WRITE_SCOPE
|
|
from govoplan_scheduling.backend.router import (
|
|
api_get_my_scheduling_availability,
|
|
api_submit_scheduling_availability,
|
|
api_update_scheduling_candidate_slot,
|
|
)
|
|
from govoplan_scheduling.backend.runtime import configure_runtime
|
|
from govoplan_scheduling.backend.schemas import (
|
|
SchedulingAvailabilityAnswerInput,
|
|
SchedulingAvailabilityResponseRequest,
|
|
SchedulingCalendarPreferences,
|
|
SchedulingCandidateSlotInput,
|
|
SchedulingCandidateSlotUpdateRequest,
|
|
SchedulingParticipantInput,
|
|
SchedulingRequestCreateRequest,
|
|
)
|
|
from govoplan_scheduling.backend.service import (
|
|
SchedulingError,
|
|
cancel_scheduling_request,
|
|
create_scheduling_request,
|
|
update_scheduling_candidate_slot,
|
|
)
|
|
|
|
|
|
class SchedulingResponseEditingTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(get_poll_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__,
|
|
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=[
|
|
SchedulingNotification.__table__,
|
|
SchedulingParticipant.__table__,
|
|
SchedulingCandidateSlot.__table__,
|
|
SchedulingRequest.__table__,
|
|
PollInvitation.__table__,
|
|
PollResponse.__table__,
|
|
PollOption.__table__,
|
|
Poll.__table__,
|
|
],
|
|
)
|
|
self.engine.dispose()
|
|
|
|
@staticmethod
|
|
def _principal(account_id: str, *, email: str | None, scopes: set[str]) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id=f"membership-{account_id}",
|
|
tenant_id="tenant-1",
|
|
email=email,
|
|
display_name=account_id,
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id=account_id),
|
|
user=SimpleNamespace(id=f"membership-{account_id}"),
|
|
)
|
|
|
|
def _request(self, *, status: str = "collecting") -> SchedulingRequest:
|
|
start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc)
|
|
request, _tokens = create_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="organizer-1",
|
|
payload=SchedulingRequestCreateRequest(
|
|
title="Response editing",
|
|
status=status,
|
|
calendar=SchedulingCalendarPreferences(),
|
|
slots=[
|
|
SchedulingCandidateSlotInput(
|
|
label="Monday",
|
|
start_at=start,
|
|
end_at=start + timedelta(hours=1),
|
|
),
|
|
SchedulingCandidateSlotInput(
|
|
label="Tuesday",
|
|
start_at=start + timedelta(days=1),
|
|
end_at=start + timedelta(days=1, hours=1),
|
|
),
|
|
],
|
|
participants=[
|
|
SchedulingParticipantInput(
|
|
display_name="Alice",
|
|
email="alice@example.test",
|
|
)
|
|
],
|
|
),
|
|
)
|
|
return request
|
|
|
|
def _submit_both(self, request: SchedulingRequest) -> None:
|
|
api_submit_scheduling_availability(
|
|
request.id,
|
|
SchedulingAvailabilityResponseRequest(
|
|
answers=[
|
|
SchedulingAvailabilityAnswerInput(slot_id=request.slots[0].id, value="available"),
|
|
SchedulingAvailabilityAnswerInput(slot_id=request.slots[1].id, value="maybe"),
|
|
]
|
|
),
|
|
session=self.session,
|
|
principal=self._principal(
|
|
"alice-account",
|
|
email="alice@example.test",
|
|
scopes={RESPOND_SCOPE},
|
|
),
|
|
)
|
|
|
|
def test_current_participant_response_is_prefilled_and_changed_slot_is_invalidated(self) -> None:
|
|
request = self._request()
|
|
self._submit_both(request)
|
|
alice = self._principal(
|
|
"alice-account",
|
|
email="alice@example.test",
|
|
scopes={RESPOND_SCOPE},
|
|
)
|
|
|
|
before = api_get_my_scheduling_availability(
|
|
request.id,
|
|
session=self.session,
|
|
principal=alice,
|
|
)
|
|
|
|
self.assertTrue(before.has_response)
|
|
self.assertIsNotNone(before.submitted_at)
|
|
self.assertEqual(before.participant_id, request.participants[0].id)
|
|
self.assertEqual(
|
|
[(answer.slot_id, answer.value) for answer in before.answers],
|
|
[(request.slots[0].id, "available"), (request.slots[1].id, "maybe")],
|
|
)
|
|
|
|
api_update_scheduling_candidate_slot(
|
|
request.id,
|
|
request.slots[0].id,
|
|
SchedulingCandidateSlotUpdateRequest(label="Monday, revised"),
|
|
session=self.session,
|
|
principal=self._principal("organizer-1", email=None, scopes={WRITE_SCOPE}),
|
|
)
|
|
after = api_get_my_scheduling_availability(
|
|
request.id,
|
|
session=self.session,
|
|
principal=alice,
|
|
)
|
|
|
|
self.assertTrue(after.has_response)
|
|
self.assertEqual(
|
|
[(answer.slot_id, answer.value) for answer in after.answers],
|
|
[(request.slots[1].id, "maybe")],
|
|
)
|
|
poll_response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one()
|
|
self.assertEqual(
|
|
[answer["option_id"] for answer in poll_response.answers],
|
|
[request.slots[1].poll_option_id],
|
|
)
|
|
|
|
# Re-submit and repeat the exact option snapshot: an idempotent no-op
|
|
# must preserve both current answers.
|
|
self._submit_both(request)
|
|
api_update_scheduling_candidate_slot(
|
|
request.id,
|
|
request.slots[0].id,
|
|
SchedulingCandidateSlotUpdateRequest(label="Monday, revised"),
|
|
session=self.session,
|
|
principal=self._principal("organizer-1", email=None, scopes={WRITE_SCOPE}),
|
|
)
|
|
unchanged = api_get_my_scheduling_availability(
|
|
request.id,
|
|
session=self.session,
|
|
principal=alice,
|
|
)
|
|
self.assertEqual(len(unchanged.answers), 2)
|
|
|
|
def test_calendar_hold_rejection_leaves_slot_and_answers_unchanged(self) -> None:
|
|
request = self._request()
|
|
self._submit_both(request)
|
|
slot = request.slots[0]
|
|
original_start = slot.start_at
|
|
slot.tentative_hold_event_id = "event-1"
|
|
self.session.flush()
|
|
|
|
with self.assertRaisesRegex(SchedulingError, "tentative calendar hold"):
|
|
update_scheduling_candidate_slot(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
request_id=request.id,
|
|
slot_id=slot.id,
|
|
payload=SchedulingCandidateSlotUpdateRequest(
|
|
start_at=original_start + timedelta(minutes=15),
|
|
),
|
|
)
|
|
|
|
self.assertEqual(slot.start_at, original_start)
|
|
response = self.session.query(PollResponse).filter(PollResponse.poll_id == request.poll_id).one()
|
|
self.assertEqual(len(response.answers), 2)
|
|
|
|
def test_cancelling_draft_request_preserves_draft_poll(self) -> None:
|
|
request = self._request(status="draft")
|
|
|
|
cancelled = cancel_scheduling_request(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
request_id=request.id,
|
|
)
|
|
poll = self.session.query(Poll).filter(Poll.id == request.poll_id).one()
|
|
|
|
self.assertEqual(cancelled.status, "cancelled")
|
|
self.assertIsNotNone(cancelled.cancelled_at)
|
|
self.assertEqual(poll.status, "draft")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|