GovOPlaN/govoplan-scheduling
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scheduling): add explicit invitation link lifecycle

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 03:08:56 +02:00
parent ed828685f6
commit 2cb86c90dc
6 changed files with 785 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
tests/test_response_editing.py
View File

@@ -60,6 +60,7 @@ from govoplan_scheduling.backend.service import (
cancel_scheduling_request,
create_scheduling_request,
get_public_scheduling_participation,
issue_scheduling_participant_invitation,
scheduling_request_summary,
scheduling_slot_revision,
submit_scheduling_availability,
@@ -149,8 +150,9 @@ class SchedulingResponseEditingTests(unittest.TestCase):
participants: list[SchedulingParticipantInput] | None = None,
**settings,
) -> tuple[SchedulingRequest, dict[str, str]]:
issue_links = bool(settings.pop("create_participant_invitations", True))
start = datetime(2026, 7, 20, 9, tzinfo=timezone.utc)
return create_scheduling_request(
request, automatic_tokens = create_scheduling_request(
self.session,
tenant_id="tenant-1",
user_id="organizer-1",
@@ -184,6 +186,31 @@ class SchedulingResponseEditingTests(unittest.TestCase):
**settings,
),
)
self.assertEqual(automatic_tokens, {})
tokens = (
{
participant.id: self._issue_copy(request, participant)
for participant in request.participants
}
if issue_links
else {}
)
return request, tokens
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,
action="copy",
)
self.assertIsNotNone(result.action_url)
return str(result.action_url).rsplit("/", 1)[-1]
def _answer(
self,
@@ -475,7 +502,8 @@ class SchedulingResponseEditingTests(unittest.TestCase):
],
),
)
token = tokens[public_request.participants[0].id]
self.assertEqual(tokens, {})
token = self._issue_copy(public_request, public_request.participants[0])
invitation = self.session.query(PollInvitation).filter(
PollInvitation.id == public_request.participants[0].poll_invitation_id
).one()
@@ -568,7 +596,8 @@ class SchedulingResponseEditingTests(unittest.TestCase):
participants=[SchedulingParticipantInput(display_name="Guest")],
),
)
token = tokens[request.participants[0].id]
self.assertEqual(tokens, {})
token = self._issue_copy(request, request.participants[0])
wrong = SchedulingPublicParticipationAccessRequest(password="wrong password")
for attempt in range(10):
@@ -623,7 +652,8 @@ class SchedulingResponseEditingTests(unittest.TestCase):
),
)
participant = request.participants[0]
token = tokens[participant.id]
self.assertEqual(tokens, {})
token = self._issue_copy(request, participant)
answer = SchedulingAvailabilityAnswerInput(
slot_id=request.slots[0].id,
value="available",
@@ -1036,8 +1066,16 @@ class SchedulingResponseEditingTests(unittest.TestCase):
self.assertIsNotNone(removed_slot.deleted_at)
bob = next(item for item in response.participants if item.display_name == "Bob")
self.assertEqual(bob.email, "bob@example.test")
self.assertIsNotNone(bob.poll_invitation_id)
self.assertIsNotNone(bob.invitation_token)
self.assertIsNone(bob.poll_invitation_id)
self.assertIsNone(bob.invitation_token)
bob_model = next(
item
for item in request.participants
if item.deleted_at is None and item.display_name == "Bob"
)
self._issue_copy(request, bob_model)
bob_invitation_id = bob_model.poll_invitation_id
self.assertIsNotNone(bob_invitation_id)
current = api_get_my_scheduling_availability(
request.id,
session=self.session,
@@ -1091,7 +1129,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
),
)
invitation = self.session.query(PollInvitation).filter(
PollInvitation.id == bob.poll_invitation_id
PollInvitation.id == bob_invitation_id
).one()
self.assertIsNotNone(invitation.revoked_at)

312
tests/test_service.py
View File

@@ -5,7 +5,7 @@ from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import patch
from fastapi import HTTPException
from fastapi import HTTPException, Response
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
@@ -42,7 +42,9 @@ from govoplan_scheduling.backend.schemas import (
SchedulingCalendarPreferences,
SchedulingCandidateSlotInput,
SchedulingDecisionRequest,
SchedulingInvitationActionRequest,
SchedulingParticipantInput,
SchedulingPublicParticipationAccessRequest,
SchedulingPublicParticipationSubmitRequest,
SchedulingRequestCreateRequest,
SchedulingRequestUpdateRequest,
@@ -55,9 +57,11 @@ from govoplan_scheduling.backend.router import (
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,
@@ -70,6 +74,8 @@ from govoplan_scheduling.backend.service import (
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,
@@ -206,7 +212,22 @@ class SchedulingServiceTests(unittest.TestCase):
],
)
def test_create_request_creates_poll_slots_and_signed_invitations(self) -> None:
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,
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",
@@ -224,8 +245,20 @@ class SchedulingServiceTests(unittest.TestCase):
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))
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:
@@ -331,11 +364,13 @@ class SchedulingServiceTests(unittest.TestCase):
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=tokens[first_participant.id],
token=token,
payload=SchedulingPublicParticipationSubmitRequest(
answers=[
SchedulingAvailabilityAnswerInput(
@@ -584,7 +619,7 @@ class SchedulingServiceTests(unittest.TestCase):
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.assertEqual(len(all_jobs), 2)
self.assertTrue(all(job.status == "pending" for job in reminder_jobs))
organizer_jobs = list_visible_scheduling_notifications(
@@ -1057,6 +1092,7 @@ class SchedulingServiceTests(unittest.TestCase):
payload=payload,
)
target = request.participants[1]
self._issue_copy(request, target)
attacker = self._principal(
"attacker",
email="alice@example.test",
@@ -1095,7 +1131,7 @@ class SchedulingServiceTests(unittest.TestCase):
principal=attacker,
)
self.assertEqual(direct_response.exception.status_code, 400)
self.assertEqual(direct_response.exception.status_code, 404)
self.assertEqual([request.id], [item.id for item in listed.requests])
self.assertFalse(current.has_response)
@@ -1225,10 +1261,12 @@ class SchedulingServiceTests(unittest.TestCase):
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=tokens[participant.id],
token=token,
payload=SchedulingPublicParticipationSubmitRequest(
answers=[
SchedulingAvailabilityAnswerInput(
@@ -1436,7 +1474,7 @@ class SchedulingServiceTests(unittest.TestCase):
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:
def test_explicit_invitation_send_uses_signed_link_and_verified_recipient_id(self) -> None:
class CapturingNotificationProvider:
def __init__(self) -> None:
self.requests = []
@@ -1466,22 +1504,41 @@ class SchedulingServiceTests(unittest.TestCase):
"govoplan_scheduling.backend.service.notification_dispatch_provider",
return_value=provider,
):
request, tokens = create_scheduling_request(
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,
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"})
self.assertEqual(
{item.action_url for item in provider.requests},
{
f"/scheduling/public/{request.id}/{token}"
for token in tokens.values()
},
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",
@@ -1489,7 +1546,228 @@ class SchedulingServiceTests(unittest.TestCase):
)
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()))
self.assertTrue(all(token not in serialized for token in tokens))
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"),
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"),
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"),
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,
Response(),
session=self.session,
principal=organizer,
)
replayed = api_revoke_scheduling_participant_invitation(
request.id,
participant.id,
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"),
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"),
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})