From 6abe292ac8c6b8e31f112f1c01866b4a55880a00 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 03:21:03 +0200 Subject: [PATCH] feat(core): define governed poll participation contract --- src/govoplan_core/core/poll_participation.py | 272 +++++++++++++++++++ tests/test_poll_participation_contract.py | 84 ++++++ 2 files changed, 356 insertions(+) create mode 100644 src/govoplan_core/core/poll_participation.py create mode 100644 tests/test_poll_participation_contract.py diff --git a/src/govoplan_core/core/poll_participation.py b/src/govoplan_core/core/poll_participation.py new file mode 100644 index 0000000..08cf25e --- /dev/null +++ b/src/govoplan_core/core/poll_participation.py @@ -0,0 +1,272 @@ +from __future__ import annotations + +import hashlib +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Protocol, runtime_checkable + +from govoplan_core.core.poll import ( + PollAnswerRequest, + PollInvitationRef, + PollOptionRequest, + PollResponseRef, +) + + +CAPABILITY_POLL_PARTICIPATION_GATEWAY = "poll.participation_gateway" +# Policy-attestation identifier, not a credential or credential default. +ANONYMOUS_PASSWORD_REQUIREMENT = "anonymous_password" # nosec B105 # noqa: S105 +PARTICIPATION_POLICY_VERSION = 1 + + +def participation_token_fingerprint(token: str) -> str: + """Return a non-reversible identifier suitable for audit/throttle keys.""" + + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True, slots=True) +class PollResponseGatewayRef: + """Stable identity of the module resource governing a participation link.""" + + module_id: str + resource_type: str + resource_id: str + + +@dataclass(frozen=True, slots=True) +class PollParticipationPolicy: + """Generic response rules snapshotted onto one signed invitation. + + ``single_choice`` treats every non-``unavailable`` availability answer as + a selection. Capacity is reserved only by ``available`` answers (and by + selected answers for non-availability polls), so ``maybe`` never consumes + a place. + """ + + version: int = PARTICIPATION_POLICY_VERSION + single_choice: bool = False + allow_maybe: bool = True + max_participants_per_option: int | None = None + allow_comments: bool = False + participant_email_required: bool = False + anonymous_password_required: bool = False + + +@dataclass(frozen=True, slots=True) +class PollGovernedInvitationCommand: + gateway: PollResponseGatewayRef + policy: PollParticipationPolicy + respondent_id: str | None = None + respondent_label: str | None = None + email: str | None = None + expires_at: datetime | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollGovernedResponseCommand: + """Submission already authorized by the named in-process gateway. + + Passwords never cross this boundary. A gateway that owns a password + verifier reports the completed check through ``verified_requirements``. + Poll independently re-enforces the remaining snapshotted rules while + holding its Poll-row lock. + """ + + respondent_id: str | None = None + respondent_label: str | None = None + participant_email: str | None = None + participant_is_authenticated: bool = False + answers: tuple[PollAnswerRequest, ...] = () + comment: str | None = None + verified_requirements: frozenset[str] = frozenset() + idempotency_key: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollGovernedResponseRef: + response: PollResponseRef + participant_email: str | None = None + comment: str | None = None + replayed: bool = False + + +@dataclass(frozen=True, slots=True) +class PollOptionMutationRef: + id: str + position: int + replayed: bool = False + invalidated_response_count: int = 0 + + +@dataclass(frozen=True, slots=True) +class PollInvitationRevocationRef: + id: str + revoked_at: datetime + replayed: bool = False + + +@dataclass(frozen=True, slots=True) +class PollInvitationExpiryRef: + id: str + expires_at: datetime | None + replayed: bool = False + + +@dataclass(frozen=True, slots=True) +class PollParticipationContextRef: + invitation_id: str + tenant_id: str + poll_id: str + gateway: PollResponseGatewayRef + policy: PollParticipationPolicy + respondent_id: str | None = None + respondent_label: str | None = None + email: str | None = None + response: PollGovernedResponseRef | None = None + + +@runtime_checkable +class PollParticipationGatewayProvider(Protocol): + def create_governed_invitation( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollGovernedInvitationCommand, + ) -> PollInvitationRef: + ... + + def resolve_participation( + self, + session: object, + *, + token: str, + gateway: PollResponseGatewayRef, + respondent_id: str | None = None, + participant_email: str | None = None, + participant_is_authenticated: bool = False, + verified_requirements: frozenset[str] = frozenset(), + ) -> PollParticipationContextRef: + """Resolve and prefill one valid invitation for the exact gateway.""" + + ... + + def submit_governed_response( + self, + session: object, + *, + token: str, + gateway: PollResponseGatewayRef, + command: PollGovernedResponseCommand, + ) -> PollGovernedResponseRef: + ... + + def resolve_authenticated_participation( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + invitation_id: str, + gateway: PollResponseGatewayRef, + respondent_id: str, + ) -> PollParticipationContextRef: + """Resolve one governed invitation without retaining its public token.""" + + ... + + def submit_authenticated_response( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + invitation_id: str, + gateway: PollResponseGatewayRef, + respondent_id: str, + command: PollGovernedResponseCommand, + ) -> PollGovernedResponseRef: + """Submit atomically for the exact authenticated invitation identity.""" + + ... + + def add_option( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollOptionRequest, + ) -> PollOptionMutationRef: + ... + + def remove_option( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + option_id: str, + ) -> PollOptionMutationRef: + ... + + def revoke_invitation( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + invitation_id: str, + ) -> PollInvitationRevocationRef: + ... + + def update_invitation_expiry( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + invitation_id: str, + gateway: PollResponseGatewayRef, + expires_at: datetime | None, + ) -> PollInvitationExpiryRef: + """Expire or extend a non-revoked governed invitation in place.""" + + ... + + +def poll_participation_gateway_provider( + registry: object | None, +) -> PollParticipationGatewayProvider | None: + """Resolve the governed Poll gateway without importing its implementation.""" + + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY): + return None + capability = registry.capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY) + return capability if isinstance(capability, PollParticipationGatewayProvider) else None + + +__all__ = [ + "ANONYMOUS_PASSWORD_REQUIREMENT", + "CAPABILITY_POLL_PARTICIPATION_GATEWAY", + "PARTICIPATION_POLICY_VERSION", + "PollGovernedInvitationCommand", + "PollGovernedResponseCommand", + "PollGovernedResponseRef", + "PollInvitationExpiryRef", + "PollInvitationRevocationRef", + "PollOptionMutationRef", + "PollParticipationContextRef", + "PollParticipationGatewayProvider", + "PollParticipationPolicy", + "PollResponseGatewayRef", + "participation_token_fingerprint", + "poll_participation_gateway_provider", +] diff --git a/tests/test_poll_participation_contract.py b/tests/test_poll_participation_contract.py new file mode 100644 index 0000000..e79ac2f --- /dev/null +++ b/tests/test_poll_participation_contract.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.poll_participation import ( + CAPABILITY_POLL_PARTICIPATION_GATEWAY, + PollParticipationGatewayProvider, + PollParticipationPolicy, + participation_token_fingerprint, + poll_participation_gateway_provider, +) + + +class _CompleteGateway: + def create_governed_invitation(self, *args, **kwargs): + raise NotImplementedError + + def resolve_participation(self, *args, **kwargs): + raise NotImplementedError + + def submit_governed_response(self, *args, **kwargs): + raise NotImplementedError + + def resolve_authenticated_participation(self, *args, **kwargs): + raise NotImplementedError + + def submit_authenticated_response(self, *args, **kwargs): + raise NotImplementedError + + def add_option(self, *args, **kwargs): + raise NotImplementedError + + def remove_option(self, *args, **kwargs): + raise NotImplementedError + + def revoke_invitation(self, *args, **kwargs): + raise NotImplementedError + + def update_invitation_expiry(self, *args, **kwargs): + raise NotImplementedError + + +class _Registry: + def __init__(self, capability: object | None) -> None: + self.capability_value = capability + + def has_capability(self, name: str) -> bool: + return ( + name == CAPABILITY_POLL_PARTICIPATION_GATEWAY + and self.capability_value is not None + ) + + def capability(self, name: str) -> object: + if not self.has_capability(name): + raise KeyError(name) + return self.capability_value + + +class PollParticipationContractTests(unittest.TestCase): + def test_capability_name_and_policy_defaults_remain_stable(self) -> None: + policy = PollParticipationPolicy() + + self.assertEqual("poll.participation_gateway", CAPABILITY_POLL_PARTICIPATION_GATEWAY) + self.assertEqual(1, policy.version) + self.assertTrue(policy.allow_maybe) + + def test_token_fingerprint_is_deterministic_and_non_reversible(self) -> None: + fingerprint = participation_token_fingerprint("secret-token") + + self.assertEqual(fingerprint, participation_token_fingerprint("secret-token")) + self.assertEqual(64, len(fingerprint)) + self.assertNotIn("secret-token", fingerprint) + + def test_registry_lookup_accepts_only_the_complete_structural_contract(self) -> None: + provider = _CompleteGateway() + + self.assertIsInstance(provider, PollParticipationGatewayProvider) + self.assertIs(provider, poll_participation_gateway_provider(_Registry(provider))) + self.assertIsNone(poll_participation_gateway_provider(_Registry(object()))) + self.assertIsNone(poll_participation_gateway_provider(None)) + + +if __name__ == "__main__": + unittest.main()