Add institutional provenance and approval gates
This commit is contained in:
@@ -0,0 +1,147 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from govoplan_core.core.approvals import (
|
||||
ApprovalActorSelector,
|
||||
ApprovalCheck,
|
||||
ApprovalRequestRef,
|
||||
ApprovalStepDefinition,
|
||||
)
|
||||
from govoplan_campaign.backend.approval_gate import (
|
||||
CampaignApprovalGateError,
|
||||
assert_campaign_approval,
|
||||
request_campaign_approval,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Principal:
|
||||
tenant_id: str = "tenant-1"
|
||||
account_id: str = "requester"
|
||||
|
||||
|
||||
class Session:
|
||||
def add(self, _value: object) -> None:
|
||||
return None
|
||||
|
||||
def flush(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class ApprovalStub:
|
||||
available = True
|
||||
|
||||
def __init__(self, *, state: str = "pending") -> None:
|
||||
self.state = state
|
||||
self.command = None
|
||||
|
||||
def create_request(self, _session, _principal, *, command, idempotency_key):
|
||||
assert idempotency_key == "request-1"
|
||||
self.command = command
|
||||
return ApprovalRequestRef("approval-1", 1, "pending", "release")
|
||||
|
||||
def check_approved(self, _session, _principal, **kwargs):
|
||||
return ApprovalCheck(
|
||||
request_id=str(kwargs["request_id"]),
|
||||
revision=2,
|
||||
state=self.state,
|
||||
approved=self.state == "approved",
|
||||
subject_module=str(kwargs["subject_module"]),
|
||||
subject_type=str(kwargs["subject_type"]),
|
||||
subject_id=str(kwargs["subject_id"]),
|
||||
subject_version=kwargs["subject_version"],
|
||||
subject_digest=str(kwargs["subject_digest"]),
|
||||
)
|
||||
|
||||
|
||||
def _objects():
|
||||
campaign = SimpleNamespace(
|
||||
id="campaign-1",
|
||||
tenant_id="tenant-1",
|
||||
created_by_user_id="author",
|
||||
owner_user_id="owner",
|
||||
)
|
||||
version = SimpleNamespace(
|
||||
id="version-1",
|
||||
campaign_id="campaign-1",
|
||||
version_number=7,
|
||||
execution_snapshot_hash="a" * 64,
|
||||
editor_state={"review_send": {"updated_by_user_id": "reviewer"}},
|
||||
validation_summary={"validated_by_user_id": "validator"},
|
||||
build_summary={"built_by_user_id": "builder"},
|
||||
locked_by_user_id="validator",
|
||||
)
|
||||
return campaign, version
|
||||
|
||||
|
||||
def test_request_binds_snapshot_and_action_evidence() -> None:
|
||||
campaign, version = _objects()
|
||||
provider = ApprovalStub()
|
||||
snapshot = SimpleNamespace(build_token="build-7", snapshot_version="7")
|
||||
with (
|
||||
patch(
|
||||
"govoplan_campaign.backend.approval_gate.ensure_execution_snapshot",
|
||||
return_value=snapshot,
|
||||
),
|
||||
patch(
|
||||
"govoplan_campaign.backend.approval_gate.approvals_integration",
|
||||
return_value=provider,
|
||||
),
|
||||
):
|
||||
request_campaign_approval(
|
||||
Session(),
|
||||
Principal(),
|
||||
campaign=campaign,
|
||||
version=version,
|
||||
title="Release",
|
||||
description=None,
|
||||
steps=(
|
||||
ApprovalStepDefinition(
|
||||
"release",
|
||||
"Release",
|
||||
(ApprovalActorSelector("role", "sender"),),
|
||||
forbidden_evidence_roles=("builder", "reviewer"),
|
||||
),
|
||||
),
|
||||
idempotency_key="request-1",
|
||||
)
|
||||
assert provider.command.subject_digest == "a" * 64
|
||||
assert provider.command.evidence_actors["builder"] == ("builder",)
|
||||
assert provider.command.evidence_actors["reviewer"] == ("reviewer",)
|
||||
assert version.editor_state["approval_gate"]["request_id"] == "approval-1"
|
||||
|
||||
|
||||
def test_gate_fails_closed_until_exact_request_is_approved() -> None:
|
||||
campaign, version = _objects()
|
||||
version.editor_state["approval_gate"] = {
|
||||
"request_id": "approval-1",
|
||||
"request_revision": 1,
|
||||
"subject_version": "build-7",
|
||||
"subject_digest": "a" * 64,
|
||||
"requested_at": "2026-08-01T00:00:00+00:00",
|
||||
"requested_by_user_id": "requester",
|
||||
}
|
||||
snapshot = SimpleNamespace(build_token="build-7", snapshot_version="7")
|
||||
provider = ApprovalStub(state="pending")
|
||||
with (
|
||||
patch(
|
||||
"govoplan_campaign.backend.approval_gate.ensure_execution_snapshot",
|
||||
return_value=snapshot,
|
||||
),
|
||||
patch(
|
||||
"govoplan_campaign.backend.approval_gate.approvals_integration",
|
||||
return_value=provider,
|
||||
),
|
||||
):
|
||||
with pytest.raises(CampaignApprovalGateError, match="pending"):
|
||||
assert_campaign_approval(Session(), tenant_id="tenant-1", version=version)
|
||||
provider.state = "approved"
|
||||
assert_campaign_approval(Session(), tenant_id="tenant-1", version=version)
|
||||
version.execution_snapshot_hash = "b" * 64
|
||||
with pytest.raises(CampaignApprovalGateError, match="changed"):
|
||||
assert_campaign_approval(Session(), tenant_id="tenant-1", version=version)
|
||||
@@ -38,7 +38,7 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None:
|
||||
actual = _operation_keys(router)
|
||||
|
||||
assert actual == expected
|
||||
assert len(actual) == 64
|
||||
assert len(actual) == 65
|
||||
assert not [operation for operation, count in Counter(actual).items() if count > 1]
|
||||
|
||||
|
||||
@@ -50,6 +50,13 @@ def test_key_routes_are_owned_by_their_focused_router() -> None:
|
||||
(reports_router, ("GET", "/campaigns/{campaign_id}/report")),
|
||||
(sharing_router, ("POST", "/campaigns/{campaign_id}/shares")),
|
||||
(delivery_router, ("POST", "/campaigns/{campaign_id}/send-now")),
|
||||
(
|
||||
delivery_router,
|
||||
(
|
||||
"POST",
|
||||
"/campaigns/{campaign_id}/versions/{version_id}/approval-request",
|
||||
),
|
||||
),
|
||||
(
|
||||
attachments_router,
|
||||
(
|
||||
|
||||
Reference in New Issue
Block a user