148 lines
4.7 KiB
Python
148 lines
4.7 KiB
Python
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)
|