459 lines
17 KiB
Python
459 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.approvals import (
|
|
ApprovalActorSelector,
|
|
ApprovalDecisionCommand,
|
|
ApprovalRequestCreateCommand,
|
|
ApprovalStepDefinition,
|
|
ApprovalTemplateCreateCommand,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_approvals.backend.service import ApprovalStoreError, SqlApprovalRequests
|
|
from govoplan_core.core.tasks import WorkItemQuery
|
|
from govoplan_approvals.backend.work_items import ApprovalWorkItemProvider
|
|
|
|
|
|
DIGEST = "a" * 64
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str
|
|
account_id: str
|
|
group_ids: tuple[str, ...] = ()
|
|
role_ids: tuple[str, ...] = ()
|
|
function_assignment_ids: tuple[str, ...] = ()
|
|
acting_for_account_id: str | None = None
|
|
acting_assignment_id: str | None = None
|
|
|
|
|
|
def request_command() -> ApprovalRequestCreateCommand:
|
|
return ApprovalRequestCreateCommand(
|
|
title="Approve Campaign delivery",
|
|
subject_module="campaigns",
|
|
subject_type="campaign_version",
|
|
subject_id="campaign-1",
|
|
subject_version="version-7",
|
|
subject_digest=DIGEST,
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
key="review",
|
|
label="Review",
|
|
selectors=(ApprovalActorSelector("group", "reviewers"),),
|
|
),
|
|
ApprovalStepDefinition(
|
|
key="release",
|
|
label="Release",
|
|
selectors=(ApprovalActorSelector("role", "senders"),),
|
|
signature_required=True,
|
|
),
|
|
),
|
|
separation_of_duties=True,
|
|
unique_actors_across_steps=True,
|
|
)
|
|
|
|
|
|
class ApprovalRuntimeTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
Base.metadata.create_all(self.engine)
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.service = SqlApprovalRequests()
|
|
self.requester = Principal("tenant-1", "requester")
|
|
|
|
def tearDown(self) -> None:
|
|
Base.metadata.drop_all(self.engine)
|
|
self.engine.dispose()
|
|
|
|
def test_sequential_chain_exact_subject_and_signature(self) -> None:
|
|
with self.Session() as session:
|
|
created = self.service.create_request(
|
|
session,
|
|
self.requester,
|
|
command=request_command(),
|
|
idempotency_key="create-1",
|
|
)
|
|
review = self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "reviewer", group_ids=("reviewers",)),
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand("approved", "Reviewed.", 1, "review-1"),
|
|
)
|
|
self.assertEqual(2, review.revision)
|
|
current = self.service.get_request(
|
|
session, self.requester, request_id=created.id
|
|
)
|
|
self.assertEqual("release", current["current_step_key"])
|
|
with self.assertRaisesRegex(ApprovalStoreError, "signature"):
|
|
self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "sender", role_ids=("senders",)),
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand(
|
|
"approved", "Release.", 2, "release-no-signature"
|
|
),
|
|
)
|
|
released = self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "sender", role_ids=("senders",)),
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand(
|
|
"approved",
|
|
"Release.",
|
|
2,
|
|
"release-1",
|
|
signature_ref={"provider": "signatures", "id": "sig-1"},
|
|
),
|
|
)
|
|
self.assertEqual(3, released.revision)
|
|
check = self.service.check_approved(
|
|
session,
|
|
self.requester,
|
|
request_id=created.id,
|
|
subject_module="campaigns",
|
|
subject_type="campaign_version",
|
|
subject_id="campaign-1",
|
|
subject_version="version-7",
|
|
subject_digest=DIGEST,
|
|
)
|
|
self.assertTrue(check.approved)
|
|
with self.assertRaisesRegex(ApprovalStoreError, "exact requested subject"):
|
|
self.service.check_approved(
|
|
session,
|
|
self.requester,
|
|
request_id=created.id,
|
|
subject_module="campaigns",
|
|
subject_type="campaign_version",
|
|
subject_id="campaign-1",
|
|
subject_version="version-8",
|
|
subject_digest=DIGEST,
|
|
)
|
|
|
|
def test_separation_of_duties_rejection_replay_and_tenant_isolation(self) -> None:
|
|
direct = ApprovalRequestCreateCommand(
|
|
title="Direct review",
|
|
subject_module="campaigns",
|
|
subject_type="campaign_version",
|
|
subject_id="campaign-1",
|
|
subject_version="version-7",
|
|
subject_digest=DIGEST,
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"review", "Review", (ApprovalActorSelector("account", "requester"),)
|
|
),
|
|
),
|
|
)
|
|
with self.Session() as session:
|
|
created = self.service.create_request(
|
|
session, self.requester, command=direct, idempotency_key="create-direct"
|
|
)
|
|
with self.assertRaisesRegex(ApprovalStoreError, "separation of duties"):
|
|
self.service.decide(
|
|
session,
|
|
self.requester,
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand("approved", "Self.", 1, "self-1"),
|
|
)
|
|
self.assertIsNone(
|
|
self.service.get_request(
|
|
session, Principal("tenant-2", "requester"), request_id=created.id
|
|
)
|
|
)
|
|
|
|
rejected_command = ApprovalRequestCreateCommand(
|
|
title="Review",
|
|
subject_module="cases",
|
|
subject_type="case",
|
|
subject_id="case-1",
|
|
subject_version="1",
|
|
subject_digest=DIGEST,
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"review",
|
|
"Review",
|
|
(ApprovalActorSelector("account", "reviewer"),),
|
|
),
|
|
),
|
|
)
|
|
rejected = self.service.create_request(
|
|
session,
|
|
self.requester,
|
|
command=rejected_command,
|
|
idempotency_key="create-reject",
|
|
)
|
|
receipt = self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "reviewer"),
|
|
request_id=rejected.id,
|
|
command=ApprovalDecisionCommand(
|
|
"rejected", "Insufficient evidence.", 1, "reject-1"
|
|
),
|
|
)
|
|
replay = self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "reviewer"),
|
|
request_id=rejected.id,
|
|
command=ApprovalDecisionCommand(
|
|
"rejected", "Insufficient evidence.", 1, "reject-1"
|
|
),
|
|
)
|
|
self.assertEqual(receipt.receipt_sha256, replay.receipt_sha256)
|
|
self.assertTrue(replay.replayed)
|
|
self.assertEqual(
|
|
"rejected",
|
|
self.service.get_request(
|
|
session, self.requester, request_id=rejected.id
|
|
)["state"],
|
|
)
|
|
|
|
def test_due_step_escalates_explicitly(self) -> None:
|
|
command = ApprovalRequestCreateCommand(
|
|
title="Due review",
|
|
subject_module="files",
|
|
subject_type="file",
|
|
subject_id="file-1",
|
|
subject_version="1",
|
|
subject_digest=DIGEST,
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"review",
|
|
"Review",
|
|
(ApprovalActorSelector("account", "reviewer"),),
|
|
due_at=datetime.now(UTC) - timedelta(minutes=1),
|
|
),
|
|
),
|
|
)
|
|
with self.Session() as session:
|
|
created = self.service.create_request(
|
|
session, self.requester, command=command, idempotency_key="create-due"
|
|
)
|
|
escalated = self.service.escalate_due(
|
|
session,
|
|
self.requester,
|
|
request_id=created.id,
|
|
expected_revision=1,
|
|
idempotency_key="escalate-1",
|
|
)
|
|
self.assertEqual("escalated", escalated.state)
|
|
|
|
def test_work_provider_reuses_exact_decision_eligibility(self) -> None:
|
|
with self.Session() as session:
|
|
created = self.service.create_request(
|
|
session,
|
|
self.requester,
|
|
command=request_command(),
|
|
idempotency_key="work-provider-request",
|
|
)
|
|
reviewer = ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="reviewer",
|
|
membership_id="membership-reviewer",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(
|
|
{
|
|
"approvals:workspace:read",
|
|
"approvals:workspace:decide",
|
|
}
|
|
),
|
|
group_ids=frozenset({"reviewers"}),
|
|
),
|
|
account=object(),
|
|
user=object(),
|
|
)
|
|
provider = ApprovalWorkItemProvider()
|
|
|
|
page = provider.list_items(
|
|
session,
|
|
reviewer,
|
|
query=WorkItemQuery(tenant_id="tenant-1"),
|
|
)
|
|
self.assertEqual(1, page.total)
|
|
self.assertEqual(created.id, page.items[0].id)
|
|
self.assertEqual("Review", page.items[0].required_action)
|
|
self.assertEqual("campaign_version", page.items[0].sources[0].resource_type)
|
|
|
|
self.service.decide(
|
|
session,
|
|
reviewer,
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand(
|
|
"approved",
|
|
"Reviewed.",
|
|
1,
|
|
"work-provider-decision",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
0,
|
|
provider.list_items(
|
|
session,
|
|
reviewer,
|
|
query=WorkItemQuery(tenant_id="tenant-1"),
|
|
).total,
|
|
)
|
|
|
|
def test_template_and_evidence_role_constraints_are_frozen(self) -> None:
|
|
template_command = ApprovalTemplateCreateCommand(
|
|
key="campaign-release",
|
|
title="Campaign release",
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"release",
|
|
"Release",
|
|
(ApprovalActorSelector("role", "senders"),),
|
|
forbidden_evidence_roles=("builder",),
|
|
),
|
|
),
|
|
)
|
|
with self.Session() as session:
|
|
draft = self.service.create_template(
|
|
session,
|
|
self.requester,
|
|
command=template_command,
|
|
idempotency_key="template-1",
|
|
)
|
|
published = self.service.publish_template(
|
|
session,
|
|
self.requester,
|
|
template_id=draft.id,
|
|
expected_revision=1,
|
|
idempotency_key="publish-1",
|
|
)
|
|
request = ApprovalRequestCreateCommand(
|
|
title="Approve exact execution",
|
|
subject_module="campaigns",
|
|
subject_type="campaign_execution",
|
|
subject_id="campaign-1",
|
|
subject_version="build-7",
|
|
subject_digest=DIGEST,
|
|
steps=(),
|
|
evidence_actors={"builder": ("builder",)},
|
|
template_id=published.id,
|
|
template_revision=published.revision,
|
|
)
|
|
created = self.service.create_request(
|
|
session,
|
|
self.requester,
|
|
command=request,
|
|
idempotency_key="templated-request-1",
|
|
)
|
|
with self.assertRaisesRegex(ApprovalStoreError, "builder actor"):
|
|
self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "builder", role_ids=("senders",)),
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand(
|
|
"approved", "Built and release attempted.", 1, "builder-release"
|
|
),
|
|
)
|
|
receipt = self.service.decide(
|
|
session,
|
|
Principal("tenant-1", "sender", role_ids=("senders",)),
|
|
request_id=created.id,
|
|
command=ApprovalDecisionCommand(
|
|
"approved", "Independent release.", 1, "sender-release"
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
"role", receipt.authority_provenance["matched_selector"]["kind"]
|
|
)
|
|
with self.assertRaisesRegex(ApprovalStoreError, "content digest"):
|
|
self.service.check_approved(
|
|
session,
|
|
self.requester,
|
|
request_id=created.id,
|
|
subject_module="campaigns",
|
|
subject_type="campaign_execution",
|
|
subject_id="campaign-1",
|
|
subject_version="build-7",
|
|
subject_digest="b" * 64,
|
|
)
|
|
|
|
def test_template_history_and_structural_compare_are_tenant_bound(self) -> None:
|
|
original = ApprovalTemplateCreateCommand(
|
|
key="monthly-release",
|
|
title="Monthly release",
|
|
description="Initial process",
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"review",
|
|
"Review",
|
|
(ApprovalActorSelector("role", "reviewers"),),
|
|
),
|
|
),
|
|
)
|
|
revised_command = ApprovalTemplateCreateCommand(
|
|
key="monthly-release",
|
|
title="Monthly release approval",
|
|
description="Initial process",
|
|
steps=(
|
|
ApprovalStepDefinition(
|
|
"review",
|
|
"Independent review",
|
|
(ApprovalActorSelector("role", "reviewers"),),
|
|
),
|
|
),
|
|
)
|
|
with self.Session() as session:
|
|
created = self.service.create_template(
|
|
session,
|
|
self.requester,
|
|
command=original,
|
|
idempotency_key="history-template-1",
|
|
)
|
|
revised = self.service.revise_template(
|
|
session,
|
|
self.requester,
|
|
template_id=created.id,
|
|
command=revised_command,
|
|
expected_revision=1,
|
|
idempotency_key="history-template-2",
|
|
)
|
|
self.assertEqual(2, revised.revision)
|
|
|
|
history = self.service.template_history(
|
|
session,
|
|
self.requester,
|
|
template_id=created.id,
|
|
)
|
|
self.assertEqual([2, 1], [item["revision"] for item in history])
|
|
self.assertIsNotNone(history[1]["superseded_at"])
|
|
self.assertEqual("requester", history[0]["actor_id"])
|
|
|
|
comparison = self.service.compare_template_revisions(
|
|
session,
|
|
self.requester,
|
|
template_id=created.id,
|
|
from_revision=1,
|
|
to_revision=2,
|
|
)
|
|
changes = {item["path"]: item for item in comparison["changes"]}
|
|
self.assertEqual(
|
|
"Monthly release",
|
|
changes["/title"]["before"],
|
|
)
|
|
self.assertEqual(
|
|
"Independent review",
|
|
changes["/steps/0/label"]["after"],
|
|
)
|
|
|
|
with self.assertRaises(LookupError):
|
|
self.service.template_history(
|
|
session,
|
|
Principal("tenant-2", "other"),
|
|
template_id=created.id,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|