feat(poll): enforce auditable lifecycle transitions

2026-07-20 18:18:20 +02:00
parent f4d5cac40d
commit 65e2d1fd65
14 changed files with 1161 additions and 68 deletions
@@ -0,0 +1,183 @@
+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
+
+
+@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")
+            elif rule.target_status not in POLL_STATUSES:
+                raise ValueError(f"Poll transition rule {action!r} has an unknown target status")
+        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,
+        ),
+    }
+)
+
+
+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 not in POLL_STATUSES - {"archived"}:
+                raise ValueError("Archived poll has no valid status to restore")
+            target_status = archived_from_status
+        assert target_status is not None
+        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 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",
+]