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()