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

feat: add governed public self-enrollment links

Browse Source
This commit is contained in:
zemion
2026-08-20 13:03:57 +02:00
parent 79cfaa951a
commit 539e3cbb5e
18 changed files with 2481 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_manifest.py
+8 -1
View File
@@ -33,7 +33,10 @@ class SchedulingManifestTests(unittest.TestCase):
self.assertIsNotNone(manifest.migration_spec)
self.assertIsNotNone(manifest.frontend)
self.assertEqual(
["/scheduling/public/:requestId/:token"],
[
"/scheduling/public/:requestId/:token",
"/scheduling/enrol/:requestId/:token",
],
[route.path for route in manifest.frontend.public_routes],
)
self.assertIn("poll.availability_matrix", {interface.name for interface in manifest.requires_interfaces})
@@ -63,6 +66,10 @@ class SchedulingManifestTests(unittest.TestCase):
"scheduling.calendar-coordination",
documentation["scheduling.calendar-coordination"].metadata["help_contexts"],
)
self.assertIn(
"scheduling.public-self-enrollment",
documentation["scheduling.public-self-enrollment"].metadata["help_contexts"],
)
if __name__ == "__main__":
tests/test_migrations.py
+9 -1
View File
@@ -20,7 +20,7 @@ from govoplan_scheduling.backend.db.models import (
from govoplan_scheduling.backend.manifest import get_manifest as get_scheduling_manifest
_SCHEDULING_HEAD = "c9d4e7f1a2b3"
_SCHEDULING_HEAD = "d7a4c1e8f205"
_SCHEDULING_RESPONSE_SETTINGS_REVISION = "ad7e3c9b2f10"
_ENABLED_MODULES = ("poll", "scheduling")
_MANIFEST_FACTORIES = (get_poll_manifest, get_scheduling_manifest)
@@ -150,6 +150,14 @@ class SchedulingMigrationTests(unittest.TestCase):
self.assertIn("max_participants_per_option", columns)
self.assertIn("response_comment", participant_columns)
self.assertIn("participation_gateway", participant_columns)
self.assertIn("self_enrollment_link_id", participant_columns)
self.assertIn("self_enrollment_proof_hash", participant_columns)
self.assertIn("bound_account_id", participant_columns)
self.assertIn("account_bound_at", participant_columns)
self.assertIn(
"scheduling_public_enrollment_links",
inspect(engine).get_table_names(),
)
self.assertEqual(visibility, "aggregates_only")
self.assertEqual(tuple(response_defaults), (1, 0, None, 1, 0, 0, 0, None))
self.assertEqual(set(counts.values()), {1})
tests/test_self_enrollment.py
+328
View File
@@ -0,0 +1,328 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from pydantic import SecretStr
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,
PollLifecycleTransition,
PollOption,
PollParticipationSubmission,
PollResponse,
)
from govoplan_poll.backend.manifest import get_manifest as get_poll_manifest
from govoplan_scheduling.backend.db.models import (
SchedulingCandidateSlot,
SchedulingNotification,
SchedulingParticipant,
SchedulingPublicEnrollmentLink,
SchedulingRequest,
)
from govoplan_scheduling.backend.schemas import (
SchedulingAuthenticatedEnrollmentSubmitRequest,
SchedulingAvailabilityAnswerInput,
SchedulingCandidateSlotInput,
SchedulingEnrollmentLinkCreateRequest,
SchedulingPublicEnrollmentAccessRequest,
SchedulingPublicEnrollmentSubmitRequest,
SchedulingRequestCreateRequest,
)
from govoplan_scheduling.backend.runtime import configure_runtime
from govoplan_scheduling.backend.service import (
SchedulingConflictError,
SchedulingPublicParticipationError,
create_scheduling_enrollment_link,
create_scheduling_request,
get_public_scheduling_enrollment,
revoke_scheduling_enrollment_link,
scheduling_request_is_visible,
scheduling_slot_revision,
submit_authenticated_scheduling_enrollment,
submit_public_scheduling_enrollment,
)
class SchedulingSelfEnrollmentTests(unittest.TestCase):
def setUp(self) -> None:
self.now = datetime(2026, 8, 20, 10, tzinfo=timezone.utc)
self.patches = (
patch("govoplan_scheduling.backend.service._now", return_value=self.now),
patch("govoplan_poll.backend.service._now", return_value=self.now),
patch("govoplan_poll.backend.participation_service._now", return_value=self.now),
)
for item in self.patches:
item.start()
registry = PlatformRegistry()
registry.register(get_poll_manifest())
settings = SimpleNamespace(
redis_url=None,
scheduling_public_self_enrollment_enabled=True,
scheduling_public_self_enrollment_max_capacity=100,
)
registry.configure_capability_context(
ModuleContext(registry=registry, settings=settings)
)
configure_runtime(registry=registry, settings=settings)
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__,
SchedulingRequest.__table__,
SchedulingPublicEnrollmentLink.__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)
self.engine.dispose()
for item in reversed(self.patches):
item.stop()
def _request(self, *, email_required: bool = False) -> SchedulingRequest:
start = self.now + timedelta(days=1)
request, _tokens = create_scheduling_request(
self.session,
tenant_id="tenant-1",
user_id="organizer-1",
payload=SchedulingRequestCreateRequest(
title="Public planning",
status="collecting",
deadline_at=self.now + timedelta(days=4),
participant_email_required=email_required,
slots=[
SchedulingCandidateSlotInput(
label="First option",
start_at=start,
end_at=start + timedelta(hours=1),
)
],
),
)
return request
def _link(
self,
request: SchedulingRequest,
*,
capacity: int = 2,
) -> tuple[SchedulingPublicEnrollmentLink, str]:
return create_scheduling_enrollment_link(
self.session,
tenant_id=request.tenant_id,
request_id=request.id,
created_by="organizer-1",
payload=SchedulingEnrollmentLinkCreateRequest(
expires_at=self.now + timedelta(days=2),
max_enrollments=capacity,
),
)
@staticmethod
def _answer(request: SchedulingRequest) -> SchedulingAvailabilityAnswerInput:
slot = request.slots[0]
return SchedulingAvailabilityAnswerInput(
slot_id=slot.id,
value="available",
option_revision=scheduling_slot_revision(slot),
)
def test_anonymous_enrollment_requires_proof_for_updates_and_is_idempotent(self) -> None:
request = self._request(email_required=True)
link, token = self._link(request)
opened = get_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=SchedulingPublicEnrollmentAccessRequest(),
client_address="192.0.2.1",
)
self.assertTrue(opened["participant_email_required"])
self.assertEqual(opened["enrollment_remaining"], 2)
payload = SchedulingPublicEnrollmentSubmitRequest(
display_name="Ada Example",
email="ADA@example.test",
participant_proof="proof-" + "a" * 40,
idempotency_key="enrollment-submit-1",
answers=[self._answer(request)],
)
created = submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=payload,
client_address="192.0.2.1",
)
replayed = submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=payload,
client_address="192.0.2.1",
)
self.assertTrue(created["enrolled"])
self.assertTrue(created["has_response"])
self.assertTrue(replayed["replayed"])
self.assertEqual(self.session.query(SchedulingParticipant).count(), 1)
participant = self.session.query(SchedulingParticipant).one()
self.assertEqual(participant.email, "ada@example.test")
self.assertNotEqual(participant.self_enrollment_proof_hash, payload.participant_proof.get_secret_value())
self.assertNotIn(token, repr(participant.metadata_))
with self.assertRaises(SchedulingPublicParticipationError):
submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=payload.model_copy(
update={
"participant_proof": SecretStr("proof-" + "b" * 40),
"idempotency_key": "enrollment-submit-2",
}
),
client_address="192.0.2.2",
)
def test_capacity_is_serialized_but_existing_proof_can_update(self) -> None:
request = self._request()
_link, token = self._link(request, capacity=1)
first = SchedulingPublicEnrollmentSubmitRequest(
display_name="Ada",
participant_proof="proof-" + "a" * 40,
idempotency_key="first",
answers=[self._answer(request)],
)
submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=first,
client_address="192.0.2.3",
)
updated = submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=first.model_copy(update={"idempotency_key": "update"}),
client_address="192.0.2.3",
)
self.assertTrue(updated["has_response"])
with self.assertRaises(SchedulingConflictError):
submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=first.model_copy(
update={
"display_name": "Grace",
"participant_proof": SecretStr("proof-" + "g" * 40),
"idempotency_key": "second",
}
),
client_address="192.0.2.4",
)
def test_signed_in_binding_requires_confirmation_and_can_claim_proof(self) -> None:
request = self._request()
link, token = self._link(request)
anonymous = SchedulingPublicEnrollmentSubmitRequest(
display_name="Ada",
participant_proof="proof-" + "a" * 40,
idempotency_key="anonymous",
answers=[self._answer(request)],
)
submit_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=anonymous,
client_address="192.0.2.5",
)
authenticated = SchedulingAuthenticatedEnrollmentSubmitRequest(
display_name="Ada",
bind_account_confirmed=True,
participant_proof=anonymous.participant_proof,
idempotency_key="bound",
answers=[self._answer(request)],
)
bound = submit_authenticated_scheduling_enrollment(
self.session,
tenant_id="tenant-1",
request_id=request.id,
token=token,
account_id="account-ada",
account_email="ada@example.test",
payload=authenticated,
client_address="192.0.2.5",
)
participant = self.session.query(SchedulingParticipant).one()
self.assertTrue(bound["account_bound"])
self.assertEqual(participant.bound_account_id, "account-ada")
self.assertEqual(participant.self_enrollment_link_id, link.id)
self.assertTrue(
scheduling_request_is_visible(
request,
actor_ids=("account-ada",),
)
)
with self.assertRaisesRegex(ValueError, "Account binding must be confirmed"):
submit_authenticated_scheduling_enrollment(
self.session,
tenant_id="tenant-1",
request_id=request.id,
token=token,
account_id="other-account",
account_email=None,
payload=authenticated.model_copy(update={"bind_account_confirmed": False}),
client_address="192.0.2.6",
)
def test_revocation_invalidates_link_without_storing_raw_token(self) -> None:
request = self._request()
link, token = self._link(request)
self.assertNotEqual(link.token_hash, token)
revoked, replayed = revoke_scheduling_enrollment_link(
self.session,
tenant_id=request.tenant_id,
request_id=request.id,
link_id=link.id,
)
self.assertFalse(replayed)
self.assertIsNotNone(revoked.revoked_at)
with self.assertRaises(SchedulingPublicParticipationError):
get_public_scheduling_enrollment(
self.session,
request_id=request.id,
token=token,
payload=SchedulingPublicEnrollmentAccessRequest(),
client_address="192.0.2.7",
)
if __name__ == "__main__":
unittest.main()