329 lines
12 KiB
Python
329 lines
12 KiB
Python
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()
|