Compare commits

..

19 Commits

Author SHA1 Message Date
36ceb24954 feat(poll): make exact transitions idempotent 2026-07-22 04:39:06 +02:00
8fc030772b chore(poll): bump version to 0.1.11 2026-07-22 03:42:11 +02:00
52cd362343 docs(poll): describe response retirement contract 2026-07-22 03:40:29 +02:00
8ceb3935f5 feat(poll): retire responses without erasing history 2026-07-22 03:32:05 +02:00
8292c9d709 feat(poll): reorder options without invalidating answers 2026-07-22 03:22:04 +02:00
38ecff60a5 refactor(poll): implement core participation contract 2026-07-22 03:21:08 +02:00
eb003208e4 fix(poll): reconcile concurrent respondent submissions 2026-07-22 02:54:14 +02:00
cac7733bee feat(poll): enforce active respondent uniqueness 2026-07-22 02:54:09 +02:00
e2e816cb21 chore(poll): bump version to 0.1.10 2026-07-21 21:17:04 +02:00
2e1ed6e0d8 docs(poll): document governed participation boundary 2026-07-21 21:17:04 +02:00
156486fcee feat(poll): enforce governed participation ownership 2026-07-21 21:17:04 +02:00
f839545605 refactor: simplify poll transition orchestration 2026-07-21 12:01:24 +02:00
83d04d2f33 Replace Poll runtime assertions 2026-07-21 03:16:24 +02:00
2c776437b4 fix(poll): harden lifecycle policy boundaries 2026-07-20 18:42:07 +02:00
12b3175b07 fix(responses): invalidate changed options safely 2026-07-20 18:30:04 +02:00
e0de54b756 feat(responses): support safe option editing 2026-07-20 18:20:33 +02:00
65e2d1fd65 feat(poll): enforce auditable lifecycle transitions 2026-07-20 18:18:20 +02:00
f4d5cac40d feat(poll): expose the scheduling provider 2026-07-20 17:34:03 +02:00
e7f3b1d7bf fix(poll): enforce actor-bound access and persistence 2026-07-20 17:33:56 +02:00
26 changed files with 7421 additions and 106 deletions

133
README.md
View File

@@ -50,6 +50,96 @@ and authenticated API routes for:
- signed participation links for reduced/no-Access participation
- context and workflow metadata for modules such as Scheduling
## Governed signed participation
The v0.1.10 contract lets a consuming module bind an invitation to an exact
response gateway (`module_id`, `resource_type`, and `resource_id`) and snapshot
generic participation rules onto it. A bound invitation cannot use the legacy
`GET /poll/public/{token}` or `POST /poll/public/{token}/responses` routes: both
return the same generic not-found response used for invalid, expired, and
revoked links. There is deliberately no browser-facing Poll bypass when the
owning gateway is missing.
That boundary is Poll-wide. Setting `context_module` declares a module-owned
Poll; for a standalone Poll, its first governed invitation opts the whole Poll
into governed participation. From then on, direct authenticated response
writes, ordinary invitation creation, and every legacy public link fail closed,
including links created before governance was enabled. Governed invitations can
only be created through the in-process participation capability and their
gateway must match the Poll's declared module resource (or the standalone
Poll's first gateway). Polls that never opt into governance retain their
ordinary authenticated and signed-link behavior.
Module ownership also protects management boundaries. Generic Poll update,
lifecycle, option, and invitation-revocation paths cannot mutate an owned Poll,
and generic raw-response and invitation projections cannot bypass the owning
module's privacy policy. The in-process capability supplies the exact module,
resource type, and resource id again for every mutation; Poll compares that
owner while holding its row lock. Aggregate result summaries remain reusable.
Consumers resolve and submit bound invitations through the
`poll.participation_gateway` in-process capability. The policy covers:
- at most one non-`unavailable` selection
- whether `maybe` is accepted
- a maximum number of definitive participants per option
- response comments (trimmed and limited to 4,000 characters)
- email for anonymous participants
- an anonymous-password verification requirement
For availability polls, `available` reserves capacity while `maybe` does not.
Poll re-enforces these rules under its Poll-row lock, so concurrent final-place
claims serialize on PostgreSQL. A gateway-owned password never crosses the
capability boundary: the owning module stores the sole salted verifier,
throttles attempts before checking it, and supplies only the
`anonymous_password` verification attestation. Poll rejects secret-like
invitation/response metadata and stores no password column.
Authenticated module flows do not need to retain a public bearer token. They
can resolve and submit by exact tenant, Poll, governed invitation, gateway, and
respondent identifiers. Poll rejects mismatched identities and applies the same
Poll-row lock and policy checks used by the public-token path. A consumer can
lazily create that governed invitation, retain only its id, and discard the raw
token when no public link is needed.
An owning gateway can update a non-revoked invitation's expiry in place. A past
timestamp expires the existing link immediately; a later future timestamp or
`null` reactivates that same link without exposing or rotating its bearer token.
Revoked invitations remain revoked, and exact retries are idempotent.
The capability also supports durable invitation-scoped idempotency keys,
response prefill, option addition/removal, and invitation revocation. Exact
submission retries return the existing response; key reuse with different
content is rejected. Because a response is an editable resource, replay returns
its current representation rather than an immutable snapshot of the first
submission. Option removal and option changes invalidate only answers bound to
that option, and repeated removal/revocation is an idempotent replay even after
the Poll has moved out of an editable lifecycle state.
Owning modules can also retire a participant's responses through the optional
response-retirement extension. Retirement is idempotent and soft-deletes the
live rows so aggregation and capacity checks stop counting them, while answers
and a reason/source retirement record remain available for audit. The boundary
accepts only server-trusted respondent or invitation identities and rejects
secret-like metadata.
## Identified response invariant
Poll stores at most one active response for each identified respondent in a
Poll. PostgreSQL and SQLite enforce this with the partial unique index
`uq_poll_responses_active_respondent`; anonymous and tombstoned responses do
not participate in that invariant. If two submissions race, the losing insert
is rolled back to a savepoint and follows the ordinary update policy against
the winning row. Unrelated integrity errors are not converted into response
updates.
The migration deterministically retains the latest active row by
`submitted_at DESC, id DESC` and tombstones older duplicates. Apply it while
Poll response writes are quiesced or while the platform maintenance lock is
held. The released migration-head baseline must only be advanced as part of
the reviewed release that includes this migration; it is not a development
head ledger.
## Scheduling As A Poll-Backed Workflow
Scheduling should use Poll as the reusable response collection primitive, not
@@ -63,6 +153,41 @@ Poll stores the shared poll/options/responses/result summary. Scheduling owns
the domain workflow around rooms, calendars, free/busy checks, reminders,
appointment creation, and optional Workflow integration.
## Poll lifecycle
Poll configures its lifecycle through the separate, pure transition engine in
`backend/transitions.py`. The default policy is:
- `draft``open` or `archived`
- `open``draft`, `closed`, or `archived`
- `closed``open`, `decided`, or `archived`
- `decided``open`, `decided` (an explicit re-decision), or `archived`
- `archived` → the status from which the Poll was archived
Reopening and archiving preserve responses, close history, and decision
metadata. Every applied transition has a Poll-owned lifecycle audit record.
Re-deciding always appends a record and supersedes the current decision. An
exact retry carrying the same `Idempotency-Key` is a no-op and returns the
original transition record; reusing that key for a different action or option
is rejected. A command whose requested lifecycle state already holds is also
an idempotent domain no-op, even without an idempotency identity. It returns
the current Poll with `replayed=true` and `transition=null`, does not append a
lifecycle audit record, and is emitted only to operational logs as duplicate
command telemetry. A new idempotency key supplied for such an audit-free no-op
is not consumed. Re-deciding with a different option remains a new auditable
action; repeating the current decision is a no-op.
Poll API representations expose every lifecycle action with its availability
and, when unavailable, the policy reason. Management clients can use the
`GET /poll/polls/{poll_id}/transitions` endpoint to read the durable history
instead of duplicating the matrix in their UI.
Polls created through the v0.1.9 contract start in `draft` or `open`; later
states are entered through the transition engine. Older archived rows may not
contain the status from which they were archived. Their prior state cannot be
inferred safely, so unarchive restores them to the least-permissive `draft`
state and marks that legacy fallback explicitly in the audit metadata.
## Development Install
```bash
@@ -76,3 +201,11 @@ Focused manifest verification:
cd /mnt/DATA/git/govoplan-poll
PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan-core/.venv/bin/python -m unittest discover -s tests
```
Run the optional two-session PostgreSQL race check against a disposable or
development database account that may create schemas:
```bash
GOVOPLAN_POLL_TEST_POSTGRES_URL=postgresql+psycopg://user@localhost/database \
/mnt/DATA/git/govoplan/.venv/bin/python -m pytest -q tests/test_response_uniqueness.py
```

View File

@@ -4,14 +4,14 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-poll"
version = "0.1.8"
version = "0.1.11"
description = "GovOPlaN lightweight poll and availability decision module seed."
readme = "README.md"
requires-python = ">=3.12"
license = { file = "LICENSE" }
authors = [{ name = "GovOPlaN" }]
dependencies = [
"govoplan-core>=0.1.8",
"govoplan-core>=0.1.11",
]
[tool.setuptools.packages.find]

View File

@@ -2,4 +2,4 @@
__all__ = ["__version__"]
__version__ = "0.1.8"
__version__ = "0.1.11"

View File

