1159 lines
44 KiB
Python
1159 lines
44 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
from fastapi import HTTPException
|
|
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.poll import (
|
|
PollAnswerRequest,
|
|
PollCapabilityError,
|
|
PollCreateCommand,
|
|
PollOptionRequest,
|
|
PollSubmitResponseCommand,
|
|
PollUpdateCommand,
|
|
)
|
|
from govoplan_core.db.base import Base, utcnow
|
|
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
|
|
from govoplan_poll.backend.db.models import (
|
|
Poll,
|
|
PollInvitation,
|
|
PollLifecycleTransition,
|
|
PollOption,
|
|
PollParticipationSubmission,
|
|
PollResponse,
|
|
)
|
|
from govoplan_poll.backend.participation import (
|
|
ANONYMOUS_PASSWORD_REQUIREMENT,
|
|
PollGovernedInvitationCommand,
|
|
PollGovernedResponseCommand,
|
|
PollParticipationPolicy,
|
|
PollResponseGatewayRef,
|
|
)
|
|
from govoplan_poll.backend.router import (
|
|
api_close_poll,
|
|
api_create_poll,
|
|
api_create_poll_invitation,
|
|
api_get_public_poll,
|
|
api_list_poll_invitations,
|
|
api_list_poll_responses,
|
|
api_revoke_poll_invitation,
|
|
api_submit_poll_response,
|
|
api_submit_public_poll_response,
|
|
api_update_poll,
|
|
)
|
|
from govoplan_poll.backend.schemas import (
|
|
PollAnswerInput,
|
|
PollCreateRequest,
|
|
PollInvitationCreateRequest,
|
|
PollOptionInput,
|
|
PollSubmitResponseRequest,
|
|
PollUpdateRequest,
|
|
)
|
|
from govoplan_poll.backend.service import (
|
|
PollError,
|
|
_lock_poll_for_response,
|
|
add_poll_option,
|
|
create_poll,
|
|
create_poll_invitation,
|
|
)
|
|
|
|
|
|
class PollParticipationGatewayTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
self.tables = [
|
|
Poll.__table__,
|
|
PollOption.__table__,
|
|
PollResponse.__table__,
|
|
PollInvitation.__table__,
|
|
PollParticipationSubmission.__table__,
|
|
PollLifecycleTransition.__table__,
|
|
]
|
|
Base.metadata.create_all(self.engine, tables=self.tables)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.session: Session = self.Session()
|
|
self.provider = SqlPollSchedulingProvider()
|
|
self.gateway = PollResponseGatewayRef(
|
|
module_id="example_gateway",
|
|
resource_type="availability_request",
|
|
resource_id="request-1",
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables)))
|
|
self.engine.dispose()
|
|
|
|
def _poll(self) -> Poll:
|
|
poll = create_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="owner-1",
|
|
payload=PollCreateRequest(
|
|
title="Availability",
|
|
kind="availability",
|
|
status="open",
|
|
options=[
|
|
PollOptionInput(key="slot-1", label="Monday"),
|
|
PollOptionInput(key="slot-2", label="Tuesday"),
|
|
],
|
|
),
|
|
)
|
|
return poll
|
|
|
|
def _invitation(
|
|
self,
|
|
poll: Poll,
|
|
*,
|
|
policy: PollParticipationPolicy | None = None,
|
|
email: str | None = None,
|
|
expires_at=None,
|
|
):
|
|
return self.provider.create_governed_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollGovernedInvitationCommand(
|
|
gateway=self.gateway,
|
|
policy=policy or PollParticipationPolicy(),
|
|
email=email,
|
|
expires_at=expires_at,
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _principal(*scopes: str) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
display_name="Participant One",
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
|
|
@staticmethod
|
|
def _answers(*values: str) -> tuple[PollAnswerRequest, ...]:
|
|
return tuple(
|
|
PollAnswerRequest(option_key=f"slot-{index}", value=value)
|
|
for index, value in enumerate(values, start=1)
|
|
)
|
|
|
|
def test_gateway_bound_invitation_cannot_use_legacy_public_routes(self) -> None:
|
|
poll = self._poll()
|
|
invitation_ref = self._invitation(poll)
|
|
|
|
with self.assertRaises(HTTPException) as view_error:
|
|
api_get_public_poll(invitation_ref.token, session=self.session)
|
|
with self.assertRaises(HTTPException) as response_error:
|
|
api_submit_public_poll_response(
|
|
invitation_ref.token,
|
|
PollSubmitResponseRequest(),
|
|
session=self.session,
|
|
)
|
|
with self.assertRaises(PollCapabilityError) as wrong_gateway:
|
|
self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation_ref.token,
|
|
gateway=PollResponseGatewayRef(
|
|
module_id="other",
|
|
resource_type="availability_request",
|
|
resource_id="request-1",
|
|
),
|
|
)
|
|
|
|
self.assertEqual(view_error.exception.status_code, 404)
|
|
self.assertEqual(response_error.exception.status_code, 404)
|
|
self.assertEqual(view_error.exception.detail, "Poll invitation not found")
|
|
self.assertEqual(response_error.exception.detail, "Poll invitation not found")
|
|
self.assertEqual(str(wrong_gateway.exception), "Poll invitation not found")
|
|
|
|
def test_module_owned_poll_has_no_ordinary_participation_bypass(self) -> None:
|
|
poll = self._poll()
|
|
legacy_invitation, legacy_token = create_poll_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
payload=PollInvitationCreateRequest(respondent_id="membership-1"),
|
|
)
|
|
poll.context_module = self.gateway.module_id
|
|
poll.context_resource_type = self.gateway.resource_type
|
|
poll.context_resource_id = self.gateway.resource_id
|
|
self.session.flush()
|
|
|
|
with self.assertRaises(HTTPException) as direct_response:
|
|
api_submit_poll_response(
|
|
poll.id,
|
|
PollSubmitResponseRequest(
|
|
answers=[
|
|
PollAnswerInput(option_key="slot-1", value="available")
|
|
]
|
|
),
|
|
session=self.session,
|
|
principal=self._principal("poll:response:write"),
|
|
)
|
|
with self.assertRaises(HTTPException) as ordinary_invitation:
|
|
api_create_poll_invitation(
|
|
poll.id,
|
|
PollInvitationCreateRequest(respondent_id="membership-2"),
|
|
session=self.session,
|
|
principal=self._principal("poll:poll:write"),
|
|
)
|
|
with self.assertRaises(HTTPException) as forged_governed_invitation:
|
|
api_create_poll_invitation(
|
|
poll.id,
|
|
PollInvitationCreateRequest(
|
|
respondent_id="membership-2",
|
|
response_gateway={
|
|
"module_id": self.gateway.module_id,
|
|
"resource_type": self.gateway.resource_type,
|
|
"resource_id": self.gateway.resource_id,
|
|
},
|
|
participation_policy={},
|
|
),
|
|
session=self.session,
|
|
principal=self._principal("poll:poll:write"),
|
|
)
|
|
with self.assertRaises(HTTPException) as legacy_view:
|
|
api_get_public_poll(legacy_token, session=self.session)
|
|
with self.assertRaises(HTTPException) as legacy_response:
|
|
api_submit_public_poll_response(
|
|
legacy_token,
|
|
PollSubmitResponseRequest(
|
|
answers=[
|
|
PollAnswerInput(option_key="slot-1", value="available")
|
|
]
|
|
),
|
|
session=self.session,
|
|
)
|
|
|
|
self.assertEqual(direct_response.exception.status_code, 400)
|
|
self.assertEqual(
|
|
direct_response.exception.detail,
|
|
"Poll participation is governed by its owning module",
|
|
)
|
|
self.assertEqual(ordinary_invitation.exception.status_code, 400)
|
|
self.assertEqual(forged_governed_invitation.exception.status_code, 400)
|
|
self.assertEqual(
|
|
forged_governed_invitation.exception.detail,
|
|
"Governed invitations must be created through the participation gateway capability",
|
|
)
|
|
self.assertEqual(legacy_view.exception.status_code, 404)
|
|
self.assertEqual(legacy_response.exception.status_code, 404)
|
|
self.assertIsNone(legacy_invitation.last_used_at)
|
|
self.assertEqual(self.session.query(PollResponse).count(), 0)
|
|
|
|
with self.assertRaisesRegex(
|
|
PollCapabilityError,
|
|
"Participation gateway does not own this poll",
|
|
):
|
|
self.provider.create_governed_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollGovernedInvitationCommand(
|
|
gateway=PollResponseGatewayRef(
|
|
module_id="other",
|
|
resource_type=self.gateway.resource_type,
|
|
resource_id=self.gateway.resource_id,
|
|
),
|
|
policy=PollParticipationPolicy(),
|
|
),
|
|
)
|
|
self.assertIsNotNone(self._invitation(poll).id)
|
|
|
|
def test_governance_adoption_blocks_existing_legacy_links_but_not_other_polls(self) -> None:
|
|
governed_poll = self._poll()
|
|
_legacy_invitation, legacy_token = create_poll_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=governed_poll.id,
|
|
payload=PollInvitationCreateRequest(),
|
|
)
|
|
self._invitation(governed_poll)
|
|
|
|
with self.assertRaises(HTTPException) as legacy_view:
|
|
api_get_public_poll(legacy_token, session=self.session)
|
|
self.assertEqual(legacy_view.exception.status_code, 404)
|
|
with self.assertRaisesRegex(
|
|
PollCapabilityError,
|
|
"Poll participation is governed by its owning module",
|
|
):
|
|
self.provider.submit_response(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=governed_poll.id,
|
|
command=PollSubmitResponseCommand(
|
|
respondent_id="account-1",
|
|
respondent_label="Participant One",
|
|
answers=(),
|
|
metadata={},
|
|
),
|
|
)
|
|
|
|
ordinary_poll = self._poll()
|
|
ordinary_invitation, ordinary_token = create_poll_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=ordinary_poll.id,
|
|
payload=PollInvitationCreateRequest(),
|
|
)
|
|
self.assertEqual(
|
|
api_get_public_poll(ordinary_token, session=self.session).id,
|
|
ordinary_poll.id,
|
|
)
|
|
self.assertEqual(ordinary_invitation.poll_id, ordinary_poll.id)
|
|
|
|
def test_module_owner_controls_management_options_revocation_and_raw_data(self) -> None:
|
|
poll_ref = self.provider.create_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="owner-1",
|
|
command=PollCreateCommand(
|
|
title="Owned availability",
|
|
kind="availability",
|
|
status="open",
|
|
visibility="private",
|
|
result_visibility="organizer",
|
|
context_module=self.gateway.module_id,
|
|
context_resource_type=self.gateway.resource_type,
|
|
context_resource_id=self.gateway.resource_id,
|
|
options=(
|
|
PollOptionRequest(key="slot-1", label="Monday"),
|
|
PollOptionRequest(key="slot-2", label="Tuesday"),
|
|
),
|
|
),
|
|
)
|
|
poll = self.session.query(Poll).filter(Poll.id == poll_ref.id).one()
|
|
invitation = self._invitation(poll)
|
|
manager = self._principal("poll:poll:read", "poll:poll:write")
|
|
|
|
with self.assertRaises(HTTPException) as update_error:
|
|
api_update_poll(
|
|
poll.id,
|
|
PollUpdateRequest(title="Generic overwrite"),
|
|
session=self.session,
|
|
principal=manager,
|
|
)
|
|
with self.assertRaises(HTTPException) as transition_error:
|
|
api_close_poll(
|
|
poll.id,
|
|
idempotency_key=None,
|
|
session=self.session,
|
|
principal=manager,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
PollError,
|
|
"Poll mutations must use its owning module capability",
|
|
):
|
|
add_poll_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
option=PollOptionInput(key="slot-3", label="Wednesday"),
|
|
)
|
|
with self.assertRaises(HTTPException) as revoke_error:
|
|
api_revoke_poll_invitation(
|
|
poll.id,
|
|
invitation.id,
|
|
session=self.session,
|
|
principal=manager,
|
|
)
|
|
with self.assertRaises(HTTPException) as responses_error:
|
|
api_list_poll_responses(
|
|
poll.id,
|
|
session=self.session,
|
|
principal=manager,
|
|
)
|
|
with self.assertRaises(HTTPException) as invitations_error:
|
|
api_list_poll_invitations(
|
|
poll.id,
|
|
session=self.session,
|
|
principal=manager,
|
|
)
|
|
|
|
self.assertEqual(update_error.exception.status_code, 409)
|
|
self.assertEqual(transition_error.exception.status_code, 409)
|
|
self.assertEqual(revoke_error.exception.status_code, 409)
|
|
self.assertEqual(responses_error.exception.status_code, 403)
|
|
self.assertEqual(invitations_error.exception.status_code, 403)
|
|
self.assertEqual(poll.title, "Owned availability")
|
|
self.assertEqual(poll.status, "open")
|
|
self.assertIsNone(
|
|
self.session.query(PollInvitation)
|
|
.filter(PollInvitation.id == invitation.id)
|
|
.one()
|
|
.revoked_at
|
|
)
|
|
|
|
self.provider.update_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollUpdateCommand(
|
|
title="Owner update",
|
|
description=None,
|
|
visibility="private",
|
|
result_visibility="organizer",
|
|
allow_anonymous=False,
|
|
allow_response_update=True,
|
|
closes_at=None,
|
|
),
|
|
)
|
|
added = self.provider.add_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollOptionRequest(key="slot-3", label="Wednesday"),
|
|
)
|
|
revoked = self.provider.revoke_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
)
|
|
closed = self.provider.close_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
)
|
|
|
|
self.assertEqual(poll.title, "Owner update")
|
|
self.assertEqual(added.position, 2)
|
|
self.assertIsNotNone(revoked.revoked_at)
|
|
self.assertEqual(closed.status, "closed")
|
|
self.assertEqual(self.session.query(PollLifecycleTransition).count(), 1)
|
|
|
|
def test_ordinary_api_cannot_claim_module_context_but_remains_manageable(self) -> None:
|
|
principal = self._principal("poll:poll:write")
|
|
poll = self._poll()
|
|
|
|
with self.assertRaises(HTTPException) as create_error:
|
|
api_create_poll(
|
|
PollCreateRequest(
|
|
title="Forged module Poll",
|
|
kind="yes_no",
|
|
context_module="scheduling",
|
|
context_resource_type="scheduling_request",
|
|
context_resource_id="request-forged",
|
|
workflow_state="collecting_availability",
|
|
),
|
|
session=self.session,
|
|
principal=principal,
|
|
)
|
|
with self.assertRaises(HTTPException) as ownership_error:
|
|
api_update_poll(
|
|
poll.id,
|
|
PollUpdateRequest(
|
|
context_module="scheduling",
|
|
context_resource_type="scheduling_request",
|
|
context_resource_id="request-forged",
|
|
workflow_state="collecting_availability",
|
|
),
|
|
session=self.session,
|
|
principal=principal,
|
|
)
|
|
self.assertEqual(create_error.exception.status_code, 409)
|
|
self.assertEqual(ownership_error.exception.status_code, 409)
|
|
self.assertIsNone(poll.context_module)
|
|
|
|
updated = api_update_poll(
|
|
poll.id,
|
|
PollUpdateRequest(title="Ordinary update"),
|
|
session=self.session,
|
|
principal=principal,
|
|
)
|
|
closed = api_close_poll(
|
|
poll.id,
|
|
idempotency_key=None,
|
|
session=self.session,
|
|
principal=principal,
|
|
)
|
|
self.assertEqual(updated.title, "Ordinary update")
|
|
self.assertEqual(closed.poll.status, "closed")
|
|
|
|
def test_response_lock_requests_database_row_lock_for_concurrent_capacity(self) -> None:
|
|
session = MagicMock()
|
|
query = MagicMock()
|
|
poll = MagicMock()
|
|
session.query.return_value = query
|
|
query.filter.return_value = query
|
|
query.populate_existing.return_value = query
|
|
query.with_for_update.return_value = query
|
|
query.first.return_value = poll
|
|
|
|
result = _lock_poll_for_response(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
poll_id="poll-1",
|
|
)
|
|
|
|
self.assertIs(result, poll)
|
|
query.with_for_update.assert_called_once_with()
|
|
|
|
def test_invalid_revoked_and_expired_links_have_the_same_public_error(self) -> None:
|
|
poll = self._poll()
|
|
active = self._invitation(poll)
|
|
revoked = self._invitation(poll)
|
|
expired = self._invitation(poll, expires_at=utcnow() + timedelta(minutes=1))
|
|
self.session.query(PollInvitation).filter(PollInvitation.id == revoked.id).one().revoked_at = utcnow()
|
|
self.session.query(PollInvitation).filter(
|
|
PollInvitation.id == expired.id
|
|
).one().expires_at = utcnow() - timedelta(seconds=1)
|
|
self.session.flush()
|
|
|
|
details: list[tuple[int, str]] = []
|
|
for token in ("not-a-token", active.token, revoked.token, expired.token):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
api_get_public_poll(token, session=self.session)
|
|
details.append((caught.exception.status_code, caught.exception.detail))
|
|
|
|
self.assertEqual(details, [(404, "Poll invitation not found")] * 4)
|
|
|
|
def test_policy_is_reenforced_by_poll_after_gateway_authorization(self) -> None:
|
|
poll = self._poll()
|
|
invitation = self._invitation(
|
|
poll,
|
|
policy=PollParticipationPolicy(
|
|
single_choice=True,
|
|
allow_maybe=False,
|
|
allow_comments=False,
|
|
participant_email_required=True,
|
|
anonymous_password_required=True,
|
|
),
|
|
)
|
|
|
|
attempts = (
|
|
(
|
|
PollGovernedResponseCommand(
|
|
answers=self._answers("available"),
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
),
|
|
"email is required",
|
|
),
|
|
(
|
|
PollGovernedResponseCommand(
|
|
participant_email="person@example.test",
|
|
answers=self._answers("available"),
|
|
),
|
|
"password verification is required",
|
|
),
|
|
(
|
|
PollGovernedResponseCommand(
|
|
participant_email="person@example.test",
|
|
answers=self._answers("maybe"),
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
),
|
|
"Maybe responses are not enabled",
|
|
),
|
|
(
|
|
PollGovernedResponseCommand(
|
|
participant_email="person@example.test",
|
|
answers=self._answers("available", "maybe"),
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
),
|
|
"Maybe responses are not enabled",
|
|
),
|
|
(
|
|
PollGovernedResponseCommand(
|
|
participant_email="person@example.test",
|
|
answers=self._answers("available"),
|
|
comment="Not permitted",
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
),
|
|
"Comments are not enabled",
|
|
),
|
|
)
|
|
for command, message in attempts:
|
|
with self.subTest(message=message), self.assertRaisesRegex(
|
|
PollCapabilityError,
|
|
message,
|
|
):
|
|
self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=command,
|
|
)
|
|
|
|
accepted = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email=" Person@Example.Test ",
|
|
answers=self._answers("available"),
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
participant_email="person@example.test",
|
|
)
|
|
protected_context = self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
participant_email="person@example.test",
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
)
|
|
self.assertEqual(accepted.participant_email, "person@example.test")
|
|
self.assertIsNotNone(protected_context.response)
|
|
stored_invitation = self.session.query(PollInvitation).filter(PollInvitation.id == invitation.id).one()
|
|
self.assertEqual(stored_invitation.participation_policy_["anonymous_password_required"], True)
|
|
self.assertNotIn("correct horse", str(stored_invitation.participation_policy_))
|
|
self.assertFalse(any("password" in column.name for column in PollInvitation.__table__.columns))
|
|
|
|
def test_authenticated_participant_does_not_need_anonymous_challenges(self) -> None:
|
|
poll = self._poll()
|
|
invitation = self._invitation(
|
|
poll,
|
|
policy=PollParticipationPolicy(
|
|
participant_email_required=True,
|
|
anonymous_password_required=True,
|
|
),
|
|
)
|
|
|
|
result = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
respondent_id="account-1",
|
|
participant_is_authenticated=True,
|
|
answers=self._answers("available"),
|
|
),
|
|
)
|
|
context = self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
participant_is_authenticated=True,
|
|
)
|
|
anonymous_context = self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
participant_is_authenticated=False,
|
|
verified_requirements=frozenset({ANONYMOUS_PASSWORD_REQUIREMENT}),
|
|
)
|
|
|
|
self.assertEqual(result.response.respondent_id, "account-1")
|
|
self.assertIsNotNone(context.response)
|
|
self.assertEqual(context.response.response.respondent_id, "account-1")
|
|
self.assertIsNone(anonymous_context.response)
|
|
|
|
def test_authenticated_invitation_id_path_is_bound_and_tokenless(self) -> None:
|
|
poll = self._poll()
|
|
invitation = self.provider.create_governed_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollGovernedInvitationCommand(
|
|
gateway=self.gateway,
|
|
policy=PollParticipationPolicy(max_participants_per_option=1),
|
|
respondent_id="account-1",
|
|
),
|
|
)
|
|
response = self.provider.submit_authenticated_response(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
command=PollGovernedResponseCommand(
|
|
respondent_id="account-1",
|
|
participant_is_authenticated=True,
|
|
answers=self._answers("available"),
|
|
),
|
|
)
|
|
context = self.provider.resolve_authenticated_participation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
)
|
|
|
|
self.assertEqual(response.response.respondent_id, "account-1")
|
|
self.assertEqual(context.response.response.answers, response.response.answers)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_authenticated_participation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
respondent_id="account-2",
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_authenticated_participation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=PollResponseGatewayRef(
|
|
module_id="other",
|
|
resource_type="availability_request",
|
|
resource_id="request-1",
|
|
),
|
|
respondent_id="account-1",
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.submit_authenticated_response(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
command=PollGovernedResponseCommand(
|
|
respondent_id="account-1",
|
|
participant_is_authenticated=False,
|
|
answers=self._answers("available"),
|
|
),
|
|
)
|
|
self.provider.revoke_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_authenticated_participation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
respondent_id="account-1",
|
|
)
|
|
|
|
def test_allow_maybe_uses_resolved_option_identity(self) -> None:
|
|
poll = create_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="owner-1",
|
|
payload=PollCreateRequest(
|
|
title="Decision",
|
|
kind="yes_no_maybe",
|
|
status="open",
|
|
),
|
|
)
|
|
maybe = next(option for option in poll.options if option.key == "maybe")
|
|
invitation = self._invitation(
|
|
poll,
|
|
policy=PollParticipationPolicy(allow_maybe=False),
|
|
)
|
|
|
|
with self.assertRaisesRegex(PollCapabilityError, "Maybe responses are not enabled"):
|
|
self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
answers=(PollAnswerRequest(option_id=maybe.id),),
|
|
),
|
|
)
|
|
|
|
def test_invalid_policy_uses_stable_capability_error(self) -> None:
|
|
poll = self._poll()
|
|
|
|
with self.assertRaisesRegex(PollCapabilityError, "Invalid participation policy"):
|
|
self._invitation(
|
|
poll,
|
|
policy=PollParticipationPolicy(max_participants_per_option=0),
|
|
)
|
|
|
|
def test_secret_like_gateway_metadata_is_never_persisted(self) -> None:
|
|
poll = self._poll()
|
|
with self.assertRaisesRegex(PollCapabilityError, "Sensitive participation metadata"):
|
|
self.provider.create_governed_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollGovernedInvitationCommand(
|
|
gateway=self.gateway,
|
|
policy=PollParticipationPolicy(),
|
|
metadata={"anonymous_password": "correct horse"},
|
|
),
|
|
)
|
|
invitation = self._invitation(poll)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Sensitive participation metadata"):
|
|
self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
answers=self._answers("available"),
|
|
metadata={"nested": {"credential": "correct horse"}},
|
|
),
|
|
)
|
|
self.assertNotIn("correct horse", str(self.session.query(PollInvitation).all()))
|
|
self.assertEqual(self.session.query(PollResponse).count(), 0)
|
|
|
|
for key in ("passwordHash", "accessToken", "credentialvalue"):
|
|
with self.subTest(key=key), self.assertRaisesRegex(
|
|
PollCapabilityError,
|
|
"Sensitive participation metadata",
|
|
):
|
|
self.provider.create_governed_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollGovernedInvitationCommand(
|
|
gateway=self.gateway,
|
|
policy=PollParticipationPolicy(),
|
|
metadata={key: "sensitive"},
|
|
),
|
|
)
|
|
|
|
def test_capacity_editing_prefill_and_idempotency_share_one_locked_path(self) -> None:
|
|
poll = self._poll()
|
|
policy = PollParticipationPolicy(
|
|
max_participants_per_option=1,
|
|
allow_comments=True,
|
|
participant_email_required=True,
|
|
)
|
|
first_invitation = self._invitation(poll, policy=policy)
|
|
second_invitation = self._invitation(poll, policy=policy)
|
|
first_command = PollGovernedResponseCommand(
|
|
participant_email="one@example.test",
|
|
answers=self._answers("available"),
|
|
comment=" First choice ",
|
|
idempotency_key="submission-1",
|
|
)
|
|
|
|
first = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
command=first_command,
|
|
)
|
|
replay = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
command=first_command,
|
|
)
|
|
context = self.provider.resolve_participation(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
participant_email="one@example.test",
|
|
)
|
|
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(replay.response.submitted_at, first.response.submitted_at)
|
|
self.assertEqual(context.response.response.answers, first.response.answers)
|
|
self.assertEqual(context.response.comment, "First choice")
|
|
self.assertEqual(self.session.query(PollParticipationSubmission).count(), 1)
|
|
|
|
with self.assertRaisesRegex(PollCapabilityError, "different response"):
|
|
self.provider.submit_governed_response(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email="one@example.test",
|
|
answers=self._answers("unavailable"),
|
|
idempotency_key="submission-1",
|
|
),
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Participant limit reached"):
|
|
self.provider.submit_governed_response(
|
|
self.session,
|
|
token=second_invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email="two@example.test",
|
|
answers=self._answers("available"),
|
|
),
|
|
)
|
|
|
|
edited = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email="one@example.test",
|
|
answers=self._answers("unavailable"),
|
|
comment="Changed",
|
|
idempotency_key="submission-2",
|
|
),
|
|
)
|
|
claimed = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=second_invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email="two@example.test",
|
|
answers=self._answers("available"),
|
|
),
|
|
)
|
|
current_resource_replay = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=first_invitation.token,
|
|
gateway=self.gateway,
|
|
command=first_command,
|
|
)
|
|
self.assertEqual(edited.response.respondent_id, first.response.respondent_id)
|
|
self.assertNotEqual(claimed.response.respondent_id, first.response.respondent_id)
|
|
self.assertTrue(current_resource_replay.replayed)
|
|
self.assertEqual(current_resource_replay.response.answers[0].value, "unavailable")
|
|
self.assertEqual(current_resource_replay.comment, "Changed")
|
|
|
|
def test_option_mutations_and_invitation_revocation_are_safe_and_idempotent(self) -> None:
|
|
poll = self._poll()
|
|
invitation = self._invitation(poll)
|
|
added = self.provider.add_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollOptionRequest(key="slot-3", label="Wednesday"),
|
|
)
|
|
replayed_add = self.provider.add_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollOptionRequest(key="slot-3", label="Wednesday"),
|
|
)
|
|
self.assertEqual(poll.min_choices, 1)
|
|
self.assertEqual(poll.max_choices, 3)
|
|
response = self.provider.submit_governed_response(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
command=PollGovernedResponseCommand(
|
|
participant_email="person@example.test",
|
|
answers=self._answers("available", "maybe", "unavailable"),
|
|
),
|
|
)
|
|
|
|
removed = self.provider.remove_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
option_id=poll.options[0].id,
|
|
)
|
|
replayed_remove = self.provider.remove_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
option_id=poll.options[0].id,
|
|
)
|
|
self.assertEqual(poll.min_choices, 1)
|
|
self.assertEqual(poll.max_choices, 2)
|
|
remaining = self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
participant_email="person@example.test",
|
|
).response
|
|
|
|
self.assertFalse(added.replayed)
|
|
self.assertTrue(replayed_add.replayed)
|
|
self.assertEqual(replayed_add.id, added.id)
|
|
self.assertEqual(removed.invalidated_response_count, 1)
|
|
self.assertTrue(replayed_remove.replayed)
|
|
self.assertEqual(len(remaining.response.answers), 2)
|
|
self.assertEqual(remaining.response.answers[0].option_key, "slot-2")
|
|
self.assertEqual(response.response.respondent_id, remaining.response.respondent_id)
|
|
|
|
revoked = self.provider.revoke_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
)
|
|
replayed_revoke = self.provider.revoke_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
)
|
|
self.assertFalse(revoked.replayed)
|
|
self.assertTrue(replayed_revoke.replayed)
|
|
self.assertEqual(revoked.revoked_at, replayed_revoke.revoked_at)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
)
|
|
|
|
def test_invitation_expiry_updates_preserve_token_and_exact_owner(self) -> None:
|
|
poll = self._poll()
|
|
invitation = self._invitation(
|
|
poll,
|
|
expires_at=utcnow() + timedelta(hours=1),
|
|
)
|
|
stored = (
|
|
self.session.query(PollInvitation)
|
|
.filter(PollInvitation.id == invitation.id)
|
|
.one()
|
|
)
|
|
token_hash = stored.token_hash
|
|
expired_at = utcnow() - timedelta(minutes=1)
|
|
|
|
expired = self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
expires_at=expired_at,
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
)
|
|
future_expiry = utcnow() + timedelta(days=1)
|
|
extended = self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
expires_at=future_expiry,
|
|
)
|
|
replayed = self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
expires_at=future_expiry,
|
|
)
|
|
context = self.provider.resolve_participation(
|
|
self.session,
|
|
token=invitation.token,
|
|
gateway=self.gateway,
|
|
)
|
|
|
|
self.assertFalse(expired.replayed)
|
|
self.assertFalse(extended.replayed)
|
|
self.assertTrue(replayed.replayed)
|
|
self.assertEqual(context.invitation_id, invitation.id)
|
|
self.assertEqual(stored.token_hash, token_hash)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=PollResponseGatewayRef(
|
|
module_id="other",
|
|
resource_type=self.gateway.resource_type,
|
|
resource_id=self.gateway.resource_id,
|
|
),
|
|
expires_at=None,
|
|
)
|
|
stored.participation_policy_ = {"version": 999}
|
|
self.session.flush()
|
|
with self.assertRaises(PollCapabilityError) as corrupt_policy:
|
|
self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
expires_at=None,
|
|
)
|
|
self.assertEqual(str(corrupt_policy.exception), "Poll invitation not found")
|
|
self.provider.revoke_invitation(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
)
|
|
with self.assertRaisesRegex(PollCapabilityError, "Poll invitation not found"):
|
|
self.provider.update_invitation_expiry(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
invitation_id=invitation.id,
|
|
gateway=self.gateway,
|
|
expires_at=None,
|
|
)
|
|
|
|
def test_option_mutation_replays_survive_close_and_bounds_stay_valid(self) -> None:
|
|
poll = self._poll()
|
|
added = self.provider.add_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollOptionRequest(key="slot-3", label="Wednesday"),
|
|
)
|
|
removed = self.provider.remove_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
option_id=poll.options[0].id,
|
|
)
|
|
poll.status = "closed"
|
|
self.session.flush()
|
|
|
|
replayed_add = self.provider.add_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
command=PollOptionRequest(key="slot-3", label="Wednesday"),
|
|
)
|
|
replayed_remove = self.provider.remove_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=poll.id,
|
|
option_id=removed.id,
|
|
)
|
|
|
|
self.assertEqual(replayed_add.id, added.id)
|
|
self.assertTrue(replayed_add.replayed)
|
|
self.assertTrue(replayed_remove.replayed)
|
|
|
|
multiple_choice = create_poll(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
user_id="owner-1",
|
|
payload=PollCreateRequest(
|
|
title="Choose",
|
|
kind="multiple_choice",
|
|
status="open",
|
|
max_choices=3,
|
|
options=[
|
|
PollOptionInput(key=f"choice-{index}", label=f"Choice {index}")
|
|
for index in range(3)
|
|
],
|
|
),
|
|
)
|
|
self.provider.remove_option(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
poll_id=multiple_choice.id,
|
|
option_id=multiple_choice.options[0].id,
|
|
)
|
|
self.assertEqual(multiple_choice.max_choices, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|