GovOPlaN/govoplan-poll
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(poll): harden lifecycle policy boundaries

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-20 18:42:07 +02:00
parent 12b3175b07
commit 2c776437b4
5 changed files with 212 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/govoplan_poll/backend/schemas.py
View File

@@ -8,6 +8,7 @@ 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"]
@@ -31,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)

37
src/govoplan_poll/backend/service.py
View File

@@ -19,7 +19,7 @@ from govoplan_poll.backend.schemas import (
PollSubmitResponseRequest,
PollUpdateRequest,
)
from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, POLL_STATUSES
from govoplan_poll.backend.transitions import DEFAULT_POLL_TRANSITION_ENGINE, PollTransitionEngine
class PollError(ValueError):
@@ -27,6 +27,7 @@ class PollError(ValueError):
POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice", "availability"}
POLL_INITIAL_STATUSES = {"draft", "open"}
CHOICE_POLL_KINDS = {"single_choice", "multiple_choice", "yes_no", "yes_no_maybe", "ranked_choice"}
AVAILABILITY_VALUES = {"available", "maybe", "unavailable"}
YES_NO_OPTIONS = (
@@ -149,8 +150,8 @@ def _validate_choice_bounds(kind: str, min_choices: int, max_choices: int | None
def _ensure_valid_poll_payload(payload: PollCreateRequest) -> tuple[list[PollOptionInput], int, int | None]:
if payload.kind not in POLL_KINDS:
raise PollError(f"Unsupported poll kind: {payload.kind}")
if payload.status not in POLL_STATUSES:
raise PollError(f"Unsupported poll status: {payload.status}")
if payload.status not in POLL_INITIAL_STATUSES:
raise PollError("Poll initial status must be draft or open")
if payload.opens_at is not None and payload.closes_at is not None and payload.closes_at <= payload.opens_at:
raise PollError("closes_at must be after opens_at")
options = _normalize_options(payload.kind, payload.options)
@@ -305,7 +306,9 @@ def poll_results_are_visible(
if poll.result_visibility == "public":
return True
if poll.result_visibility == "after_close":
if poll.status in {"closed", "decided", "archived"}:
if poll.status in {"closed", "decided"}:
return True
if poll.status == "archived" and poll.archived_from_status in {"closed", "decided"}:
return True
return poll.closes_at is not None and response_datetime(poll.closes_at) <= _now()
if poll.result_visibility == "after_response" and ids:
@@ -431,6 +434,7 @@ def transition_poll(
idempotency_key: str | None = None,
actor_user_id: str | None = None,
actor_api_key_id: str | None = None,
transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
) -> PollTransitionResult:
"""Apply one policy-controlled transition and append its durable audit record."""
@@ -454,7 +458,7 @@ def transition_poll(
return PollTransitionResult(poll=poll, transition=existing, replayed=True)
try:
plan = DEFAULT_POLL_TRANSITION_ENGINE.plan(
plan = transition_engine.plan(
current_status=poll.status,
action=action,
archived_from_status=poll.archived_from_status,
@@ -464,6 +468,14 @@ def transition_poll(
previous_decision_option_id = poll.decided_option_id
now = _now()
used_legacy_unarchive_fallback = action == "unarchive" and poll.archived_from_status is None
cleared_expired_closes_at: datetime | None = None
if action == "open" and poll.closes_at is not None:
normalized_closes_at = response_datetime(poll.closes_at)
if normalized_closes_at is not None and normalized_closes_at <= now:
cleared_expired_closes_at = normalized_closes_at
poll.closes_at = None
if action == "archive":
poll.archived_from_status = plan.from_status
elif action == "unarchive":
@@ -483,6 +495,13 @@ def transition_poll(
transition_metadata["archived_from_status"] = plan.from_status
elif action == "unarchive":
transition_metadata["restored_status"] = plan.to_status
if used_legacy_unarchive_fallback:
transition_metadata["legacy_restore_fallback"] = {
"reason": "missing_archived_from_status",
"restored_status": plan.to_status,
}
if cleared_expired_closes_at is not None:
transition_metadata["cleared_expired_closes_at"] = cleared_expired_closes_at.isoformat()
transition = PollLifecycleTransition(
tenant_id=tenant_id,
poll_id=poll.id,
@@ -913,8 +932,12 @@ def poll_lifecycle_transition_response(transition: PollLifecycleTransition) -> d
}
def poll_response(poll: Poll) -> dict[str, Any]:
lifecycle_actions = DEFAULT_POLL_TRANSITION_ENGINE.available_actions(
def poll_response(
poll: Poll,
*,
transition_engine: PollTransitionEngine = DEFAULT_POLL_TRANSITION_ENGINE,
) -> dict[str, Any]:
lifecycle_actions = transition_engine.available_actions(
current_status=poll.status,
archived_from_status=poll.archived_from_status,
)

19
src/govoplan_poll/backend/transitions.py
View File

@@ -22,6 +22,7 @@ class PollTransitionRule:
source_statuses: frozenset[str]
target_status: str | None
restore_archived_status: bool = False
missing_restore_target_status: str | None = None
@dataclass(frozen=True)
@@ -56,8 +57,12 @@ class PollTransitionPolicy:
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
@@ -97,6 +102,7 @@ DEFAULT_POLL_TRANSITION_POLICY = PollTransitionPolicy(
source_statuses=frozenset({"archived"}),
target_status=None,
restore_archived_status=True,
missing_restore_target_status="draft",
),
}
)
@@ -125,9 +131,12 @@ class PollTransitionEngine:
target_status = rule.target_status
if rule.restore_archived_status:
if archived_from_status not in POLL_STATUSES - {"archived"}:
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")
target_status = archived_from_status
else:
target_status = archived_from_status
assert target_status is not None
return PollTransitionPlan(action=action, from_status=current_status, to_status=target_status)
@@ -140,7 +149,11 @@ class PollTransitionEngine:
actions: list[PollTransitionAvailability] = []
for action in POLL_TRANSITION_ACTIONS:
rule = self.policy.rules[action]
target_status = archived_from_status if rule.restore_archived_status else rule.target_status
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,