@@ -0,0 +1,825 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from datetime import datetime
from pydantic import ValidationError
from govoplan_core.core.poll import (
PollAnswerRef,
PollCapabilityError,
PollCreateCommand,
PollInvitationCommand,
PollInvitationRef,
PollOptionOrderCommand,
PollOptionRequest,
PollOptionRef,
PollOptionUpdateCommand,
PollRef,
PollResponseRef,
PollResponseRetirementCommand,
PollResponseRetirementRef,
PollSchedulingProvider,
PollSubmitResponseCommand,
PollUpdateCommand,
)
from govoplan_poll.backend.participation import (
ANONYMOUS_PASSWORD_REQUIREMENT,
PollGovernedInvitationCommand,
PollGovernedResponseCommand,
PollGovernedResponseRef,
PollInvitationExpiryRef,
PollInvitationRevocationRef,
PollOptionMutationRef,
PollParticipationContextRef,
PollResponseGatewayRef,
)
from govoplan_poll.backend.participation_service import (
governed_invitation,
governed_invitation_by_id,
participation_policy_payload,
participation_policy_ref,
response_for_invitation,
response_gateway_payload,
response_gateway_ref,
response_metadata,
submit_governed_poll_response,
submit_authenticated_poll_response,
update_governed_invitation_expiry,
)
from govoplan_poll.backend.db.models import PollInvitation
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollCreateRequest,
PollDecisionRequest,
PollInvitationCreateRequest,
PollOptionInput,
PollSubmitResponseRequest,
)
from govoplan_poll.backend.service import (
PollError,
PollMutationOwner,
close_poll,
add_poll_option,
create_poll,
create_poll_invitation,
decide_poll,
get_poll,
get_poll_response_for_respondents,
list_poll_responses,
open_poll,
poll_mutation_owner,
poll_owner_ref,
poll_result_summary_by_id,
response_datetime,
retire_poll_responses,
remove_poll_option,
reorder_poll_options,
revoke_poll_invitation_with_replay,
set_poll_workflow_context,
submit_poll_response,
update_poll_snapshot,
update_poll_option,
)
def _poll_ref(poll: object) -> PollRef:
return PollRef(
id=poll.id,
status=poll.status,
options=tuple(
PollOptionRef(id=option.id, position=option.position)
for option in sorted(
(option for option in poll.options if option.deleted_at is None),
key=lambda item: (item.position, item.id),
)
),
)
def _command_owner(command: PollCreateCommand) -> PollMutationOwner | None:
context = (
command.context_module,
command.context_resource_type,
command.context_resource_id,
)
if not any(value is not None for value in context):
return None
return poll_owner_ref(
module_id=command.context_module,
resource_type=command.context_resource_type,
resource_id=command.context_resource_id,
)
def _stored_owner(
session: object,
*,
tenant_id: str,
poll_id: str,
) -> PollMutationOwner | None:
return poll_mutation_owner(
get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
)
def _response_invitation_id(response: object) -> str | None:
value = (response.metadata_ or {}).get("invitation_id")
return value if isinstance(value, str) else None
def _response_ref(response: object) -> PollResponseRef:
answers: list[PollAnswerRef] = []
for answer in response.answers or []:
if not isinstance(answer, Mapping):
continue
answers.append(
PollAnswerRef(
option_id=answer.get("option_id"),
option_key=answer.get("option_key"),
value=answer.get("value"),
rank=answer.get("rank"),
)
)
return PollResponseRef(
invitation_id=_response_invitation_id(response),
submitted_at=response_datetime(response.submitted_at),
respondent_id=response.respondent_id,
answers=tuple(answers),
)
def _participation_context_ref(
session: object,
*,
invitation: PollInvitation,
respondent_id: str | None = None,
participant_email: str | None = None,
) -> PollParticipationContextRef:
policy = participation_policy_ref(invitation.participation_policy_)
response = response_for_invitation(
session,
invitation=invitation,
respondent_id=respondent_id,
participant_email=participant_email,
)
response_ref = None
if response is not None:
email, comment = response_metadata(response)
response_ref = PollGovernedResponseRef(
response=_response_ref(response),
participant_email=email,
comment=comment,
)
return PollParticipationContextRef(
invitation_id=invitation.id,
tenant_id=invitation.tenant_id,
poll_id=invitation.poll_id,
gateway=response_gateway_ref(invitation.response_gateway_),
policy=policy,
respondent_id=invitation.respondent_id or respondent_id,
respondent_label=invitation.respondent_label,
email=invitation.email,
response=response_ref,
)
class SqlPollSchedulingProvider(PollSchedulingProvider):
def create_poll(
self,
session: object,
*,
tenant_id: str,
user_id: str | None,
command: PollCreateCommand,
) -> PollRef:
payload = PollCreateRequest(
title=command.title,
description=command.description,
kind=command.kind,
status=command.status,
visibility=command.visibility,
result_visibility=command.result_visibility,
context_module=command.context_module,
context_resource_type=command.context_resource_type,
context_resource_id=command.context_resource_id,
workflow_state=command.workflow_state,
workflow_steps=[dict(step) for step in command.workflow_steps],
allow_anonymous=command.allow_anonymous,
allow_response_update=command.allow_response_update,
min_choices=command.min_choices,
max_choices=command.max_choices,
opens_at=command.opens_at,
closes_at=command.closes_at,
options=[
PollOptionInput(
key=option.key,
label=option.label,
description=option.description,
value=dict(option.value) if option.value is not None else None,
metadata=dict(option.metadata),
)
for option in command.options
],
metadata=dict(command.metadata),
)
try:
poll = create_poll(
session,
tenant_id=tenant_id,
user_id=user_id,
payload=payload,
mutation_owner=_command_owner(command),
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)
def create_invitation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollInvitationCommand,
) -> PollInvitationRef:
payload = PollInvitationCreateRequest(
respondent_id=command.respondent_id,
respondent_label=command.respondent_label,
email=command.email,
expires_at=command.expires_at,
metadata=dict(command.metadata),
)
try:
invitation, token = create_poll_invitation(
session,
tenant_id=tenant_id,
poll_id=poll_id,
payload=payload,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return PollInvitationRef(id=invitation.id, token=token)
def create_governed_invitation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollGovernedInvitationCommand,
) -> PollInvitationRef:
try:
payload = PollInvitationCreateRequest(
respondent_id=command.respondent_id,
respondent_label=command.respondent_label,
email=command.email,
expires_at=command.expires_at,
response_gateway=response_gateway_payload(command.gateway),
participation_policy=participation_policy_payload(command.policy),
metadata=dict(command.metadata),
)
invitation, token = create_poll_invitation(
session,
tenant_id=tenant_id,
poll_id=poll_id,
payload=payload,
governed_capability=True,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
return PollInvitationRef(id=invitation.id, token=token)
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:
try:
invitation = governed_invitation(
session,
token=token,
gateway=gateway,
)
policy = participation_policy_ref(invitation.participation_policy_)
if (
not participant_is_authenticated
and policy.anonymous_password_required
and ANONYMOUS_PASSWORD_REQUIREMENT not in verified_requirements
):
raise PollError("Poll invitation not found")
return _participation_context_ref(
session,
invitation=invitation,
respondent_id=(respondent_id if participant_is_authenticated else None),
participant_email=participant_email,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
def submit_governed_response(
self,
session: object,
*,
token: str,
gateway: PollResponseGatewayRef,
command: PollGovernedResponseCommand,
) -> PollGovernedResponseRef:
try:
_invitation, response, replayed = submit_governed_poll_response(
session,
token=token,
gateway=gateway,
command=command,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
email, comment = response_metadata(response)
return PollGovernedResponseRef(
response=_response_ref(response),
participant_email=email,
comment=comment,
replayed=replayed,
)
def resolve_authenticated_participation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
) -> PollParticipationContextRef:
try:
invitation = governed_invitation_by_id(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
gateway=gateway,
respondent_id=respondent_id,
)
return _participation_context_ref(
session,
invitation=invitation,
respondent_id=respondent_id,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
def submit_authenticated_response(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
command: PollGovernedResponseCommand,
) -> PollGovernedResponseRef:
try:
_invitation, response, replayed = submit_authenticated_poll_response(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
gateway=gateway,
respondent_id=respondent_id,
command=command,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
email, comment = response_metadata(response)
return PollGovernedResponseRef(
response=_response_ref(response),
participant_email=email,
comment=comment,
replayed=replayed,
)
def add_option(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollOptionRequest,
) -> PollOptionMutationRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
option, replayed = add_poll_option(
session,
tenant_id=tenant_id,
poll_id=poll_id,
option=PollOptionInput(
key=command.key,
label=command.label,
description=command.description,
value=dict(command.value) if command.value is not None else None,
metadata=dict(command.metadata),
),
mutation_owner=mutation_owner,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
return PollOptionMutationRef(
id=option.id,
position=option.position,
replayed=replayed,
)
def remove_option(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
option_id: str,
) -> PollOptionMutationRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
option, replayed, invalidated = remove_poll_option(
session,
tenant_id=tenant_id,
poll_id=poll_id,
option_id=option_id,
mutation_owner=mutation_owner,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
return PollOptionMutationRef(
id=option.id,
position=option.position,
replayed=replayed,
invalidated_response_count=invalidated,
)
def revoke_invitation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
) -> PollInvitationRevocationRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
invitation, replayed = revoke_poll_invitation_with_replay(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
mutation_owner=mutation_owner,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
return PollInvitationRevocationRef(
id=invitation.id,
revoked_at=response_datetime(invitation.revoked_at),
replayed=replayed,
)
def update_invitation_expiry(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
expires_at: datetime | None,
) -> PollInvitationExpiryRef:
try:
invitation, replayed = update_governed_invitation_expiry(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
gateway=gateway,
expires_at=expires_at,
)
except (PollError, ValidationError) as exc:
raise PollCapabilityError(str(exc)) from exc
return PollInvitationExpiryRef(
id=invitation.id,
expires_at=response_datetime(invitation.expires_at),
replayed=replayed,
)
def submit_response(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollSubmitResponseCommand,
) -> PollResponseRef:
payload = PollSubmitResponseRequest(
respondent_id=command.respondent_id,
respondent_label=command.respondent_label,
answers=[
PollAnswerInput(
option_id=answer.option_id,
option_key=answer.option_key,
value=answer.value,
rank=answer.rank,
)
for answer in command.answers
],
metadata=dict(command.metadata),
)
try:
response = submit_poll_response(
session,
tenant_id=tenant_id,
poll_id=poll_id,
payload=payload,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _response_ref(response)
def update_poll(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollUpdateCommand,
) -> PollRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
poll = update_poll_snapshot(
session,
tenant_id=tenant_id,
poll_id=poll_id,
title=command.title,
description=command.description,
visibility=command.visibility,
result_visibility=command.result_visibility,
allow_anonymous=command.allow_anonymous,
allow_response_update=command.allow_response_update,
closes_at=command.closes_at,
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)
def update_option(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
option_id: str,
command: PollOptionUpdateCommand,
) -> PollOptionRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
option = update_poll_option(
session,
tenant_id=tenant_id,
poll_id=poll_id,
option_id=option_id,
label=command.label,
description=command.description,
value=dict(command.value) if command.value is not None else None,
metadata=dict(command.metadata),
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return PollOptionRef(id=option.id, position=option.position)
def reorder_options(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollOptionOrderCommand,
) -> PollRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
poll = reorder_poll_options(
session,
tenant_id=tenant_id,
poll_id=poll_id,
option_ids=command.option_ids,
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)
def open_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
return self._transition(open_poll, session, tenant_id=tenant_id, poll_id=poll_id)
def close_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
return self._transition(close_poll, session, tenant_id=tenant_id, poll_id=poll_id)
def decide_poll(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
option_id: str | None,
) -> PollRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
poll = decide_poll(
session,
tenant_id=tenant_id,
poll_id=poll_id,
payload=PollDecisionRequest(option_id=option_id),
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)
def get_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
try:
return _poll_ref(get_poll(session, tenant_id=tenant_id, poll_id=poll_id))
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
def set_workflow_context(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
workflow_state: str,
workflow_steps: Sequence[Mapping[str, object]],
context_module: str,
context_resource_type: str,
context_resource_id: str,
) -> PollRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
requested_owner = poll_owner_ref(
module_id=context_module,
resource_type=context_resource_type,
resource_id=context_resource_id,
)
poll = set_poll_workflow_context(
session,
tenant_id=tenant_id,
poll_id=poll_id,
workflow_state=workflow_state,
workflow_steps=[dict(step) for step in workflow_steps],
context_module=context_module,
context_resource_type=context_resource_type,
context_resource_id=context_resource_id,
mutation_owner=mutation_owner,
requested_owner=requested_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)
def result_summary(self, session: object, *, tenant_id: str, poll_id: str) -> Mapping[str, object]:
try:
return poll_result_summary_by_id(session, tenant_id=tenant_id, poll_id=poll_id)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
def list_responses(self, session: object, *, tenant_id: str, poll_id: str) -> tuple[PollResponseRef, ...]:
try:
projection_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
responses = list_poll_responses(
session,
tenant_id=tenant_id,
poll_id=poll_id,
projection_owner=projection_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return tuple(_response_ref(response) for response in responses)
def get_response(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
respondent_ids: Sequence[str],
invitation_id: str | None = None,
) -> PollResponseRef | None:
try:
projection_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
response = get_poll_response_for_respondents(
session,
tenant_id=tenant_id,
poll_id=poll_id,
respondent_ids=tuple(respondent_ids),
invitation_id=invitation_id,
projection_owner=projection_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _response_ref(response) if response is not None else None
def retire_responses(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollResponseRetirementCommand,
) -> PollResponseRetirementRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
responses, retired_at, retired_count, replayed = retire_poll_responses(
session,
tenant_id=tenant_id,
poll_id=poll_id,
respondent_ids=command.respondent_ids,
invitation_id=command.invitation_id,
reason=command.reason,
idempotency_key=command.idempotency_key,
metadata=dict(command.metadata),
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return PollResponseRetirementRef(
response_ids=tuple(response.id for response in responses),
retired_at=response_datetime(retired_at),
newly_retired_count=retired_count,
replayed=replayed,
)
@staticmethod
def _transition(callback, session: object, *, tenant_id: str, poll_id: str) -> PollRef:
try:
mutation_owner = _stored_owner(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
poll = callback(
session,
tenant_id=tenant_id,
poll_id=poll_id,
mutation_owner=mutation_owner,
)
except PollError as exc:
raise PollCapabilityError(str(exc)) from exc
return _poll_ref(poll)

View File

@@ -1,5 +1,5 @@
from __future__ import annotations
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
__all__ = ["Poll", "PollInvitation", "PollOption", "PollResponse"]
__all__ = ["Poll", "PollInvitation", "PollLifecycleTransition", "PollOption", "PollResponse"]

View File

@@ -42,6 +42,11 @@ class Poll(Base, TimestampMixin):
context_module: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
context_resource_type: Mapped[str | None] = mapped_column(String(100), nullable=True)
context_resource_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
participation_gateway_: Mapped[dict[str, Any] | None] = mapped_column(
"participation_gateway",
JSON,
nullable=True,
)
workflow_state: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
workflow_steps: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
allow_anonymous: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
@@ -53,6 +58,7 @@ class Poll(Base, TimestampMixin):
closed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
decided_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
decided_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
archived_from_status: Mapped[str | None] = mapped_column(String(30), nullable=True)
created_by_user_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
@@ -60,6 +66,11 @@ class Poll(Base, TimestampMixin):
options: Mapped[list["PollOption"]] = relationship(back_populates="poll", cascade="all, delete-orphan", order_by="PollOption.position")
responses: Mapped[list["PollResponse"]] = relationship(back_populates="poll", cascade="all, delete-orphan")
invitations: Mapped[list["PollInvitation"]] = relationship(back_populates="poll", cascade="all, delete-orphan")
lifecycle_transitions: Mapped[list["PollLifecycleTransition"]] = relationship(
back_populates="poll",
cascade="all, delete-orphan",
order_by="PollLifecycleTransition.created_at",
)
class PollOption(Base, TimestampMixin):
@@ -88,6 +99,18 @@ class PollResponse(Base, TimestampMixin):
__table_args__ = (
Index("ix_poll_responses_poll_submitted", "poll_id", "submitted_at"),
Index("ix_poll_responses_poll_respondent", "poll_id", "respondent_id"),
Index(
"uq_poll_responses_active_respondent",
"poll_id",
"respondent_id",
unique=True,
sqlite_where=text(
"deleted_at IS NULL AND respondent_id IS NOT NULL"
),
postgresql_where=text(
"deleted_at IS NULL AND respondent_id IS NOT NULL"
),
),
Index("ix_poll_responses_tenant_poll", "tenant_id", "poll_id"),
)
@@ -122,9 +145,80 @@ class PollInvitation(Base, TimestampMixin):
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
response_gateway_: Mapped[dict[str, Any] | None] = mapped_column("response_gateway", JSON, nullable=True)
participation_policy_: Mapped[dict[str, Any] | None] = mapped_column("participation_policy", JSON, nullable=True)
metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
poll: Mapped[Poll] = relationship(back_populates="invitations")
__all__ = ["Poll", "PollInvitation", "PollOption", "PollResponse", "new_uuid"]
class PollParticipationSubmission(Base, TimestampMixin):
__tablename__ = "poll_participation_submissions"
__table_args__ = (
UniqueConstraint(
"invitation_id",
"idempotency_key",
name="uq_poll_participation_submission_idempotency",
),
Index(
"ix_poll_participation_submission_poll_created",
"tenant_id",
"poll_id",
"created_at",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
poll_id: Mapped[str] = mapped_column(
ForeignKey("poll_polls.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
invitation_id: Mapped[str] = mapped_column(
ForeignKey("poll_invitations.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
response_id: Mapped[str] = mapped_column(
ForeignKey("poll_responses.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
request_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
class PollLifecycleTransition(Base, TimestampMixin):
__tablename__ = "poll_lifecycle_transitions"
__table_args__ = (
UniqueConstraint("poll_id", "idempotency_key", name="uq_poll_lifecycle_transition_idempotency"),
Index("ix_poll_lifecycle_transition_poll_created", "poll_id", "created_at"),
Index("ix_poll_lifecycle_transition_tenant_poll", "tenant_id", "poll_id"),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
poll_id: Mapped[str] = mapped_column(ForeignKey("poll_polls.id", ondelete="CASCADE"), nullable=False, index=True)
action: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
from_status: Mapped[str] = mapped_column(String(30), nullable=False)
to_status: Mapped[str] = mapped_column(String(30), nullable=False)
decision_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
previous_decision_option_id: Mapped[str | None] = mapped_column(String(36), nullable=True)
idempotency_key: Mapped[str | None] = mapped_column(String(255), nullable=True)
actor_user_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
actor_api_key_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
metadata_: Mapped[dict[str, Any] | None] = mapped_column("metadata", JSON, nullable=True)
poll: Mapped[Poll] = relationship(back_populates="lifecycle_transitions")
__all__ = [
"Poll",
"PollInvitation",
"PollLifecycleTransition",
"PollOption",
"PollParticipationSubmission",
"PollResponse",
"new_uuid",
]

View File

@@ -13,12 +13,14 @@ from govoplan_core.core.modules import (
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
from govoplan_core.db.base import Base
from govoplan_poll.backend.db import models as poll_models # noqa: F401 - populate Poll ORM metadata
MODULE_ID = "poll"
MODULE_NAME = "Poll"
MODULE_VERSION = "0.1.8"
MODULE_VERSION = "0.1.11"
READ_SCOPE = "poll:poll:read"
WRITE_SCOPE = "poll:poll:write"
ADMIN_SCOPE = "poll:poll:admin"
@@ -99,6 +101,17 @@ def _poll_router(_context: ModuleContext):
return router
def _poll_scheduling_provider(context: ModuleContext) -> object:
del context
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
return SqlPollSchedulingProvider()
def _poll_participation_gateway_provider(context: ModuleContext) -> object:
return _poll_scheduling_provider(context)
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
@@ -107,16 +120,22 @@ manifest = ModuleManifest(
optional_dependencies=("access", "calendar", "campaigns", "portal", "mail", "notifications", "forms_runtime"),
optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
provides_interfaces=(
ModuleInterfaceProvider(name="poll.option_selection", version="0.1.8"),
ModuleInterfaceProvider(name="poll.availability_matrix", version="0.1.8"),
ModuleInterfaceProvider(name="poll.response_collection", version="0.1.8"),
ModuleInterfaceProvider(name="poll.workflow_context", version="0.1.8"),
ModuleInterfaceProvider(name="poll.signed_participation", version="0.1.8"),
ModuleInterfaceProvider(name="poll.option_selection", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.option_ordering", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.availability_matrix", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.response_collection", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.workflow_context", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.signed_participation", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.governed_participation", version=MODULE_VERSION),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_poll_router,
tenant_summary_providers=(_tenant_summary,),
capability_factories={
CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider,
CAPABILITY_POLL_PARTICIPATION_GATEWAY: _poll_participation_gateway_provider,
},
migration_spec=MigrationSpec(
module_id=MODULE_ID,
metadata=Base.metadata,
@@ -125,7 +144,9 @@ manifest = ModuleManifest(
retirement_provider=drop_table_retirement_provider(
poll_models.Poll,
poll_models.PollInvitation,
poll_models.PollLifecycleTransition,
poll_models.PollOption,
poll_models.PollParticipationSubmission,
poll_models.PollResponse,
label="Poll",
),
@@ -135,7 +156,9 @@ manifest = ModuleManifest(
persistent_table_uninstall_guard(
poll_models.Poll,
poll_models.PollInvitation,
poll_models.PollLifecycleTransition,
poll_models.PollOption,
poll_models.PollParticipationSubmission,
poll_models.PollResponse,
label="Poll",
),

View File

@@ -0,0 +1,67 @@
"""v0.1.9 configurable Poll lifecycle history
Revision ID: 3b4c5d6e7f8a
Revises: 2a3b4c5d6e7f
Create Date: 2026-07-20 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "3b4c5d6e7f8a"
down_revision = "2a3b4c5d6e7f"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column("poll_polls", sa.Column("archived_from_status", sa.String(length=30), nullable=True))
op.create_table(
"poll_lifecycle_transitions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("poll_id", sa.String(length=36), nullable=False),
sa.Column("action", sa.String(length=30), nullable=False),
sa.Column("from_status", sa.String(length=30), nullable=False),
sa.Column("to_status", sa.String(length=30), nullable=False),
sa.Column("decision_option_id", sa.String(length=36), nullable=True),
sa.Column("previous_decision_option_id", sa.String(length=36), nullable=True),
sa.Column("idempotency_key", sa.String(length=255), nullable=True),
sa.Column("actor_user_id", sa.String(length=36), nullable=True),
sa.Column("actor_api_key_id", sa.String(length=36), nullable=True),
sa.Column("metadata", sa.JSON(), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["poll_id"],
["poll_polls.id"],
name=op.f("fk_poll_lifecycle_transitions_poll_id_poll_polls"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_poll_lifecycle_transitions")),
sa.UniqueConstraint("poll_id", "idempotency_key", name="uq_poll_lifecycle_transition_idempotency"),
)
op.create_index(op.f("ix_poll_lifecycle_transitions_action"), "poll_lifecycle_transitions", ["action"], unique=False)
op.create_index(op.f("ix_poll_lifecycle_transitions_actor_api_key_id"), "poll_lifecycle_transitions", ["actor_api_key_id"], unique=False)
op.create_index(op.f("ix_poll_lifecycle_transitions_actor_user_id"), "poll_lifecycle_transitions", ["actor_user_id"], unique=False)
op.create_index(op.f("ix_poll_lifecycle_transitions_poll_id"), "poll_lifecycle_transitions", ["poll_id"], unique=False)
op.create_index(op.f("ix_poll_lifecycle_transitions_tenant_id"), "poll_lifecycle_transitions", ["tenant_id"], unique=False)
op.create_index(
"ix_poll_lifecycle_transition_poll_created",
"poll_lifecycle_transitions",
["poll_id", "created_at"],
unique=False,
)
op.create_index(
"ix_poll_lifecycle_transition_tenant_poll",
"poll_lifecycle_transitions",
["tenant_id", "poll_id"],
unique=False,
)
def downgrade() -> None:
op.drop_table("poll_lifecycle_transitions")
op.drop_column("poll_polls", "archived_from_status")

View File

@@ -0,0 +1,104 @@
"""v0.1.10 governed signed participation
Revision ID: 4c5d6e7f8a9b
Revises: 3b4c5d6e7f8a
Create Date: 2026-07-21 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "4c5d6e7f8a9b"
down_revision = "3b4c5d6e7f8a"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.add_column(
"poll_invitations",
sa.Column("response_gateway", sa.JSON(), nullable=True),
)
op.add_column(
"poll_invitations",
sa.Column("participation_policy", sa.JSON(), nullable=True),
)
op.create_table(
"poll_participation_submissions",
sa.Column("id", sa.String(length=36), nullable=False),
sa.Column("tenant_id", sa.String(length=36), nullable=False),
sa.Column("poll_id", sa.String(length=36), nullable=False),
sa.Column("invitation_id", sa.String(length=36), nullable=False),
sa.Column("response_id", sa.String(length=36), nullable=False),
sa.Column("idempotency_key", sa.String(length=255), nullable=False),
sa.Column("request_fingerprint", sa.String(length=64), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(
["poll_id"],
["poll_polls.id"],
name=op.f("fk_poll_participation_submissions_poll_id_poll_polls"),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["invitation_id"],
["poll_invitations.id"],
name=op.f(
"fk_poll_participation_submissions_invitation_id_poll_invitations"
),
ondelete="CASCADE",
),
sa.ForeignKeyConstraint(
["response_id"],
["poll_responses.id"],
name=op.f("fk_poll_participation_submissions_response_id_poll_responses"),
ondelete="CASCADE",
),
sa.PrimaryKeyConstraint(
"id",
name=op.f("pk_poll_participation_submissions"),
),
sa.UniqueConstraint(
"invitation_id",
"idempotency_key",
name="uq_poll_participation_submission_idempotency",
),
)
op.create_index(
op.f("ix_poll_participation_submissions_tenant_id"),
"poll_participation_submissions",
["tenant_id"],
unique=False,
)
op.create_index(
op.f("ix_poll_participation_submissions_poll_id"),
"poll_participation_submissions",
["poll_id"],
unique=False,
)
op.create_index(
op.f("ix_poll_participation_submissions_invitation_id"),
"poll_participation_submissions",
["invitation_id"],
unique=False,
)
op.create_index(
op.f("ix_poll_participation_submissions_response_id"),
"poll_participation_submissions",
["response_id"],
unique=False,
)
op.create_index(
"ix_poll_participation_submission_poll_created",
"poll_participation_submissions",
["tenant_id", "poll_id", "created_at"],
unique=False,
)
def downgrade() -> None:
op.drop_table("poll_participation_submissions")
op.drop_column("poll_invitations", "participation_policy")
op.drop_column("poll_invitations", "response_gateway")

View File

@@ -0,0 +1,64 @@
"""v0.1.10 durable Poll participation owner
Revision ID: 5d6e7f8a9b0c
Revises: 4c5d6e7f8a9b
Create Date: 2026-07-21 00:00:01.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "5d6e7f8a9b0c"
down_revision = "4c5d6e7f8a9b"
branch_labels = None
depends_on = None
def _backfill_existing_governed_polls() -> None:
"""Retain the owner of installations that already used revision 4c5."""
polls = sa.table(
"poll_polls",
sa.column("id", sa.String(length=36)),
sa.column("participation_gateway", sa.JSON()),
)
invitations = sa.table(
"poll_invitations",
sa.column("id", sa.String(length=36)),
sa.column("poll_id", sa.String(length=36)),
sa.column("response_gateway", sa.JSON()),
)
connection = op.get_bind()
rows = connection.execute(
sa.select(
invitations.c.poll_id,
invitations.c.response_gateway,
).order_by(invitations.c.poll_id.asc(), invitations.c.id.asc())
).mappings()
claimed_poll_ids: set[str] = set()
for row in rows:
poll_id = row["poll_id"]
gateway = row["response_gateway"]
if poll_id in claimed_poll_ids or not isinstance(gateway, dict):
continue
connection.execute(
polls.update()
.where(polls.c.id == poll_id)
.where(polls.c.participation_gateway.is_(None))
.values(participation_gateway=gateway)
)
claimed_poll_ids.add(poll_id)
def upgrade() -> None:
op.add_column(
"poll_polls",
sa.Column("participation_gateway", sa.JSON(), nullable=True),
)
_backfill_existing_governed_polls()
def downgrade() -> None:
op.drop_column("poll_polls", "participation_gateway")

View File

@@ -0,0 +1,99 @@
"""v0.1.11 active identified Poll response uniqueness
Revision ID: 6e7f8a9b0c1d
Revises: 5d6e7f8a9b0c
Create Date: 2026-07-22 00:00:00.000000
This migration deduplicates active identified responses before creating the
partial unique index. Run it while Poll response writes are quiesced or while
the platform maintenance lock is held. PostgreSQL additionally takes a table
lock so an unexpected writer cannot enter between cleanup and index creation.
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "6e7f8a9b0c1d"
down_revision = "5d6e7f8a9b0c"
branch_labels = None
depends_on = None
_INDEX_NAME = "uq_poll_responses_active_respondent"
_ACTIVE_IDENTIFIED_PREDICATE = (
"deleted_at IS NULL AND respondent_id IS NOT NULL"
)
def _tombstone_duplicate_active_responses() -> None:
responses = sa.table(
"poll_responses",
sa.column("id", sa.String(length=36)),
sa.column("poll_id", sa.String(length=36)),
sa.column("respondent_id", sa.String(length=255)),
sa.column("submitted_at", sa.DateTime(timezone=True)),
sa.column("deleted_at", sa.DateTime(timezone=True)),
sa.column("updated_at", sa.DateTime(timezone=True)),
)
ranked = (
sa.select(
responses.c.id.label("id"),
sa.func.row_number()
.over(
partition_by=(
responses.c.poll_id,
responses.c.respondent_id,
),
order_by=(
responses.c.submitted_at.desc(),
responses.c.id.desc(),
),
)
.label("response_rank"),
)
.where(
responses.c.deleted_at.is_(None),
responses.c.respondent_id.is_not(None),
)
.subquery("ranked_active_poll_responses")
)
duplicate_ids = sa.select(ranked.c.id).where(ranked.c.response_rank > 1)
migration_timestamp = sa.func.current_timestamp()
op.get_bind().execute(
responses.update()
.where(
responses.c.id.in_(duplicate_ids),
responses.c.deleted_at.is_(None),
)
.values(
deleted_at=migration_timestamp,
updated_at=migration_timestamp,
)
)
def upgrade() -> None:
connection = op.get_bind()
if connection.dialect.name == "postgresql":
connection.execute(
sa.text(
"LOCK TABLE poll_responses IN SHARE ROW EXCLUSIVE MODE"
)
)
_tombstone_duplicate_active_responses()
op.create_index(
_INDEX_NAME,
"poll_responses",
["poll_id", "respondent_id"],
unique=True,
sqlite_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
postgresql_where=sa.text(_ACTIVE_IDENTIFIED_PREDICATE),
)
def downgrade() -> None:
# Deduplicated rows remain tombstoned: a downgrade must not silently revive
# responses that were no longer part of the active result set.
op.drop_index(_INDEX_NAME, table_name="poll_responses")

View File

@@ -0,0 +1,43 @@
"""Backward-compatible governed-participation contract imports.
The provider contract belongs to Core so consumers can resolve Poll through
the platform capability registry without importing Poll implementation
internals. This module intentionally preserves the original public import
path for existing Poll integrations.
"""
from govoplan_core.core.poll_participation import (
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,
)
__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",
]

View File

@@ -0,0 +1,739 @@
from __future__ import annotations
import hashlib
import json
from collections.abc import Mapping
from datetime import datetime
from typing import Any
from pydantic import ValidationError
from sqlalchemy.orm import Session
from govoplan_poll.backend.db.models import (
Poll,
PollInvitation,
PollParticipationSubmission,
PollResponse,
)
from govoplan_poll.backend.participation import (
ANONYMOUS_PASSWORD_REQUIREMENT,
PollGovernedResponseCommand,
PollParticipationPolicy,
PollResponseGatewayRef,
)
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollParticipationPolicyInput,
PollResponseGatewayInput,
PollSubmitResponseRequest,
)
from govoplan_poll.backend.service import (
PollError,
_assert_governed_invitation_gateway_allowed,
_assert_poll_accepts_responses,
_existing_response,
_insert_or_reconcile_identified_response,
_lock_poll_for_response,
_now,
_update_existing_response,
assert_no_sensitive_participation_metadata,
get_poll_invitation_by_token,
normalize_response_answers,
response_datetime,
)
GENERIC_PARTICIPATION_ERROR = "Poll invitation not found"
MAX_COMMENT_LENGTH = 4_000
def response_gateway_payload(gateway: PollResponseGatewayRef) -> dict[str, str]:
try:
return PollResponseGatewayInput(
module_id=gateway.module_id,
resource_type=gateway.resource_type,
resource_id=gateway.resource_id,
).model_dump(mode="json")
except ValidationError as exc:
raise PollError("Invalid participation response gateway") from exc
def participation_policy_payload(policy: PollParticipationPolicy) -> dict[str, Any]:
try:
return PollParticipationPolicyInput(
version=policy.version,
single_choice=policy.single_choice,
allow_maybe=policy.allow_maybe,
max_participants_per_option=policy.max_participants_per_option,
allow_comments=policy.allow_comments,
participant_email_required=policy.participant_email_required,
anonymous_password_required=policy.anonymous_password_required,
).model_dump(mode="json")
except ValidationError as exc:
raise PollError("Invalid participation policy") from exc
def response_gateway_ref(value: Mapping[str, Any] | None) -> PollResponseGatewayRef:
try:
parsed = PollResponseGatewayInput.model_validate(value)
except ValidationError as exc:
raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
return PollResponseGatewayRef(
module_id=parsed.module_id,
resource_type=parsed.resource_type,
resource_id=parsed.resource_id,
)
def participation_policy_ref(value: Mapping[str, Any] | None) -> PollParticipationPolicy:
try:
parsed = PollParticipationPolicyInput.model_validate(value)
except ValidationError as exc:
raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
return PollParticipationPolicy(
version=parsed.version,
single_choice=parsed.single_choice,
allow_maybe=parsed.allow_maybe,
max_participants_per_option=parsed.max_participants_per_option,
allow_comments=parsed.allow_comments,
participant_email_required=parsed.participant_email_required,
anonymous_password_required=parsed.anonymous_password_required,
)
def governed_invitation(
session: Session,
*,
token: str,
gateway: PollResponseGatewayRef,
lock: bool = False,
) -> PollInvitation:
"""Resolve a gateway-bound token without revealing which check failed."""
try:
invitation = get_poll_invitation_by_token(session, token=token)
_assert_governed_invitation(invitation, gateway=gateway)
if lock:
invitation = (
session.query(PollInvitation)
.filter(PollInvitation.id == invitation.id)
.populate_existing()
.with_for_update()
.one_or_none()
)
_assert_governed_invitation(invitation, gateway=gateway)
except (PollError, ValidationError) as exc:
raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
if invitation is None:
raise PollError(GENERIC_PARTICIPATION_ERROR)
return invitation
def _assert_governed_invitation(
invitation: PollInvitation | None,
*,
gateway: PollResponseGatewayRef,
authenticated_respondent_id: str | None = None,
) -> None:
if invitation is None or invitation.poll.deleted_at is not None:
raise PollError(GENERIC_PARTICIPATION_ERROR)
if invitation.revoked_at is not None:
raise PollError(GENERIC_PARTICIPATION_ERROR)
if (
invitation.expires_at is not None
and response_datetime(invitation.expires_at) <= _now()
):
raise PollError(GENERIC_PARTICIPATION_ERROR)
_assert_governed_invitation_gateway_allowed(
poll=invitation.poll,
gateway=response_gateway_payload(gateway),
)
if response_gateway_ref(invitation.response_gateway_) != gateway:
raise PollError(GENERIC_PARTICIPATION_ERROR)
participation_policy_ref(invitation.participation_policy_)
if authenticated_respondent_id is not None and (
invitation.respondent_id is not None
and invitation.respondent_id != authenticated_respondent_id
):
raise PollError(GENERIC_PARTICIPATION_ERROR)
def governed_invitation_by_id(
session: Session,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
lock: bool = False,
) -> PollInvitation:
"""Resolve an active governed invitation for one authenticated identity."""
if not respondent_id:
raise PollError(GENERIC_PARTICIPATION_ERROR)
try:
query = session.query(PollInvitation).filter(
PollInvitation.tenant_id == tenant_id,
PollInvitation.poll_id == poll_id,
PollInvitation.id == invitation_id,
)
if lock:
query = query.populate_existing().with_for_update()
invitation = query.one_or_none()
_assert_governed_invitation(
invitation,
gateway=gateway,
authenticated_respondent_id=respondent_id,
)
except PollError as exc:
raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
if invitation is None:
raise PollError(GENERIC_PARTICIPATION_ERROR)
return invitation
def update_governed_invitation_expiry(
session: Session,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
expires_at: datetime | None,
) -> tuple[PollInvitation, bool]:
"""Update expiry without rotating the bearer token or reviving revocation."""
poll = _lock_poll_for_response(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
invitation = (
session.query(PollInvitation)
.filter(
PollInvitation.tenant_id == tenant_id,
PollInvitation.poll_id == poll.id,
PollInvitation.id == invitation_id,
)
.populate_existing()
.with_for_update()
.one_or_none()
)
try:
if invitation is None or invitation.revoked_at is not None:
raise PollError(GENERIC_PARTICIPATION_ERROR)
_assert_governed_invitation_gateway_allowed(
poll=poll,
gateway=response_gateway_payload(gateway),
)
if response_gateway_ref(invitation.response_gateway_) != gateway:
raise PollError(GENERIC_PARTICIPATION_ERROR)
participation_policy_ref(invitation.participation_policy_)
except (PollError, ValidationError) as exc:
raise PollError(GENERIC_PARTICIPATION_ERROR) from exc
normalized_expiry = response_datetime(expires_at)
replayed = response_datetime(invitation.expires_at) == normalized_expiry
if not replayed:
invitation.expires_at = normalized_expiry
session.flush()
return invitation, replayed
def _normalize_email(value: str | None) -> str | None:
if value is None:
return None
normalized = value.strip().casefold()
if not normalized:
return None
if (
normalized.count("@") != 1
or any(character.isspace() for character in normalized)
or not all(normalized.split("@", 1))
):
raise PollError("Participant email is invalid")
return normalized
def _canonical_respondent_id(
invitation: PollInvitation,
*,
respondent_id: str | None,
participant_email: str | None,
participant_is_authenticated: bool,
) -> str:
if invitation.respondent_id:
return invitation.respondent_id
if participant_is_authenticated and respondent_id:
return respondent_id
if participant_email and invitation.email is None:
email_fingerprint = hashlib.sha256(participant_email.encode("utf-8")).hexdigest()
return f"invitation:{invitation.id}:email:{email_fingerprint}"
return f"invitation:{invitation.id}"
def _normalize_comment(value: str | None) -> str | None:
if value is None:
return None
normalized = value.strip()
if len(normalized) > MAX_COMMENT_LENGTH:
raise PollError(f"Participation comments cannot exceed {MAX_COMMENT_LENGTH} characters")
return normalized or None
def _answer_reserves_capacity(poll: Poll, answer: Mapping[str, Any]) -> bool:
if poll.kind == "availability":
return answer.get("value") == "available"
return True
def _validate_gateway_rules(
*,
invitation: PollInvitation,
policy: PollParticipationPolicy,
command: PollGovernedResponseCommand,
) -> tuple[str | None, str | None, list[PollAnswerInput]]:
normalized_email = _normalize_email(command.participant_email or invitation.email)
invitation_email = _normalize_email(invitation.email)
if invitation_email is not None and normalized_email != invitation_email:
raise PollError("Participant email does not match this invitation")
if (
not command.participant_is_authenticated
and policy.participant_email_required
and normalized_email is None
):
raise PollError("Participant email is required")
if (
not command.participant_is_authenticated
and policy.anonymous_password_required
and ANONYMOUS_PASSWORD_REQUIREMENT not in command.verified_requirements
):
raise PollError("Anonymous password verification is required")
comment = _normalize_comment(command.comment)
if comment is not None and not policy.allow_comments:
raise PollError("Comments are not enabled for this participation link")
answers = [
PollAnswerInput(
option_id=answer.option_id,
option_key=answer.option_key,
value=answer.value,
rank=answer.rank,
)
for answer in command.answers
]
return normalized_email, comment, answers
def _validate_normalized_gateway_answers(
*,
poll: Poll,
policy: PollParticipationPolicy,
normalized_answers: list[dict[str, Any]],
) -> None:
"""Enforce policy against Poll-resolved option identities and values."""
if not policy.allow_maybe and any(
answer.get("value") == "maybe" or answer.get("option_key") == "maybe"
for answer in normalized_answers
):
raise PollError("Maybe responses are not enabled for this participation link")
selected_count = sum(
(
answer.get("value") != "unavailable"
if poll.kind == "availability"
else True
)
for answer in normalized_answers
)
if policy.single_choice and selected_count > 1:
raise PollError("Only one poll option may be selected")
def _selected_capacity_option_ids(
poll: Poll,
normalized_answers: list[dict[str, Any]],
) -> set[str]:
return {
str(answer["option_id"])
for answer in normalized_answers
if answer.get("option_id") and _answer_reserves_capacity(poll, answer)
}
def _enforce_capacity(
session: Session,
*,
poll: Poll,
existing: PollResponse | None,
normalized_answers: list[dict[str, Any]],
limit: int | None,
) -> None:
if limit is None:
return
selected = _selected_capacity_option_ids(poll, normalized_answers)
if not selected:
return
counts = dict.fromkeys(selected, 0)
responses = (
session.query(PollResponse)
.filter(
PollResponse.tenant_id == poll.tenant_id,
PollResponse.poll_id == poll.id,
PollResponse.deleted_at.is_(None),
)
.order_by(PollResponse.id.asc())
.all()
)
for response in responses:
if existing is not None and response.id == existing.id:
continue
for answer in response.answers or []:
if (
isinstance(answer, Mapping)
and answer.get("option_id") in counts
and _answer_reserves_capacity(poll, answer)
):
counts[str(answer["option_id"])] += 1
full = {option_id for option_id, count in counts.items() if count >= limit}
if full:
raise PollError("Participant limit reached for one or more poll options")
def _normalize_idempotency_key(value: str | None) -> str | None:
if value is None:
return None
normalized = value.strip()
if not normalized:
raise PollError("Idempotency key cannot be empty")
if len(normalized) > 255:
raise PollError("Idempotency key cannot be longer than 255 characters")
return normalized
def _submission_fingerprint(
*,
gateway: PollResponseGatewayRef,
command: PollGovernedResponseCommand,
normalized_email: str | None,
comment: str | None,
) -> str:
value = {
"gateway": response_gateway_payload(gateway),
"respondent_id": command.respondent_id,
"respondent_label": command.respondent_label,
"participant_email": normalized_email,
"participant_is_authenticated": command.participant_is_authenticated,
"answers": [
{
"option_id": answer.option_id,
"option_key": answer.option_key,
"value": answer.value,
"rank": answer.rank,
}
for answer in command.answers
],
"comment": comment,
"verified_requirements": sorted(command.verified_requirements),
"metadata": dict(command.metadata),
}
encoded = json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
default=str,
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _idempotent_submission(
session: Session,
*,
invitation_id: str,
idempotency_key: str | None,
request_fingerprint: str,
) -> PollResponse | None:
if idempotency_key is None:
return None
submission = (
session.query(PollParticipationSubmission)
.filter(
PollParticipationSubmission.invitation_id == invitation_id,
PollParticipationSubmission.idempotency_key == idempotency_key,
)
.one_or_none()
)
if submission is None:
return None
if submission.request_fingerprint != request_fingerprint:
raise PollError("Idempotency key was already used for a different response")
response = (
session.query(PollResponse)
.filter(PollResponse.id == submission.response_id)
.one_or_none()
)
if response is None:
raise PollError("Idempotent response record is no longer available")
return response
def submit_governed_poll_response(
session: Session,
*,
token: str,
gateway: PollResponseGatewayRef,
command: PollGovernedResponseCommand,
) -> tuple[PollInvitation, PollResponse, bool]:
"""Submit through a bound gateway and enforce its snapshot atomically."""
initial = governed_invitation(session, token=token, gateway=gateway)
poll = _lock_poll_for_response(
session,
tenant_id=initial.tenant_id,
poll_id=initial.poll_id,
)
invitation = governed_invitation(session, token=token, gateway=gateway, lock=True)
if invitation.poll_id != poll.id:
raise PollError(GENERIC_PARTICIPATION_ERROR)
return _submit_locked_governed_response(
session,
poll=poll,
invitation=invitation,
gateway=gateway,
command=command,
)
def submit_authenticated_poll_response(
session: Session,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
command: PollGovernedResponseCommand,
) -> tuple[PollInvitation, PollResponse, bool]:
"""Submit by governed invitation id for one authenticated respondent."""
if (
not command.participant_is_authenticated
or not respondent_id
or command.respondent_id != respondent_id
):
raise PollError(GENERIC_PARTICIPATION_ERROR)
initial = governed_invitation_by_id(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
gateway=gateway,
respondent_id=respondent_id,
)
poll = _lock_poll_for_response(
session,
tenant_id=tenant_id,
poll_id=poll_id,
)
invitation = governed_invitation_by_id(
session,
tenant_id=tenant_id,
poll_id=poll_id,
invitation_id=invitation_id,
gateway=gateway,
respondent_id=respondent_id,
lock=True,
)
if initial.id != invitation.id or invitation.poll_id != poll.id:
raise PollError(GENERIC_PARTICIPATION_ERROR)
return _submit_locked_governed_response(
session,
poll=poll,
invitation=invitation,
gateway=gateway,
command=command,
)
def _submit_locked_governed_response(
session: Session,
*,
poll: Poll,
invitation: PollInvitation,
gateway: PollResponseGatewayRef,
command: PollGovernedResponseCommand,
) -> tuple[PollInvitation, PollResponse, bool]:
policy = participation_policy_ref(invitation.participation_policy_)
assert_no_sensitive_participation_metadata(command.metadata)
normalized_email, comment, answer_inputs = _validate_gateway_rules(
invitation=invitation,
policy=policy,
command=command,
)
respondent_id = _canonical_respondent_id(
invitation,
respondent_id=command.respondent_id,
participant_email=normalized_email,
participant_is_authenticated=command.participant_is_authenticated,
)
idempotency_key = _normalize_idempotency_key(command.idempotency_key)
request_fingerprint = _submission_fingerprint(
gateway=gateway,
command=command,
normalized_email=normalized_email,
comment=comment,
)
replay = _idempotent_submission(
session,
invitation_id=invitation.id,
idempotency_key=idempotency_key,
request_fingerprint=request_fingerprint,
)
if replay is not None:
return invitation, replay, True
_assert_poll_accepts_responses(poll)
payload = PollSubmitResponseRequest(
respondent_id=respondent_id,
respondent_label=(
command.respondent_label
or invitation.respondent_label
or invitation.email
or normalized_email
),
answers=answer_inputs,
metadata={
key: value
for key, value in command.metadata.items()
if key not in {"invitation_id", "participant_email", "comment", "response_gateway"}
},
)
normalized_answers = normalize_response_answers(poll, payload)
_validate_normalized_gateway_answers(
poll=poll,
policy=policy,
normalized_answers=normalized_answers,
)
existing = _existing_response(session, poll=poll, respondent_id=respondent_id)
_enforce_capacity(
session,
poll=poll,
existing=existing,
normalized_answers=normalized_answers,
limit=policy.max_participants_per_option,
)
trusted_metadata = dict(payload.metadata)
trusted_metadata["invitation_id"] = invitation.id
trusted_metadata["response_gateway"] = response_gateway_payload(gateway)
if normalized_email is not None:
trusted_metadata["participant_email"] = normalized_email
if comment is not None:
trusted_metadata["comment"] = comment
now = _now()
if existing is not None:
response = _update_existing_response(
session,
poll=poll,
response=existing,
answers=normalized_answers,
respondent_label=payload.respondent_label,
submitted_at=now,
metadata=trusted_metadata,
)
else:
response, _reconciled = _insert_or_reconcile_identified_response(
session,
poll=poll,
respondent_id=respondent_id,
respondent_label=payload.respondent_label,
answers=normalized_answers,
submitted_at=now,
metadata=trusted_metadata,
conflict_validator=lambda winner: _enforce_capacity(
session,
poll=poll,
existing=winner,
normalized_answers=normalized_answers,
limit=policy.max_participants_per_option,
),
)
if idempotency_key is not None:
session.add(
PollParticipationSubmission(
tenant_id=poll.tenant_id,
poll_id=poll.id,
invitation_id=invitation.id,
response_id=response.id,
idempotency_key=idempotency_key,
request_fingerprint=request_fingerprint,
)
)
invitation.last_used_at = response.submitted_at
session.flush()
return invitation, response, False
def response_metadata(response: PollResponse) -> tuple[str | None, str | None]:
metadata = response.metadata_ or {}
email = metadata.get("participant_email")
comment = metadata.get("comment")
return (
email if isinstance(email, str) else None,
comment if isinstance(comment, str) else None,
)
def response_for_invitation(
session: Session,
*,
invitation: PollInvitation,
respondent_id: str | None = None,
participant_email: str | None = None,
) -> PollResponse | None:
normalized_email = _normalize_email(participant_email)
if invitation.respondent_id:
response_respondent_id = invitation.respondent_id
elif respondent_id:
response_respondent_id = respondent_id
elif invitation.email:
response_respondent_id = f"invitation:{invitation.id}"
elif normalized_email:
response_respondent_id = _canonical_respondent_id(
invitation,
respondent_id=None,
participant_email=normalized_email,
participant_is_authenticated=False,
)
else:
return None
return (
session.query(PollResponse)
.filter(
PollResponse.tenant_id == invitation.tenant_id,
PollResponse.poll_id == invitation.poll_id,
PollResponse.respondent_id == response_respondent_id,
PollResponse.deleted_at.is_(None),
)
.order_by(PollResponse.submitted_at.desc(), PollResponse.id.desc())
.first()
)
__all__ = [
"GENERIC_PARTICIPATION_ERROR",
"MAX_COMMENT_LENGTH",
"governed_invitation",
"governed_invitation_by_id",
"participation_policy_payload",
"participation_policy_ref",
"response_for_invitation",
"response_gateway_payload",
"response_gateway_ref",
"response_metadata",
"submit_authenticated_poll_response",
"submit_governed_poll_response",
"update_governed_invitation_expiry",
]

View File

@@ -1,17 +1,21 @@
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, Query, status
from typing import Annotated
from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.db.session import get_session
from govoplan_poll.backend.manifest import READ_SCOPE, RESPOND_SCOPE, WRITE_SCOPE
from govoplan_poll.backend.manifest import ADMIN_SCOPE, READ_SCOPE, RESPOND_SCOPE, WRITE_SCOPE
from govoplan_poll.backend.schemas import (
PollCreateRequest,
PollDecisionRequest,
PollInvitationCreateRequest,
PollInvitationListResponse,
PollInvitationResponse,
PollLifecycleResponse,
PollLifecycleTransitionResponse,
PollListResponse,
PollResponse,
PollResponseItem,
@@ -19,27 +23,35 @@ from govoplan_poll.backend.schemas import (
PollResultSummaryResponse,
PollStatusResponse,
PollSubmitResponseRequest,
PollTransitionRequest,
PollTransitionAvailabilityResponse,
PollUpdateRequest,
)
from govoplan_poll.backend.service import (
INVALID_POLL_OWNER,
OWNED_POLL_MUTATION_REQUIRED,
OWNED_POLL_PROJECTION_RESTRICTED,
POLL_OWNERSHIP_FIELDS_RESTRICTED,
PollError,
close_poll,
assert_generic_poll_projection_allowed,
create_poll_invitation,
create_poll,
decide_poll,
get_poll_by_invitation_token,
get_poll,
get_visible_poll,
list_poll_responses,
list_poll_invitations,
list_polls,
open_poll,
list_poll_lifecycle_transitions,
list_visible_polls,
poll_invitation_response,
poll_lifecycle_transition_response,
poll_response,
poll_response_item,
poll_result_summary_by_id,
require_visible_poll_results,
revoke_poll_invitation,
submit_poll_response,
submit_poll_response_with_token,
transition_poll,
update_poll,
)
@@ -57,13 +69,109 @@ def _poll_http_error(exc: PollError) -> HTTPException:
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
if str(exc) == "Poll invitation not found":
return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
if str(exc) == "Poll results are not visible":
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
if str(exc) == OWNED_POLL_PROJECTION_RESTRICTED:
return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
if str(exc) in {
INVALID_POLL_OWNER,
OWNED_POLL_MUTATION_REQUIRED,
POLL_OWNERSHIP_FIELDS_RESTRICTED,
}:
return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))
return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
def _public_participation_http_error(exc: PollError) -> HTTPException:
"""Keep invalid, revoked, expired, and governed links indistinguishable."""
if str(exc).startswith("Poll invitation"):
return HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Poll invitation not found",
)
return _poll_http_error(exc)
def _poll_response(poll) -> PollResponse:
return PollResponse.model_validate(poll_response(poll))
def _transition_status_response(result) -> PollStatusResponse:
return PollStatusResponse(
poll=_poll_response(result.poll),
transition=(
PollLifecycleTransitionResponse.model_validate(
poll_lifecycle_transition_response(result.transition)
)
if result.transition is not None
else None
),
replayed=result.replayed,
)
def _apply_transition(
session: Session,
principal: ApiPrincipal,
*,
poll_id: str,
action: str,
option_id: str | None = None,
option_key: str | None = None,
idempotency_key: str | None = None,
) -> PollStatusResponse:
try:
result = transition_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
action=action,
option_id=option_id,
option_key=option_key,
idempotency_key=idempotency_key,
actor_user_id=getattr(principal.user, "id", None),
actor_api_key_id=getattr(principal.api_key, "id", None) if principal.api_key else None,
)
except PollError as exc:
raise _poll_http_error(exc) from exc
response = _transition_status_response(result)
session.commit()
return response
def _principal_actor_ids(principal: ApiPrincipal) -> tuple[str, ...]:
candidates = (
principal.account_id,
principal.membership_id,
getattr(principal.user, "id", None),
principal.identity_id,
principal.principal.service_account_id,
)
return tuple(dict.fromkeys(str(value) for value in candidates if value))
def _can_manage_polls(principal: ApiPrincipal) -> bool:
return has_scope(principal, WRITE_SCOPE) or has_scope(principal, ADMIN_SCOPE)
def _can_view_sensitive_poll_data(principal: ApiPrincipal, poll: object) -> bool:
return _can_manage_polls(principal) or getattr(poll, "created_by_user_id", None) in _principal_actor_ids(principal)
def _require_sensitive_poll_data_access(principal: ApiPrincipal, poll: object) -> None:
if not _can_view_sensitive_poll_data(principal, poll):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Poll respondent and invitation data is restricted to poll organizers",
)
def _require_sensitive_poll_data_scope(principal: ApiPrincipal) -> None:
if not (has_scope(principal, READ_SCOPE) or _can_manage_polls(principal)):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {READ_SCOPE}")
@router.get("/polls", response_model=PollListResponse)
def api_list_polls(
status_filter: str | None = Query(default=None, alias="status"),
@@ -72,7 +180,14 @@ def api_list_polls(
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollListResponse:
_require_scope(principal, READ_SCOPE)
polls = list_polls(session, tenant_id=principal.tenant_id, status=status_filter, kind=kind)
polls = list_visible_polls(
session,
tenant_id=principal.tenant_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
status=status_filter,
kind=kind,
)
return PollListResponse(polls=[_poll_response(poll) for poll in polls])
@@ -87,7 +202,9 @@ def api_create_poll(
poll = create_poll(session, tenant_id=principal.tenant_id, user_id=principal.account_id, payload=payload)
except PollError as exc:
raise _poll_http_error(exc) from exc
return _poll_response(poll)
response = _poll_response(poll)
session.commit()
return response
@router.get("/polls/{poll_id}", response_model=PollResponse)
@@ -98,7 +215,15 @@ def api_get_poll(
) -> PollResponse:
_require_scope(principal, READ_SCOPE)
try:
return _poll_response(get_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id))
return _poll_response(
get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
)
except PollError as exc:
raise _poll_http_error(exc) from exc
@@ -112,52 +237,172 @@ def api_update_poll(
) -> PollResponse:
_require_scope(principal, WRITE_SCOPE)
try:
return _poll_response(update_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload))
poll = update_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
except PollError as exc:
raise _poll_http_error(exc) from exc
response = _poll_response(poll)
session.commit()
return response
@router.post("/polls/{poll_id}/open", response_model=PollStatusResponse)
def api_open_poll(
poll_id: str,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
try:
poll = open_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollStatusResponse(poll=_poll_response(poll))
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="open",
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/draft", response_model=PollStatusResponse)
def api_draft_poll(
poll_id: str,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="draft",
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/close", response_model=PollStatusResponse)
def api_close_poll(
poll_id: str,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
try:
poll = close_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollStatusResponse(poll=_poll_response(poll))
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="close",
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/decide", response_model=PollStatusResponse)
def api_decide_poll(
poll_id: str,
payload: PollDecisionRequest,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="decide",
option_id=payload.option_id,
option_key=payload.option_key,
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/archive", response_model=PollStatusResponse)
def api_archive_poll(
poll_id: str,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="archive",
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/unarchive", response_model=PollStatusResponse)
def api_unarchive_poll(
poll_id: str,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
return _apply_transition(
session,
principal,
poll_id=poll_id,
action="unarchive",
idempotency_key=idempotency_key,
)
@router.post("/polls/{poll_id}/transitions", response_model=PollStatusResponse)
def api_transition_poll(
poll_id: str,
payload: PollTransitionRequest,
idempotency_key: Annotated[str | None, Header(alias="Idempotency-Key", max_length=255)] = None,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollStatusResponse:
_require_scope(principal, WRITE_SCOPE)
return _apply_transition(
session,
principal,
poll_id=poll_id,
action=payload.action,
option_id=payload.option_id,
option_key=payload.option_key,
idempotency_key=idempotency_key,
)
@router.get("/polls/{poll_id}/transitions", response_model=PollLifecycleResponse)
def api_poll_lifecycle(
poll_id: str,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollLifecycleResponse:
_require_scope(principal, WRITE_SCOPE)
try:
poll = decide_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
poll = get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=True,
)
history = list_poll_lifecycle_transitions(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollStatusResponse(poll=_poll_response(poll))
projected = _poll_response(poll)
return PollLifecycleResponse(
poll_id=poll.id,
status=poll.status,
archived_from_status=poll.archived_from_status,
actions=[PollTransitionAvailabilityResponse.model_validate(item) for item in projected.lifecycle_actions],
history=[
PollLifecycleTransitionResponse.model_validate(poll_lifecycle_transition_response(item))
for item in history
],
)
@router.post("/polls/{poll_id}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED)
@@ -169,10 +414,31 @@ def api_submit_poll_response(
) -> PollResponseItem:
_require_scope(principal, RESPOND_SCOPE)
try:
response = submit_poll_response(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
actor_payload = payload.model_copy(
update={
"respondent_id": principal.account_id,
"respondent_label": principal.display_name or principal.email,
"metadata": {key: value for key, value in payload.metadata.items() if key != "invitation_id"},
}
)
response = submit_poll_response(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
payload=actor_payload,
)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollResponseItem.model_validate(poll_response_item(response))
result = PollResponseItem.model_validate(poll_response_item(response))
session.commit()
return result
@router.get("/polls/{poll_id}/responses", response_model=PollResponseListResponse)
@@ -181,8 +447,23 @@ def api_list_poll_responses(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollResponseListResponse:
_require_scope(principal, READ_SCOPE)
_require_sensitive_poll_data_scope(principal)
try:
poll = get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
assert_generic_poll_projection_allowed(poll)
_require_sensitive_poll_data_access(principal, poll)
require_visible_poll_results(
session,
poll=poll,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
responses = list_poll_responses(session, tenant_id=principal.tenant_id, poll_id=poll_id)
except PollError as exc:
raise _poll_http_error(exc) from exc
@@ -197,6 +478,19 @@ def api_poll_summary(
) -> PollResultSummaryResponse:
_require_scope(principal, READ_SCOPE)
try:
poll = get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
require_visible_poll_results(
session,
poll=poll,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
return PollResultSummaryResponse.model_validate(poll_result_summary_by_id(session, tenant_id=principal.tenant_id, poll_id=poll_id))
except PollError as exc:
raise _poll_http_error(exc) from exc
@@ -214,7 +508,9 @@ def api_create_poll_invitation(
invitation, token = create_poll_invitation(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollInvitationResponse.model_validate(poll_invitation_response(invitation, token=token))
response = PollInvitationResponse.model_validate(poll_invitation_response(invitation, token=token))
session.commit()
return response
@router.get("/polls/{poll_id}/invitations", response_model=PollInvitationListResponse)
@@ -223,8 +519,17 @@ def api_list_poll_invitations(
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> PollInvitationListResponse:
_require_scope(principal, READ_SCOPE)
_require_sensitive_poll_data_scope(principal)
try:
poll = get_visible_poll(
session,
tenant_id=principal.tenant_id,
poll_id=poll_id,
actor_ids=_principal_actor_ids(principal),
can_manage=_can_manage_polls(principal),
)
assert_generic_poll_projection_allowed(poll)
_require_sensitive_poll_data_access(principal, poll)
invitations = list_poll_invitations(session, tenant_id=principal.tenant_id, poll_id=poll_id)
except PollError as exc:
raise _poll_http_error(exc) from exc
@@ -250,7 +555,9 @@ def api_revoke_poll_invitation(
)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollInvitationResponse.model_validate(poll_invitation_response(invitation))
response = PollInvitationResponse.model_validate(poll_invitation_response(invitation))
session.commit()
return response
@router.get("/public/{token}", response_model=PollResponse)
@@ -261,7 +568,7 @@ def api_get_public_poll(
try:
return _poll_response(get_poll_by_invitation_token(session, token=token))
except PollError as exc:
raise _poll_http_error(exc) from exc
raise _public_participation_http_error(exc) from exc
@router.post("/public/{token}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED)
@@ -273,5 +580,7 @@ def api_submit_public_poll_response(
try:
response = submit_poll_response_with_token(session, token=token, payload=payload)
except PollError as exc:
raise _poll_http_error(exc) from exc
return PollResponseItem.model_validate(poll_response_item(response))
raise _public_participation_http_error(exc) from exc
result = PollResponseItem.model_validate(poll_response_item(response))
session.commit()
return result

View File

@@ -8,6 +8,8 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
PollKind = Literal["single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"]
PollStatus = Literal["draft", "open", "closed", "decided", "archived"]
PollInitialStatus = Literal["draft", "open"]
PollTransitionAction = Literal["open", "draft", "close", "decide", "archive", "unarchive"]
PollVisibility = Literal["private", "tenant", "public", "unlisted"]
PollResultVisibility = Literal["organizer", "after_response", "after_close", "public"]
AvailabilityValue = Literal["available", "maybe", "unavailable"]
@@ -30,7 +32,7 @@ class PollCreateRequest(BaseModel):
slug: str | None = Field(default=None, max_length=120)
description: str | None = None
kind: PollKind = "single_choice"
status: PollStatus = "draft"
status: PollInitialStatus = "draft"
visibility: PollVisibility = "private"
result_visibility: PollResultVisibility = "after_close"
context_module: str | None = Field(default=None, max_length=100)
@@ -85,6 +87,13 @@ class PollOptionResponse(BaseModel):
metadata: dict[str, Any] = Field(default_factory=dict)
class PollTransitionAvailabilityResponse(BaseModel):
action: PollTransitionAction
target_status: PollStatus | None = None
available: bool
reason: str | None = None
class PollResponseItem(BaseModel):
id: str
respondent_id: str | None = None
@@ -120,11 +129,13 @@ class PollResponse(BaseModel):
closed_at: datetime | None = None
decided_at: datetime | None = None
decided_option_id: str | None = None
archived_from_status: str | None = None
created_by_user_id: str | None = None
created_at: datetime
updated_at: datetime
metadata: dict[str, Any] = Field(default_factory=dict)
options: list[PollOptionResponse] = Field(default_factory=list)
lifecycle_actions: list[PollTransitionAvailabilityResponse] = Field(default_factory=list)
class PollListResponse(BaseModel):
@@ -175,14 +186,66 @@ class PollDecisionRequest(BaseModel):
option_key: str | None = None
class PollTransitionRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
action: PollTransitionAction
option_id: str | None = None
option_key: str | None = None
class PollLifecycleTransitionResponse(BaseModel):
id: str
action: PollTransitionAction
from_status: PollStatus
to_status: PollStatus
decision_option_id: str | None = None
previous_decision_option_id: str | None = None
idempotency_key: str | None = None
actor_user_id: str | None = None
actor_api_key_id: str | None = None
created_at: datetime
metadata: dict[str, Any] = Field(default_factory=dict)
class PollStatusResponse(BaseModel):
poll: PollResponse
transition: PollLifecycleTransitionResponse | None = None
replayed: bool = False
class PollLifecycleResponse(BaseModel):
poll_id: str
status: PollStatus
archived_from_status: str | None = None
actions: list[PollTransitionAvailabilityResponse] = Field(default_factory=list)
history: list[PollLifecycleTransitionResponse] = Field(default_factory=list)
class PollResponseListResponse(BaseModel):
responses: list[PollResponseItem] = Field(default_factory=list)
class PollResponseGatewayInput(BaseModel):
model_config = ConfigDict(extra="forbid")
module_id: str = Field(min_length=1, max_length=100)
resource_type: str = Field(min_length=1, max_length=100)
resource_id: str = Field(min_length=1, max_length=255)
class PollParticipationPolicyInput(BaseModel):
model_config = ConfigDict(extra="forbid")
version: Literal[1] = 1
single_choice: bool = False
allow_maybe: bool = True
max_participants_per_option: int | None = Field(default=None, ge=1)
allow_comments: bool = False
participant_email_required: bool = False
anonymous_password_required: bool = False
class PollInvitationCreateRequest(BaseModel):
model_config = ConfigDict(extra="forbid")
@@ -190,8 +253,18 @@ class PollInvitationCreateRequest(BaseModel):
respondent_label: str | None = Field(default=None, max_length=500)
email: str | None = Field(default=None, max_length=320)
expires_at: datetime | None = None
response_gateway: PollResponseGatewayInput | None = None
participation_policy: PollParticipationPolicyInput | None = None
metadata: dict[str, Any] = Field(default_factory=dict)
@model_validator(mode="after")
def validate_governed_participation(self) -> "PollInvitationCreateRequest":
if (self.response_gateway is None) != (self.participation_policy is None):
raise ValueError(
"response_gateway and participation_policy must be configured together"
)
return self
class PollInvitationResponse(BaseModel):
id: str
@@ -205,6 +278,8 @@ class PollInvitationResponse(BaseModel):
last_used_at: datetime | None = None
created_at: datetime
updated_at: datetime
response_gateway: PollResponseGatewayInput | None = None
participation_policy: PollParticipationPolicyInput | None = None
metadata: dict[str, Any] = Field(default_factory=dict)

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,197 @@
from __future__ import annotations
from dataclasses import dataclass
from types import MappingProxyType
from typing import Mapping
POLL_STATUSES = frozenset({"draft", "open", "closed", "decided", "archived"})
POLL_TRANSITION_ACTIONS = ("open", "draft", "close", "decide", "archive", "unarchive")
@dataclass(frozen=True)
class PollTransitionRule:
"""One configurable lifecycle action.
Archive restoration is deliberately represented as a dynamic target. The
engine remains a pure policy component; persistence and domain side effects
stay in the Poll service.
"""
action: str
source_statuses: frozenset[str]
target_status: str | None
restore_archived_status: bool = False
missing_restore_target_status: str | None = None
@dataclass(frozen=True)
class PollTransitionPlan:
action: str
from_status: str
to_status: str
@dataclass(frozen=True)
class PollTransitionAvailability:
action: str
target_status: str | None
available: bool
reason: str | None = None
class PollTransitionPolicy:
"""Validated, replaceable configuration for a Poll lifecycle."""
def __init__(self, rules: Mapping[str, PollTransitionRule]) -> None:
normalized = dict(rules)
if set(normalized) != set(POLL_TRANSITION_ACTIONS):
missing = sorted(set(POLL_TRANSITION_ACTIONS) - set(normalized))
extra = sorted(set(normalized) - set(POLL_TRANSITION_ACTIONS))
raise ValueError(f"Poll transition policy has missing actions {missing} and extra actions {extra}")
for action, rule in normalized.items():
if rule.action != action:
raise ValueError(f"Poll transition rule key {action!r} does not match action {rule.action!r}")
if not rule.source_statuses <= POLL_STATUSES:
raise ValueError(f"Poll transition rule {action!r} has unknown source statuses")
if rule.restore_archived_status:
if rule.target_status is not None:
raise ValueError(f"Restoring transition {action!r} cannot have a fixed target")
if rule.missing_restore_target_status not in (POLL_STATUSES - {"archived"}) | {None}:
raise ValueError(f"Restoring transition {action!r} has an invalid missing-origin target")
elif rule.target_status not in POLL_STATUSES:
raise ValueError(f"Poll transition rule {action!r} has an unknown target status")
elif rule.missing_restore_target_status is not None:
raise ValueError(f"Non-restoring transition {action!r} cannot have a missing-origin target")
self._rules = MappingProxyType(normalized)
@property
def rules(self) -> Mapping[str, PollTransitionRule]:
return self._rules
DEFAULT_POLL_TRANSITION_POLICY = PollTransitionPolicy(
{
"open": PollTransitionRule(
action="open",
source_statuses=frozenset({"draft", "closed", "decided"}),
target_status="open",
),
"draft": PollTransitionRule(
action="draft",
source_statuses=frozenset({"open"}),
target_status="draft",
),
"close": PollTransitionRule(
action="close",
source_statuses=frozenset({"open"}),
target_status="closed",
),
"decide": PollTransitionRule(
action="decide",
source_statuses=frozenset({"closed", "decided"}),
target_status="decided",
),
"archive": PollTransitionRule(
action="archive",
source_statuses=frozenset({"draft", "open", "closed", "decided"}),
target_status="archived",
),
"unarchive": PollTransitionRule(
action="unarchive",
source_statuses=frozenset({"archived"}),
target_status=None,
restore_archived_status=True,
missing_restore_target_status="draft",
),
}
)
class PollTransitionEngine:
"""Pure lifecycle planner backed by an injected transition policy."""
def __init__(self, policy: PollTransitionPolicy = DEFAULT_POLL_TRANSITION_POLICY) -> None:
self.policy = policy
def plan(
self,
*,
current_status: str,
action: str,
archived_from_status: str | None = None,
) -> PollTransitionPlan:
if current_status not in POLL_STATUSES:
raise ValueError(f"Unknown poll status: {current_status}")
rule = self.policy.rules.get(action)
if rule is None:
raise ValueError(f"Unknown poll transition action: {action}")
if current_status not in rule.source_statuses:
raise ValueError(f"Poll transition {action!r} is not allowed from status {current_status!r}")
target_status = rule.target_status
if rule.restore_archived_status:
if archived_from_status is None and rule.missing_restore_target_status is not None:
target_status = rule.missing_restore_target_status
elif archived_from_status not in POLL_STATUSES - {"archived"}:
raise ValueError("Archived poll has no valid status to restore")
else:
target_status = archived_from_status
if target_status is None:
raise ValueError(f"Poll transition {action!r} did not resolve a target status")
return PollTransitionPlan(action=action, from_status=current_status, to_status=target_status)
def available_actions(
self,
*,
current_status: str,
archived_from_status: str | None = None,
) -> tuple[PollTransitionAvailability, ...]:
actions: list[PollTransitionAvailability] = []
for action in POLL_TRANSITION_ACTIONS:
rule = self.policy.rules[action]
target_status = (
archived_from_status or rule.missing_restore_target_status
if rule.restore_archived_status
else rule.target_status
)
try:
self.plan(
current_status=current_status,
action=action,
archived_from_status=archived_from_status,
)
except ValueError as exc:
actions.append(
PollTransitionAvailability(
action=action,
target_status=target_status,
available=False,
reason=str(exc),
)
)
else:
actions.append(
PollTransitionAvailability(
action=action,
target_status=target_status,
available=True,
)
)
return tuple(actions)
DEFAULT_POLL_TRANSITION_ENGINE = PollTransitionEngine()
__all__ = [
"DEFAULT_POLL_TRANSITION_ENGINE",
"DEFAULT_POLL_TRANSITION_POLICY",
"POLL_STATUSES",
"POLL_TRANSITION_ACTIONS",
"PollTransitionAvailability",
"PollTransitionEngine",
"PollTransitionPlan",
"PollTransitionPolicy",
"PollTransitionRule",
]

306
tests/test_authorization.py Normal file
View File

@@ -0,0 +1,306 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
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.db.base import Base
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
from govoplan_poll.backend.router import (
api_create_poll,
api_get_poll,
api_list_poll_invitations,
api_list_poll_responses,
api_list_polls,
api_poll_summary,
api_submit_poll_response,
)
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollCreateRequest,
PollInvitationCreateRequest,
PollSubmitResponseRequest,
)
from govoplan_poll.backend.service import close_poll, create_poll, create_poll_invitation, open_poll
READ_SCOPE = "poll:poll:read"
WRITE_SCOPE = "poll:poll:write"
RESPOND_SCOPE = "poll:response:write"
class PollAuthorizationTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(
self.engine,
tables=[
Poll.__table__,
PollOption.__table__,
PollResponse.__table__,
PollInvitation.__table__,
PollLifecycleTransition.__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,
tables=[
PollLifecycleTransition.__table__,
PollInvitation.__table__,
PollResponse.__table__,
PollOption.__table__,
Poll.__table__,
],
)
self.engine.dispose()
@staticmethod
def _principal(
account_id: str,
*,
tenant_id: str = "tenant-1",
user_id: str | None = None,
scopes: set[str] | None = None,
) -> ApiPrincipal:
membership_id = user_id or f"membership-{account_id}"
display_name = f"Actor {account_id}"
return ApiPrincipal(
principal=PrincipalRef(
account_id=account_id,
membership_id=membership_id,
tenant_id=tenant_id,
scopes=frozenset(scopes or {READ_SCOPE, RESPOND_SCOPE}),
display_name=display_name,
),
account=SimpleNamespace(id=account_id),
user=SimpleNamespace(id=membership_id),
)
def _poll(
self,
*,
owner_id: str = "owner",
visibility: str = "tenant",
result_visibility: str = "after_close",
title: str = "Decision",
) -> Poll:
poll = create_poll(
self.session,
tenant_id="tenant-1",
user_id=owner_id,
payload=PollCreateRequest(
title=title,
kind="yes_no",
visibility=visibility,
result_visibility=result_visibility,
),
)
open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id)
return poll
@staticmethod
def _answer(*, claimed_respondent: str = "victim") -> PollSubmitResponseRequest:
return PollSubmitResponseRequest(
respondent_id=claimed_respondent,
respondent_label="Claimed victim",
answers=[PollAnswerInput(option_key="yes")],
)
def test_authenticated_response_identity_is_bound_to_actor(self) -> None:
poll = self._poll(result_visibility="public")
first_actor = self._principal("account-1")
second_actor = self._principal("account-2")
first = api_submit_poll_response(
poll.id,
self._answer(),
session=self.session,
principal=first_actor,
)
second = api_submit_poll_response(
poll.id,
self._answer(),
session=self.session,
principal=second_actor,
)
updated_first = api_submit_poll_response(
poll.id,
PollSubmitResponseRequest(
respondent_id="account-2",
answers=[PollAnswerInput(option_key="no")],
),
session=self.session,
principal=first_actor,
)
self.assertEqual(first.respondent_id, "account-1")
self.assertEqual(first.respondent_label, "Actor account-1")
self.assertEqual(second.respondent_id, "account-2")
self.assertNotEqual(first.id, second.id)
self.assertEqual(updated_first.id, first.id)
self.assertEqual(updated_first.answers[0]["option_key"], "no")
cross_tenant_actor = self._principal("account-1", tenant_id="tenant-2")
with self.assertRaises(HTTPException) as cross_tenant:
api_submit_poll_response(
poll.id,
self._answer(),
session=self.session,
principal=cross_tenant_actor,
)
self.assertEqual(cross_tenant.exception.status_code, 404)
def test_create_route_persists_after_request_session_closes(self) -> None:
created = api_create_poll(
PollCreateRequest(title="Persisted poll", kind="yes_no", visibility="tenant"),
session=self.session,
principal=self._principal("owner", scopes={WRITE_SCOPE}),
)
self.session.close()
self.session = self.Session()
persisted = self.session.query(Poll).filter(Poll.id == created.id).one_or_none()
self.assertIsNotNone(persisted)
self.assertEqual(persisted.title, "Persisted poll")
def test_authenticated_response_cannot_claim_an_invitation(self) -> None:
poll = self._poll(result_visibility="public")
response = api_submit_poll_response(
poll.id,
self._answer().model_copy(
update={"metadata": {"invitation_id": "forged-invitation", "client_note": "kept"}}
),
session=self.session,
principal=self._principal("account-1"),
)
self.assertEqual(response.metadata, {"client_note": "kept"})
def test_private_poll_requires_ownership_management_or_active_assignment(self) -> None:
poll = self._poll(visibility="private")
reader = self._principal("reader")
listed = api_list_polls(
status_filter=None,
kind=None,
session=self.session,
principal=reader,
)
self.assertEqual(listed.polls, [])
with self.assertRaises(HTTPException) as hidden:
api_get_poll(poll.id, session=self.session, principal=reader)
self.assertEqual(hidden.exception.status_code, 404)
invitation, _token = create_poll_invitation(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollInvitationCreateRequest(respondent_id=reader.membership_id),
)
visible = api_get_poll(poll.id, session=self.session, principal=reader)
self.assertEqual(visible.id, poll.id)
invitation.revoked_at = invitation.created_at
self.session.flush()
with self.assertRaises(HTTPException) as revoked:
api_get_poll(poll.id, session=self.session, principal=reader)
self.assertEqual(revoked.exception.status_code, 404)
manager = self._principal("manager", scopes={READ_SCOPE, WRITE_SCOPE})
self.assertEqual(api_get_poll(poll.id, session=self.session, principal=manager).id, poll.id)
other_tenant = self._principal("owner", tenant_id="tenant-2", scopes={READ_SCOPE, WRITE_SCOPE})
with self.assertRaises(HTTPException) as cross_tenant:
api_get_poll(poll.id, session=self.session, principal=other_tenant)
self.assertEqual(cross_tenant.exception.status_code, 404)
def test_after_response_results_do_not_leak_to_other_actors(self) -> None:
poll = self._poll(result_visibility="after_response")
respondent = self._principal("respondent")
bystander = self._principal("bystander")
with self.assertRaises(HTTPException) as before_response:
api_poll_summary(poll.id, session=self.session, principal=respondent)
self.assertEqual(before_response.exception.status_code, 403)
api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=respondent)
self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=respondent).response_count, 1)
with self.assertRaises(HTTPException) as own_raw_responses:
api_list_poll_responses(poll.id, session=self.session, principal=respondent)
self.assertEqual(own_raw_responses.exception.status_code, 403)
with self.assertRaises(HTTPException) as other_actor_summary:
api_poll_summary(poll.id, session=self.session, principal=bystander)
self.assertEqual(other_actor_summary.exception.status_code, 403)
with self.assertRaises(HTTPException) as other_actor_responses:
api_list_poll_responses(poll.id, session=self.session, principal=bystander)
self.assertEqual(other_actor_responses.exception.status_code, 403)
def test_participants_can_see_aggregate_but_not_raw_responses_or_invitation_roster(self) -> None:
poll = self._poll(result_visibility="public")
first_actor = self._principal("account-1")
second_actor = self._principal("account-2")
create_poll_invitation(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollInvitationCreateRequest(respondent_id=first_actor.membership_id, email="one@example.test"),
)
create_poll_invitation(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollInvitationCreateRequest(respondent_id=second_actor.membership_id, email="two@example.test"),
)
api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=first_actor)
api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=second_actor)
self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=first_actor).response_count, 2)
with self.assertRaises(HTTPException) as raw_responses:
api_list_poll_responses(poll.id, session=self.session, principal=first_actor)
self.assertEqual(raw_responses.exception.status_code, 403)
with self.assertRaises(HTTPException) as invitation_roster:
api_list_poll_invitations(poll.id, session=self.session, principal=first_actor)
self.assertEqual(invitation_roster.exception.status_code, 403)
organizer = self._principal("owner", scopes={READ_SCOPE})
self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=organizer).responses), 2)
self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=organizer).invitations), 2)
manager = self._principal("manager", scopes={WRITE_SCOPE})
self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=manager).responses), 2)
self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=manager).invitations), 2)
def test_after_close_and_organizer_result_visibility_are_enforced(self) -> None:
after_close = self._poll(result_visibility="after_close", title="After close")
reader = self._principal("reader")
with self.assertRaises(HTTPException) as still_open:
api_poll_summary(after_close.id, session=self.session, principal=reader)
self.assertEqual(still_open.exception.status_code, 403)
close_poll(self.session, tenant_id="tenant-1", poll_id=after_close.id)
self.assertEqual(api_poll_summary(after_close.id, session=self.session, principal=reader).response_count, 0)
organizer_only = self._poll(result_visibility="organizer", title="Organizer only")
api_submit_poll_response(organizer_only.id, self._answer(), session=self.session, principal=reader)
with self.assertRaises(HTTPException) as participant_denied:
api_poll_summary(organizer_only.id, session=self.session, principal=reader)
self.assertEqual(participant_denied.exception.status_code, 403)
owner = self._principal("owner", scopes={READ_SCOPE})
self.assertEqual(api_poll_summary(organizer_only.id, session=self.session, principal=owner).response_count, 1)
if __name__ == "__main__":
unittest.main()

563
tests/test_lifecycle.py Normal file
View File

@@ -0,0 +1,563 @@
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from fastapi import HTTPException
from pydantic import ValidationError
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.db.base import Base
from govoplan_poll.backend.db.models import (
Poll,
PollInvitation,
PollLifecycleTransition,
PollOption,
PollResponse,
)
from govoplan_poll.backend.router import api_poll_lifecycle, api_transition_poll
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollCreateRequest,
PollOptionInput,
PollSubmitResponseRequest,
PollTransitionRequest,
)
from govoplan_poll.backend.service import (
PollError,
create_poll,
list_poll_lifecycle_transitions,
poll_response,
poll_results_are_visible,
response_datetime,
submit_poll_response,
transition_poll,
)
from govoplan_poll.backend.transitions import (
DEFAULT_POLL_TRANSITION_POLICY,
PollTransitionEngine,
PollTransitionPolicy,
PollTransitionRule,
)
WRITE_SCOPE = "poll:poll:write"
class PollLifecycleTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(
self.engine,
tables=[
Poll.__table__,
PollOption.__table__,
PollResponse.__table__,
PollInvitation.__table__,
PollLifecycleTransition.__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,
tables=[
PollLifecycleTransition.__table__,
PollInvitation.__table__,
PollResponse.__table__,
PollOption.__table__,
Poll.__table__,
],
)
self.engine.dispose()
def _poll(self, *, status: str = "draft", title: str = "Decision") -> Poll:
return create_poll(
self.session,
tenant_id="tenant-1",
user_id="owner",
payload=PollCreateRequest(
title=title,
kind="single_choice",
status=status,
visibility="tenant",
options=[
PollOptionInput(key="yes", label="Yes"),
PollOptionInput(key="no", label="No"),
],
),
)
@staticmethod
def _principal() -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id="owner",
membership_id="owner-membership",
tenant_id="tenant-1",
scopes=frozenset({WRITE_SCOPE}),
display_name="Owner",
),
account=SimpleNamespace(id="owner"),
user=SimpleNamespace(id="owner-membership"),
)
def test_default_policy_exposes_every_allowed_and_rejected_transition(self) -> None:
expected = {
"draft": {"open", "archive"},
"open": {"draft", "close", "archive"},
"closed": {"open", "decide", "archive"},
"decided": {"open", "decide", "archive"},
"archived": {"unarchive"},
}
engine = PollTransitionEngine()
for status, allowed in expected.items():
archived_from_status = "closed" if status == "archived" else None
availability = engine.available_actions(
current_status=status,
archived_from_status=archived_from_status,
)
self.assertEqual({item.action for item in availability if item.available}, allowed)
for item in availability:
if item.available:
self.assertIsNone(item.reason)
else:
self.assertTrue(item.reason)
with self.assertRaisesRegex(ValueError, "not allowed"):
engine.plan(current_status="draft", action="close")
def test_injected_policy_governs_service_transition_and_projection(self) -> None:
rules = dict(DEFAULT_POLL_TRANSITION_POLICY.rules)
rules["close"] = PollTransitionRule(
action="close",
source_statuses=frozenset({"draft", "open"}),
target_status="closed",
)
custom_engine = PollTransitionEngine(PollTransitionPolicy(rules))
poll = self._poll()
projected = poll_response(poll, transition_engine=custom_engine)
close_action = next(item for item in projected["lifecycle_actions"] if item["action"] == "close")
result = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="close",
transition_engine=custom_engine,
)
self.assertTrue(close_action["available"])
self.assertEqual(result.poll.status, "closed")
self.assertEqual(result.transition.from_status, "draft")
def test_only_draft_or_open_are_valid_initial_states(self) -> None:
for initial_status in ("closed", "decided", "archived"):
with self.subTest(initial_status=initial_status):
with self.assertRaises(ValidationError):
PollCreateRequest(title="Invalid initial state", kind="yes_no", status=initial_status)
bypassed_schema = PollCreateRequest(title="Invalid initial state", kind="yes_no").model_copy(
update={"status": "closed"}
)
with self.assertRaisesRegex(PollError, "initial status must be draft or open"):
create_poll(
self.session,
tenant_id="tenant-1",
user_id="owner",
payload=bypassed_schema,
)
def test_reopening_preserves_responses_close_history_and_previous_decision(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
response = submit_poll_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(
respondent_id="participant",
answers=[PollAnswerInput(option_key="yes")],
),
)
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
first_closed_at = poll.closed_at
transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
)
decided_at = poll.decided_at
decided_option_id = poll.decided_option_id
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
self.assertEqual(poll.status, "open")
self.assertEqual(response_datetime(poll.closed_at), response_datetime(first_closed_at))
self.assertEqual(response_datetime(poll.decided_at), response_datetime(decided_at))
self.assertEqual(poll.decided_option_id, decided_option_id)
self.assertEqual(response.deleted_at, None)
self.assertEqual(response.answers[0]["option_key"], "yes")
history = list_poll_lifecycle_transitions(self.session, tenant_id="tenant-1", poll_id=poll.id)
self.assertEqual([item.action for item in history], ["open", "close", "decide", "open"])
def test_direct_redecision_is_audited_and_exact_retry_is_idempotent(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
first = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
idempotency_key="decision-1",
)
replay = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
idempotency_key="decision-1",
)
second = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="no",
idempotency_key="decision-2",
)
self.assertFalse(first.replayed)
self.assertTrue(replay.replayed)
self.assertEqual(replay.transition.id, first.transition.id)
self.assertFalse(second.replayed)
self.assertEqual(second.transition.previous_decision_option_id, first.transition.decision_option_id)
self.assertEqual(poll.decided_option_id, second.transition.decision_option_id)
self.assertEqual(
self.session.query(PollLifecycleTransition)
.filter(
PollLifecycleTransition.poll_id == poll.id,
PollLifecycleTransition.action == "decide",
)
.count(),
2,
)
with self.assertRaisesRegex(PollError, "different poll transition"):
transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
idempotency_key="decision-2",
)
def test_repeated_exact_transition_without_an_identity_is_an_audit_free_noop(self) -> None:
poll = self._poll()
applied = transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
with self.assertLogs("govoplan.poll.lifecycle", level="INFO") as logs:
repeated = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="open",
actor_user_id="owner",
)
self.assertFalse(applied.replayed)
self.assertTrue(repeated.replayed)
self.assertIsNone(repeated.transition)
self.assertEqual(poll.status, "open")
self.assertEqual(len(poll.lifecycle_transitions), 1)
self.assertIn("Ignored repeated exact Poll lifecycle transition", logs.output[0])
def test_same_decision_is_a_noop_but_a_changed_decision_is_audited(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
first = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
)
repeated = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="yes",
)
changed = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="decide",
option_key="no",
)
self.assertTrue(repeated.replayed)
self.assertIsNone(repeated.transition)
self.assertIsNotNone(first.transition)
self.assertIsNotNone(changed.transition)
self.assertEqual(changed.transition.previous_decision_option_id, first.transition.decision_option_id)
self.assertEqual(
[item.action for item in poll.lifecycle_transitions],
["open", "close", "decide", "decide"],
)
def test_exact_unarchive_retry_is_a_noop_but_unarchive_without_history_is_invalid(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="archive")
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive")
repeated = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="unarchive",
)
self.assertTrue(repeated.replayed)
self.assertIsNone(repeated.transition)
self.assertEqual([item.action for item in poll.lifecycle_transitions], ["archive", "unarchive"])
never_archived = self._poll(title="Never archived")
with self.assertRaisesRegex(PollError, "not allowed"):
transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=never_archived.id,
action="unarchive",
)
def test_archive_and_unarchive_restore_status_and_append_audit_history(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
response = PollResponse(
tenant_id="tenant-1",
poll_id=poll.id,
respondent_id="participant",
answers=[{"option_key": "yes"}],
submitted_at=poll.created_at,
)
self.session.add(response)
self.session.flush()
archived = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="archive",
idempotency_key="archive-1",
)
restored = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="unarchive",
idempotency_key="unarchive-1",
)
self.assertEqual(archived.transition.to_status, "archived")
self.assertEqual(restored.transition.to_status, "closed")
self.assertEqual(poll.status, "closed")
self.assertIsNone(poll.archived_from_status)
self.assertIsNone(response.deleted_at)
self.assertEqual(
[item.action for item in poll.lifecycle_transitions],
["open", "close", "archive", "unarchive"],
)
def test_legacy_archive_without_origin_restores_to_draft_with_audit_marker(self) -> None:
poll = Poll(
tenant_id="tenant-1",
slug="legacy-archive",
title="Legacy archive",
kind="yes_no",
status="archived",
visibility="tenant",
result_visibility="after_close",
workflow_steps=[],
allow_anonymous=False,
allow_response_update=True,
min_choices=1,
)
self.session.add(poll)
self.session.flush()
restored = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="unarchive",
)
self.assertEqual(restored.poll.status, "draft")
self.assertEqual(restored.transition.to_status, "draft")
self.assertEqual(
restored.transition.metadata_["legacy_restore_fallback"],
{
"reason": "missing_archived_from_status",
"restored_status": "draft",
},
)
def test_reopening_clears_only_an_expired_deadline_and_preserves_close_history(self) -> None:
poll = self._poll()
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open")
transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close")
closed_at = poll.closed_at
expired_deadline = datetime.now(timezone.utc) - timedelta(days=1)
poll.closes_at = expired_deadline
reopened = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="open",
)
self.assertEqual(reopened.poll.status, "open")
self.assertIsNone(reopened.poll.closes_at)
self.assertEqual(response_datetime(reopened.poll.closed_at), response_datetime(closed_at))
self.assertEqual(
reopened.transition.metadata_["cleared_expired_closes_at"],
expired_deadline.isoformat(),
)
submit_poll_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(
respondent_id="participant-after-reopen",
answers=[PollAnswerInput(option_key="yes")],
),
)
future_deadline = datetime.now(timezone.utc) + timedelta(days=1)
future_poll = self._poll(title="Future deadline")
future_poll.closes_at = future_deadline
opened = transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=future_poll.id,
action="open",
)
self.assertEqual(response_datetime(opened.poll.closes_at), future_deadline)
self.assertNotIn("cleared_expired_closes_at", opened.transition.metadata_)
def test_archiving_draft_or_open_poll_does_not_grant_after_close_visibility(self) -> None:
for initial_status in ("draft", "open"):
with self.subTest(initial_status=initial_status):
poll = self._poll(status=initial_status, title=f"Archive from {initial_status}")
transition_poll(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
action="archive",
)
self.assertFalse(
poll_results_are_visible(
self.session,
poll=poll,
actor_ids=("participant",),
)
)
closed_poll = self._poll(title="Archive from closed")
transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="open")
transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="close")
transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="archive")
self.assertTrue(
poll_results_are_visible(
self.session,
poll=closed_poll,
actor_ids=("participant",),
)
)
def test_generic_api_returns_action_availability_history_and_actor(self) -> None:
poll = self._poll()
principal = self._principal()
opened = api_transition_poll(
poll.id,
PollTransitionRequest(action="open"),
idempotency_key="open-api-1",
session=self.session,
principal=principal,
)
replay = api_transition_poll(
poll.id,
PollTransitionRequest(action="open"),
idempotency_key="open-api-1",
session=self.session,
principal=principal,
)
lifecycle = api_poll_lifecycle(poll.id, session=self.session, principal=principal)
self.assertEqual(opened.poll.status, "open")
self.assertFalse(opened.replayed)
self.assertTrue(replay.replayed)
self.assertIsNotNone(replay.transition)
self.assertEqual(len(lifecycle.history), 1)
self.assertEqual(lifecycle.history[0].actor_user_id, "owner-membership")
self.assertEqual(
{item.action for item in lifecycle.actions if item.available},
{"draft", "close", "archive"},
)
draft = self._poll(title="Draft cannot close")
with self.assertRaises(HTTPException) as rejected:
api_transition_poll(
draft.id,
PollTransitionRequest(action="close"),
idempotency_key=None,
session=self.session,
principal=principal,
)
self.assertEqual(rejected.exception.status_code, 400)
def test_generic_api_returns_null_transition_for_domain_noop(self) -> None:
poll = self._poll(status="open", title="Already open")
repeated = api_transition_poll(
poll.id,
PollTransitionRequest(action="open"),
idempotency_key=None,
session=self.session,
principal=self._principal(),
)
self.assertEqual(repeated.poll.status, "open")
self.assertTrue(repeated.replayed)
self.assertIsNone(repeated.transition)
self.assertEqual(
self.session.query(PollLifecycleTransition)
.filter(PollLifecycleTransition.poll_id == poll.id)
.count(),
0,
)
if __name__ == "__main__":
unittest.main()

View File

@@ -4,6 +4,7 @@ import unittest
from govoplan_core.core.modules import ModuleManifest
from govoplan_poll.backend.manifest import get_manifest
from govoplan_poll.backend.participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
class PollManifestTests(unittest.TestCase):
@@ -21,6 +22,9 @@ class PollManifestTests(unittest.TestCase):
self.assertIn("poll.availability_matrix", {interface.name for interface in manifest.provides_interfaces})
self.assertIn("poll.workflow_context", {interface.name for interface in manifest.provides_interfaces})
self.assertIn("poll.signed_participation", {interface.name for interface in manifest.provides_interfaces})
self.assertIn("poll.governed_participation", {interface.name for interface in manifest.provides_interfaces})
self.assertIn(CAPABILITY_POLL_PARTICIPATION_GATEWAY, manifest.capability_factories)
self.assertEqual(manifest.version, "0.1.11")
self.assertIn("poll:response:write", {permission.scope for permission in manifest.permissions})

210
tests/test_migrations.py Normal file
View File

@@ -0,0 +1,210 @@
from __future__ import annotations
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from alembic import command
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from govoplan_core.db.migrations import alembic_config
from govoplan_poll.backend.db.models import PollResponse
from govoplan_poll.backend.manifest import get_manifest
from govoplan_poll.backend.schemas import (
PollCreateRequest,
PollOptionInput,
)
from govoplan_poll.backend.service import _existing_response, create_poll
_PREVIOUS_POLL_HEAD = "5d6e7f8a9b0c"
_POLL_HEAD = "6e7f8a9b0c1d"
_INDEX_NAME = "uq_poll_responses_active_respondent"
class PollMigrationTests(unittest.TestCase):
@staticmethod
def _config(url: str):
return alembic_config(
database_url=url,
enabled_modules=("poll",),
manifest_factories=(get_manifest,),
)
def test_deduplicates_deterministically_and_enforces_partial_index(self) -> None:
with tempfile.TemporaryDirectory(
prefix="govoplan-poll-migration-"
) as directory:
url = f"sqlite:///{Path(directory) / 'poll.db'}"
config = self._config(url)
command.upgrade(config, _PREVIOUS_POLL_HEAD)
engine = create_engine(url)
latest = datetime(2026, 7, 22, 10, 0, tzinfo=timezone.utc)
historical_deleted_at = latest - timedelta(days=2)
historical_updated_at = latest - timedelta(days=1)
try:
with Session(engine) as session:
poll = create_poll(
session,
tenant_id="tenant-1",
user_id="owner-1",
payload=PollCreateRequest(
title="Migration probe",
kind="single_choice",
status="open",
allow_anonymous=True,
options=[
PollOptionInput(key="yes", label="Yes"),
PollOptionInput(key="no", label="No"),
],
),
)
poll_id = poll.id
session.add_all(
[
PollResponse(
id="identified-oldest",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest - timedelta(hours=1),
),
PollResponse(
id="identified-latest-a",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest,
),
PollResponse(
id="identified-latest-b",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest,
),
PollResponse(
id="anonymous-a",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest,
),
PollResponse(
id="anonymous-b",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest,
),
PollResponse(
id="already-deleted",
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest + timedelta(hours=1),
deleted_at=historical_deleted_at,
updated_at=historical_updated_at,
),
]
)
session.flush()
self.assertEqual(
_existing_response(
session,
poll=poll,
respondent_id="person-1",
).id,
"identified-latest-b",
)
session.commit()
command.upgrade(config, "heads")
with Session(engine) as session:
rows = {
response.id: response
for response in session.query(PollResponse).all()
}
self.assertIsNone(rows["identified-latest-b"].deleted_at)
tombstoned = (
rows["identified-oldest"],
rows["identified-latest-a"],
)
self.assertTrue(
all(response.deleted_at is not None for response in tombstoned)
)
self.assertTrue(
all(
response.deleted_at == response.updated_at
for response in tombstoned
)
)
self.assertEqual(
len({response.deleted_at for response in tombstoned}),
1,
)
self.assertIsNone(rows["anonymous-a"].deleted_at)
self.assertIsNone(rows["anonymous-b"].deleted_at)
self.assertEqual(
rows["already-deleted"].deleted_at,
historical_deleted_at.replace(tzinfo=None),
)
self.assertEqual(
rows["already-deleted"].updated_at,
historical_updated_at.replace(tzinfo=None),
)
session.add(
PollResponse(
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id="person-1",
answers=[],
submitted_at=latest + timedelta(hours=2),
)
)
with self.assertRaises(IntegrityError):
session.flush()
session.rollback()
session.add(
PollResponse(
tenant_id="tenant-1",
poll_id=poll_id,
respondent_id=None,
answers=[],
submitted_at=latest + timedelta(hours=2),
)
)
session.commit()
with engine.connect() as connection:
heads = set(
MigrationContext.configure(connection).get_current_heads()
)
indexes = {
item["name"]: item
for item in inspect(connection).get_indexes(
"poll_responses"
)
}
self.assertIn(_POLL_HEAD, heads)
self.assertIn(_INDEX_NAME, indexes)
self.assertTrue(indexes[_INDEX_NAME]["unique"])
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,32 @@
from __future__ import annotations
import unittest
from govoplan_core.core.poll_participation import (
CAPABILITY_POLL_PARTICIPATION_GATEWAY as CORE_CAPABILITY,
PollGovernedInvitationCommand as CorePollGovernedInvitationCommand,
PollParticipationGatewayProvider as CorePollParticipationGatewayProvider,
PollParticipationPolicy as CorePollParticipationPolicy,
)
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
from govoplan_poll.backend.participation import (
CAPABILITY_POLL_PARTICIPATION_GATEWAY,
PollGovernedInvitationCommand,
PollParticipationGatewayProvider,
PollParticipationPolicy,
)
class PollParticipationCompatibilityTests(unittest.TestCase):
def test_legacy_imports_are_exact_core_contract_re_exports(self) -> None:
self.assertIs(CORE_CAPABILITY, CAPABILITY_POLL_PARTICIPATION_GATEWAY)
self.assertIs(CorePollGovernedInvitationCommand, PollGovernedInvitationCommand)
self.assertIs(CorePollParticipationPolicy, PollParticipationPolicy)
self.assertIs(CorePollParticipationGatewayProvider, PollParticipationGatewayProvider)
def test_sql_provider_implements_the_core_contract(self) -> None:
self.assertIsInstance(SqlPollSchedulingProvider(), CorePollParticipationGatewayProvider)
if __name__ == "__main__":
unittest.main()

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,297 @@
from __future__ import annotations
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.core.poll import (
PollCapabilityError,
PollOptionOrderCommand,
PollOptionUpdateCommand,
PollResponseRetirementCommand,
PollResponseRetirementProvider,
PollSchedulingProvider,
)
from govoplan_core.db.base import Base
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
from govoplan_poll.backend.schemas import PollAnswerInput, PollCreateRequest, PollOptionInput, PollSubmitResponseRequest
from govoplan_poll.backend.service import create_poll, submit_poll_response
class PollResponseEditingTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(
self.engine,
tables=[Poll.__table__, PollOption.__table__, PollResponse.__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,
tables=[PollResponse.__table__, PollOption.__table__, Poll.__table__],
)
self.engine.dispose()
def _poll(self, *, allow_response_update: bool = True) -> Poll:
return create_poll(
self.session,
tenant_id="tenant-1",
user_id="organizer-1",
payload=PollCreateRequest(
title="Availability",
kind="availability",
status="open",
min_choices=1,
max_choices=2,
allow_response_update=allow_response_update,
options=[
PollOptionInput(
key="slot-1",
label="Monday",
value={"slot_id": "slot-1"},
metadata={"source": "scheduling"},
),
PollOptionInput(
key="slot-2",
label="Tuesday",
value={"slot_id": "slot-2"},
),
],
),
)
def _submit_both(self, poll: Poll, respondent_id: str) -> PollResponse:
return submit_poll_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(
respondent_id=respondent_id,
answers=[
PollAnswerInput(option_id=poll.options[0].id, value="available"),
PollAnswerInput(option_id=poll.options[1].id, value="maybe"),
],
),
)
def test_provider_projects_answers_and_option_change_invalidates_only_that_option(self) -> None:
poll = self._poll()
first = self._submit_both(poll, "person-1")
second = self._submit_both(poll, "person-2")
provider = SqlPollSchedulingProvider()
self.assertIsInstance(provider, PollSchedulingProvider)
projected = provider.list_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
)
self.assertEqual([answer.value for answer in projected[0].answers], ["available", "maybe"])
current = provider.get_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
respondent_ids=("person-1",),
)
self.assertIsNotNone(current)
self.assertEqual([answer.value for answer in current.answers], ["available", "maybe"])
option_ref = provider.update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(
label="Monday, revised",
value={"slot_id": "slot-1"},
metadata={"source": "scheduling"},
),
)
self.assertEqual(option_ref.id, poll.options[0].id)
self.assertEqual(first.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}])
self.assertEqual(second.answers, [{"option_id": poll.options[1].id, "option_key": "slot-2", "value": "maybe"}])
# Re-answer, then replay the exact option snapshot. No response data is
# invalidated because the option did not actually change.
first = self._submit_both(poll, "person-1")
provider.update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(
label="Monday, revised",
value={"slot_id": "slot-1"},
metadata={"source": "scheduling"},
),
)
self.assertEqual(len(first.answers), 2)
def test_provider_retires_response_from_live_results_but_keeps_audit_history(self) -> None:
poll = self._poll()
response = self._submit_both(poll, "person-1")
response.metadata_ = {
"invitation_id": "invitation-1",
"participant_email": "alice@example.test",
}
provider = SqlPollSchedulingProvider()
command = PollResponseRetirementCommand(
respondent_ids=("person-1", "alice@example.test"),
invitation_id="invitation-1",
reason="scheduling_participant_removed",
idempotency_key="scheduling:request-1:participant-1:removed",
metadata={
"source_module": "scheduling",
"source_resource_type": "participant",
"source_resource_id": "participant-1",
},
)
self.assertIsInstance(provider, PollResponseRetirementProvider)
retired = provider.retire_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=command,
)
self.assertEqual(retired.response_ids, (response.id,))
self.assertEqual(retired.newly_retired_count, 1)
self.assertFalse(retired.replayed)
self.assertIsNotNone(response.deleted_at)
self.assertEqual(len(response.answers), 2)
self.assertEqual(
response.metadata_["response_retirement"]["idempotency_key"],
command.idempotency_key,
)
self.assertEqual(
provider.list_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
),
(),
)
replayed = provider.retire_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=command,
)
self.assertTrue(replayed.replayed)
self.assertEqual(replayed.newly_retired_count, 0)
self.assertEqual(replayed.response_ids, (response.id,))
def test_fully_invalidated_response_is_no_longer_active_or_counted(self) -> None:
poll = self._poll()
response = submit_poll_response(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(
respondent_id="person-1",
answers=[PollAnswerInput(option_id=poll.options[0].id, value="available")],
),
)
SqlPollSchedulingProvider().update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(label="Monday, revised", value={"slot_id": "slot-1"}, metadata={"source": "scheduling"}),
)
self.assertEqual(response.answers, [])
self.assertIsNotNone(response.deleted_at)
self.assertEqual(
SqlPollSchedulingProvider().list_responses(self.session, tenant_id="tenant-1", poll_id=poll.id),
(),
)
def test_option_change_is_rejected_when_responses_cannot_be_updated(self) -> None:
poll = self._poll(allow_response_update=False)
response = self._submit_both(poll, "person-1")
with self.assertRaisesRegex(PollCapabilityError, "response updates are disabled"):
SqlPollSchedulingProvider().update_option(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
option_id=poll.options[0].id,
command=PollOptionUpdateCommand(
label="Monday, revised",
value={"slot_id": "slot-1"},
metadata={"source": "scheduling"},
),
)
self.assertEqual(poll.options[0].label, "Monday")
self.assertEqual(len(response.answers), 2)
def test_reorder_preserves_option_identity_and_existing_answers(self) -> None:
poll = self._poll(allow_response_update=False)
response = self._submit_both(poll, "person-1")
first_id, second_id = (option.id for option in poll.options)
original_answers = [dict(answer) for answer in response.answers]
provider = SqlPollSchedulingProvider()
reordered = provider.reorder_options(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=PollOptionOrderCommand(option_ids=(second_id, first_id)),
)
self.assertEqual(
[(option.id, option.position) for option in reordered.options],
[(second_id, 0), (first_id, 1)],
)
self.assertEqual(response.answers, original_answers)
self.assertIsNone(response.deleted_at)
replayed = provider.reorder_options(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=PollOptionOrderCommand(option_ids=(second_id, first_id)),
)
self.assertEqual(replayed.options, reordered.options)
self.assertEqual(response.answers, original_answers)
def test_reorder_rejects_partial_or_duplicate_active_option_order(self) -> None:
poll = self._poll()
first_id, second_id = (option.id for option in poll.options)
provider = SqlPollSchedulingProvider()
with self.assertRaisesRegex(PollCapabilityError, "every active option"):
provider.reorder_options(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=PollOptionOrderCommand(option_ids=(first_id,)),
)
with self.assertRaisesRegex(PollCapabilityError, "duplicate"):
provider.reorder_options(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
command=PollOptionOrderCommand(option_ids=(first_id, first_id)),
)
self.assertEqual(
[(option.id, option.position) for option in poll.options],
[(first_id, 0), (second_id, 1)],
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,353 @@
from __future__ import annotations
import os
import sqlite3
import threading
import unittest
import uuid
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import MagicMock, patch
from sqlalchemy import create_engine, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.db.base import Base, utcnow
from govoplan_poll.backend import service as poll_service
from govoplan_poll.backend.db.models import Poll, PollOption, PollResponse
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollCreateRequest,
PollOptionInput,
PollSubmitResponseRequest,
)
from govoplan_poll.backend.service import (
PollError,
_share_lock_poll_for_response,
create_poll,
submit_poll_response,
)
class PollResponseUniquenessTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__]
Base.metadata.create_all(self.engine, tables=self.tables)
self.Session = sessionmaker(bind=self.engine)
self.session: Session = self.Session()
def tearDown(self) -> None:
self.session.close()
Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables)))
self.engine.dispose()
def _poll(
self,
*,
allow_anonymous: bool = False,
allow_response_update: bool = True,
) -> Poll:
return create_poll(
self.session,
tenant_id="tenant-1",
user_id="owner-1",
payload=PollCreateRequest(
title="One response each",
kind="single_choice",
status="open",
allow_anonymous=allow_anonymous,
allow_response_update=allow_response_update,
options=[
PollOptionInput(key="yes", label="Yes"),
PollOptionInput(key="no", label="No"),
],
),
)
@staticmethod
def _payload(
respondent_id: str | None,
*,
option_key: str = "yes",
) -> PollSubmitResponseRequest:
return PollSubmitResponseRequest(
respondent_id=respondent_id,
respondent_label=respondent_id,
answers=[PollAnswerInput(option_key=option_key)],
)
def _hide_first_lookup(self):
original = poll_service._existing_response
calls = 0
def hide_once(session, *, poll, respondent_id):
nonlocal calls
calls += 1
if calls == 1:
return None
return original(
session,
poll=poll,
respondent_id=respondent_id,
)
return patch.object(poll_service, "_existing_response", side_effect=hide_once)
def test_sqlite_invariant_reconciles_a_racing_insert_to_normal_update(self) -> None:
poll = self._poll()
winner = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1"),
)
with self._hide_first_lookup():
reconciled = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1", option_key="no"),
)
self.assertEqual(reconciled.id, winner.id)
self.assertEqual(reconciled.answers[0]["option_key"], "no")
self.assertEqual(
self.session.query(PollResponse)
.filter(PollResponse.deleted_at.is_(None))
.count(),
1,
)
def test_racing_insert_obeys_update_disabled_policy(self) -> None:
poll = self._poll()
winner = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1"),
)
poll.allow_response_update = False
self.session.flush()
with self._hide_first_lookup():
with self.assertRaisesRegex(
PollError,
"Response updates are not allowed",
):
submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1", option_key="no"),
)
self.assertEqual(winner.answers[0]["option_key"], "yes")
self.assertEqual(self.session.query(PollResponse).count(), 1)
def test_anonymous_and_tombstoned_responses_do_not_conflict(self) -> None:
poll = self._poll(allow_anonymous=True)
anonymous_one = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload(None),
)
anonymous_two = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload(None, option_key="no"),
)
identified_one = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1"),
)
identified_one.deleted_at = utcnow()
self.session.flush()
identified_two = submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("person-1", option_key="no"),
)
self.assertNotEqual(anonymous_one.id, anonymous_two.id)
self.assertNotEqual(identified_one.id, identified_two.id)
self.assertEqual(self.session.query(PollResponse).count(), 4)
def test_unrelated_integrity_error_is_not_reconciled(self) -> None:
poll = self._poll()
self.session.execute(
text(
"""
CREATE TRIGGER reject_blocked_poll_response
BEFORE INSERT ON poll_responses
WHEN NEW.respondent_id = 'blocked'
BEGIN
SELECT RAISE(ABORT, 'blocked by unrelated invariant');
END
"""
)
)
with self.assertRaises(IntegrityError) as caught:
submit_poll_response(
self.session,
tenant_id=poll.tenant_id,
poll_id=poll.id,
payload=self._payload("blocked"),
)
self.assertIsInstance(caught.exception.orig, sqlite3.IntegrityError)
self.assertIn("blocked by unrelated invariant", str(caught.exception.orig))
def test_response_read_lock_is_shared(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 = _share_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(read=True)
def test_orm_mirrors_both_partial_index_predicates(self) -> None:
index = next(
index
for index in PollResponse.__table__.indexes
if index.name == "uq_poll_responses_active_respondent"
)
self.assertTrue(index.unique)
self.assertEqual(
str(index.dialect_options["sqlite"]["where"]),
"deleted_at IS NULL AND respondent_id IS NOT NULL",
)
self.assertEqual(
str(index.dialect_options["postgresql"]["where"]),
"deleted_at IS NULL AND respondent_id IS NOT NULL",
)
@unittest.skipUnless(
os.environ.get("GOVOPLAN_POLL_TEST_POSTGRES_URL"),
"set GOVOPLAN_POLL_TEST_POSTGRES_URL for the two-session PostgreSQL check",
)
class PollResponsePostgresConcurrencyTests(unittest.TestCase):
def setUp(self) -> None:
database_url = os.environ["GOVOPLAN_POLL_TEST_POSTGRES_URL"]
self.schema = f"poll_response_race_{uuid.uuid4().hex}"
self.admin_engine = create_engine(database_url)
with self.admin_engine.begin() as connection:
connection.execute(text(f'CREATE SCHEMA "{self.schema}"'))
self.engine = create_engine(
database_url,
connect_args={"options": f"-c search_path={self.schema}"},
)
self.tables = [Poll.__table__, PollOption.__table__, PollResponse.__table__]
Base.metadata.create_all(self.engine, tables=self.tables)
def tearDown(self) -> None:
try:
Base.metadata.drop_all(
self.engine,
tables=list(reversed(self.tables)),
)
finally:
self.engine.dispose()
with self.admin_engine.begin() as connection:
connection.execute(text(f'DROP SCHEMA "{self.schema}"'))
self.admin_engine.dispose()
def test_two_sessions_converge_on_one_active_response(self) -> None:
with Session(self.engine) as session:
poll = create_poll(
session,
tenant_id="tenant-1",
user_id="owner-1",
payload=PollCreateRequest(
title="Concurrent response",
kind="single_choice",
status="open",
options=[
PollOptionInput(key="yes", label="Yes"),
PollOptionInput(key="no", label="No"),
],
),
)
poll_id = poll.id
session.commit()
original = poll_service._existing_response
barrier = threading.Barrier(2)
thread_state = threading.local()
def synchronize_first_lookup(session, *, poll, respondent_id):
response = original(
session,
poll=poll,
respondent_id=respondent_id,
)
lookup_count = getattr(thread_state, "lookup_count", 0) + 1
thread_state.lookup_count = lookup_count
if lookup_count == 1:
self.assertIsNone(response)
barrier.wait(timeout=10)
return response
def submit(option_key: str) -> str:
with Session(self.engine) as session:
response = submit_poll_response(
session,
tenant_id="tenant-1",
poll_id=poll_id,
payload=PollSubmitResponseRequest(
respondent_id="person-1",
respondent_label="Person One",
answers=[PollAnswerInput(option_key=option_key)],
),
)
response_id = response.id
session.commit()
return response_id
with patch.object(
poll_service,
"_existing_response",
side_effect=synchronize_first_lookup,
):
with ThreadPoolExecutor(max_workers=2) as executor:
response_ids = tuple(
executor.map(submit, ("yes", "no"))
)
self.assertEqual(len(set(response_ids)), 1)
with Session(self.engine) as session:
active = (
session.query(PollResponse)
.filter(
PollResponse.poll_id == poll_id,
PollResponse.respondent_id == "person-1",
PollResponse.deleted_at.is_(None),
)
.all()
)
self.assertEqual(len(active), 1)
self.assertIn(active[0].answers[0]["option_key"], {"yes", "no"})
if __name__ == "__main__":
unittest.main()

View File

@@ -1,12 +1,15 @@
from __future__ import annotations
import unittest
from datetime import datetime, timezone
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.db.base import Base
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
from govoplan_core.core.poll import PollResponseRef, PollResponseSubmissionProvider, PollSchedulingProvider
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse
from govoplan_poll.backend.schemas import (
PollAnswerInput,
PollCreateRequest,
@@ -19,6 +22,7 @@ from govoplan_poll.backend.service import (
create_poll,
create_poll_invitation,
open_poll,
poll_owner_ref,
poll_result_summary_by_id,
submit_poll_response,
submit_poll_response_with_token,
@@ -30,7 +34,13 @@ class PollServiceTests(unittest.TestCase):
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(
self.engine,
tables=[Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__],
tables=[
Poll.__table__,
PollOption.__table__,
PollResponse.__table__,
PollInvitation.__table__,
PollLifecycleTransition.__table__,
],
)
self.Session = sessionmaker(bind=self.engine)
self.session: Session = self.Session()
@@ -39,10 +49,25 @@ class PollServiceTests(unittest.TestCase):
self.session.close()
Base.metadata.drop_all(
self.engine,
tables=[PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__],
tables=[
PollLifecycleTransition.__table__,
PollInvitation.__table__,
PollResponse.__table__,
PollOption.__table__,
Poll.__table__,
],
)
self.engine.dispose()
def test_response_projection_and_submission_extension_are_backward_compatible(self) -> None:
submitted_at = datetime(2026, 7, 20, tzinfo=timezone.utc)
legacy_ref = PollResponseRef(None, submitted_at)
provider = SqlPollSchedulingProvider()
self.assertIsNone(legacy_ref.respondent_id)
self.assertIsInstance(provider, PollSchedulingProvider)
self.assertIsInstance(provider, PollResponseSubmissionProvider)
def test_single_choice_response_can_update_existing_respondent(self) -> None:
poll = create_poll(
self.session,
@@ -66,13 +91,24 @@ class PollServiceTests(unittest.TestCase):
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
payload=PollSubmitResponseRequest(respondent_id="person-1", answers=[PollAnswerInput(option_key="b")]),
payload=PollSubmitResponseRequest(
respondent_id="person-1",
answers=[PollAnswerInput(option_key="b")],
metadata={"invitation_id": 42},
),
)
summary = poll_result_summary_by_id(self.session, tenant_id="tenant-1", poll_id=poll.id)
self.assertEqual(first.id, second.id)
self.assertEqual(summary["response_count"], 1)
self.assertEqual(summary["leading_option_ids"], [second.answers[0]["option_id"]])
response_ref = SqlPollSchedulingProvider().list_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
)[0]
self.assertIsNone(response_ref.invitation_id)
self.assertEqual(response_ref.respondent_id, "person-1")
def test_anonymous_response_requires_poll_policy(self) -> None:
poll = create_poll(
@@ -137,6 +173,11 @@ class PollServiceTests(unittest.TestCase):
],
options=[PollOptionInput(key="slot-1", label="Monday")],
),
mutation_owner=poll_owner_ref(
module_id="scheduling",
resource_type="scheduling_request",
resource_id="request-1",
),
)
self.assertEqual(poll.context_module, "scheduling")
@@ -267,13 +308,24 @@ class PollServiceTests(unittest.TestCase):
response = submit_poll_response_with_token(
self.session,
token=token,
payload=PollSubmitResponseRequest(answers=[PollAnswerInput(option_key="yes")]),
payload=PollSubmitResponseRequest(
answers=[PollAnswerInput(option_key="yes")],
metadata={"invitation_id": "forged-invitation", "client_note": "kept"},
),
)
self.assertEqual(response.respondent_id, f"invitation:{invitation.id}")
self.assertEqual(response.respondent_label, "External participant")
self.assertEqual(response.metadata_["invitation_id"], invitation.id)
self.assertEqual(response.metadata_["client_note"], "kept")
self.assertIsNotNone(invitation.last_used_at)
response_ref = SqlPollSchedulingProvider().list_responses(
self.session,
tenant_id="tenant-1",
poll_id=poll.id,
)[0]
self.assertEqual(response_ref.invitation_id, invitation.id)
self.assertEqual(response_ref.respondent_id, f"invitation:{invitation.id}")
if __name__ == "__main__":