Project pending approvals as authorized work

src/govoplan_approvals/backend/service.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
-from collections.abc import Mapping
-from dataclasses import asdict
+from collections.abc import Mapping, Sequence
+from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
 import hashlib
 import json
@@ -33,6 +33,13 @@ class ApprovalStoreError(ValueError):
     pass
 
 
+@dataclass(frozen=True, slots=True)
+class ApprovalDecisionContext:
+    step: Mapping[str, Any]
+    effective_actor: str
+    matched_selector: Mapping[str, object]
+
+
 class SqlApprovalRequests:
     def create_template(
         self,
@@ -278,9 +285,7 @@ class SqlApprovalRequests:
             .filter(
                 ApprovalTemplateRevision.tenant_id == _tenant(principal),
                 ApprovalTemplateRevision.template_id == template_id,
-                ApprovalTemplateRevision.revision.in_(
-                    (from_revision, to_revision)
-                ),
+                ApprovalTemplateRevision.revision.in_((from_revision, to_revision)),
             )
             .all()
         )
@@ -480,56 +485,17 @@ class SqlApprovalRequests:
             raise ApprovalStoreError(
                 "This Approval request no longer accepts decisions."
             )
-        expires_at = _datetime(current.payload.get("expires_at"))
-        if expires_at is not None and _now() >= expires_at:
-            raise ApprovalStoreError("This Approval request has expired.")
+        decision_context = approval_decision_context(
+            typed_session,
+            principal,
+            current,
+            delegated_for_account_id=command.delegated_for_account_id,
+        )
         steps = list(current.payload["steps"])
         step_index = int(current.payload.get("current_step_index") or 0)
-        step = steps[step_index]
-        effective_actor = _effective_actor(principal, command.delegated_for_account_id)
-        matched_selector = _matched_selector(
-            principal, step.get("selectors") or [], effective_actor
-        )
-        if matched_selector is None:
-            raise ApprovalStoreError(
-                "The current principal is not eligible for this Approval step."
-            )
-        if bool(
-            current.payload.get("separation_of_duties")
-        ) and effective_actor == current.payload.get("requested_by"):
-            raise ApprovalStoreError(
-                "Approval separation of duties prevents requester self-approval."
-            )
-        prior = (
-            typed_session.query(ApprovalDecisionRecord)
-            .filter(
-                ApprovalDecisionRecord.tenant_id == tenant_id,
-                ApprovalDecisionRecord.request_id == request_id,
-                ApprovalDecisionRecord.effective_actor_id == effective_actor,
-            )
-            .all()
-        )
-        if any(item.step_key == step["key"] for item in prior):
-            raise ApprovalStoreError(
-                "This actor has already decided the current Approval step."
-            )
-        if bool(current.payload.get("unique_actors_across_steps")) and any(
-            item.outcome == "approved" for item in prior
-        ):
-            raise ApprovalStoreError(
-                "Approval policy requires a different actor for each step."
-            )
-        evidence_actors = {
-            str(key): {str(actor) for actor in (actors or [])}
-            for key, actors in dict(
-                current.payload.get("evidence_actors") or {}
-            ).items()
-        }
-        for role in step.get("forbidden_evidence_roles") or []:
-            if effective_actor in evidence_actors.get(str(role), set()):
-                raise ApprovalStoreError(
-                    f"Approval separation of duties prevents the {role} actor from deciding this step."
-                )
+        step = decision_context.step
+        effective_actor = decision_context.effective_actor
+        matched_selector = decision_context.matched_selector
         signature_ref = (
             dict(command.signature_ref) if command.signature_ref is not None else None
         )
@@ -764,6 +730,79 @@ class SqlApprovalRequests:
         )
 
 
+def approval_decision_context(
+    session: Session,
+    principal: object,
+    request: ApprovalRequestRevision,
+    *,
+    delegated_for_account_id: str | None = None,
+    prior_decisions: Sequence[ApprovalDecisionRecord] | None = None,
+) -> ApprovalDecisionContext:
+    if request.state not in {"pending", "escalated"}:
+        raise ApprovalStoreError("This Approval request no longer accepts decisions.")
+    expires_at = _datetime(request.payload.get("expires_at"))
+    if expires_at is not None and _now() >= expires_at:
+        raise ApprovalStoreError("This Approval request has expired.")
+    steps = list(request.payload.get("steps") or ())
+    step_index = int(request.payload.get("current_step_index") or 0)
+    if step_index < 0 or step_index >= len(steps):
+        raise ApprovalStoreError("The current Approval step is unavailable.")
+    step = steps[step_index]
+    effective_actor = _effective_actor(principal, delegated_for_account_id)
+    matched_selector = _matched_selector(
+        principal,
+        list(step.get("selectors") or ()),
+        effective_actor,
+    )
+    if matched_selector is None:
+        raise ApprovalStoreError(
+            "The current principal is not eligible for this Approval step."
+        )
+    if bool(
+        request.payload.get("separation_of_duties")
+    ) and effective_actor == request.payload.get("requested_by"):
+        raise ApprovalStoreError(
+            "Approval separation of duties prevents requester self-approval."
+        )
+    prior = (
+        list(prior_decisions)
+        if prior_decisions is not None
+        else (
+            session.query(ApprovalDecisionRecord)
+            .filter(
+                ApprovalDecisionRecord.tenant_id == request.tenant_id,
+                ApprovalDecisionRecord.request_id == request.request_id,
+                ApprovalDecisionRecord.effective_actor_id == effective_actor,
+            )
+            .all()
+        )
+    )
+    if any(item.step_key == step["key"] for item in prior):
+        raise ApprovalStoreError(
+            "This actor has already decided the current Approval step."
+        )
+    if bool(request.payload.get("unique_actors_across_steps")) and any(
+        item.outcome == "approved" for item in prior
+    ):
+        raise ApprovalStoreError(
+            "Approval policy requires a different actor for each step."
+        )
+    evidence_actors = {
+        str(key): {str(actor) for actor in (actors or [])}
+        for key, actors in dict(request.payload.get("evidence_actors") or {}).items()
+    }
+    for role in step.get("forbidden_evidence_roles") or ():
+        if effective_actor in evidence_actors.get(str(role), set()):
+            raise ApprovalStoreError(
+                f"Approval separation of duties prevents the {role} actor from deciding this step."
+            )
+    return ApprovalDecisionContext(
+        step=step,
+        effective_actor=effective_actor,
+        matched_selector=matched_selector,
+    )
+
+
 def _step_payload(step: object) -> dict[str, Any]:
     return {
         "key": str(getattr(step, "key")),
@@ -1383,4 +1422,9 @@ def _now() -> datetime:
     return datetime.now(UTC)
 
 
-__all__ = ["ApprovalStoreError", "SqlApprovalRequests"]
+__all__ = [
+    "ApprovalDecisionContext",
+    "ApprovalStoreError",
+    "SqlApprovalRequests",
+    "approval_decision_context",
+]