237 lines
8.0 KiB
Python
237 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.agent_core import harness
|
|
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
|
|
|
|
|
class _ProjectStore:
|
|
def active_world_key(self) -> str:
|
|
return "shared-approval-project"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolated_policy_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
import server.state.projects as projects_module
|
|
|
|
for env_name in (
|
|
"APS_APPROVAL_ROLE_POLICIES",
|
|
"APS_APPROVAL_INITIATOR_ROLES",
|
|
"APS_APPROVAL_ROLES",
|
|
"APS_APPROVAL_VIEW_ROLES",
|
|
):
|
|
monkeypatch.delenv(env_name, raising=False)
|
|
monkeypatch.setattr(projects_module, "get_project_store", lambda: _ProjectStore())
|
|
harness.configure_approval_store(str(tmp_path / "approvals.json"))
|
|
yield
|
|
harness._approval_store.clear()
|
|
|
|
|
|
def _identity(user_id: int, role: str) -> IdentityContext:
|
|
return IdentityContext(
|
|
user_id,
|
|
f"{role}-{user_id}",
|
|
f"{role} {user_id}",
|
|
"platform",
|
|
roles=(role,),
|
|
)
|
|
|
|
|
|
def _as(identity: IdentityContext, callback):
|
|
token = bind_identity(identity)
|
|
try:
|
|
return callback()
|
|
finally:
|
|
reset_identity(token)
|
|
|
|
|
|
def _stage(action: str) -> str:
|
|
block = harness.stage_confirmation(
|
|
"role-policy-test",
|
|
action,
|
|
{"versionId": 1},
|
|
title="Role policy test",
|
|
summary_lines=["test"],
|
|
)
|
|
return str(block.props["confirmId"])
|
|
|
|
|
|
def _set_policy(monkeypatch: pytest.MonkeyPatch, policy: dict) -> None:
|
|
monkeypatch.setenv(
|
|
"APS_APPROVAL_ROLE_POLICIES",
|
|
json.dumps(policy, ensure_ascii=True, separators=(",", ":")),
|
|
)
|
|
|
|
|
|
def _policy_entry(action: str) -> dict:
|
|
return next(item for item in harness.list_policy() if item["action"] == action)
|
|
|
|
|
|
def test_exact_action_overrides_power_per_capability(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_set_policy(monkeypatch, {
|
|
"powers": {
|
|
"P2": {
|
|
"initiate": ["p2-initiator"],
|
|
"approve": ["p2-approver"],
|
|
"history": ["p2-auditor"],
|
|
},
|
|
},
|
|
"actions": {
|
|
"schedule.publish": {
|
|
"approve": ["release-approver"],
|
|
},
|
|
},
|
|
})
|
|
|
|
confirm_id = _as(_identity(1, "p2-initiator"), lambda: _stage("schedule.publish"))
|
|
assert _as(
|
|
_identity(2, "p2-approver"),
|
|
lambda: harness.take_confirmation(confirm_id, approve=True),
|
|
) is None
|
|
approved = _as(
|
|
_identity(3, "release-approver"),
|
|
lambda: harness.take_confirmation(confirm_id, approve=True),
|
|
)
|
|
assert approved and approved["approvalStep"] == 1
|
|
assert _as(_identity(4, "p2-auditor"), harness.list_approval_history)[-1]["status"] == "APPROVED"
|
|
|
|
projected = _policy_entry("schedule.publish")["rolePolicy"]
|
|
assert projected["valid"] is True
|
|
assert projected["initiate"] == {
|
|
"roles": ["p2-initiator"], "source": "power", "scope": "P2",
|
|
}
|
|
assert projected["approve"] == {
|
|
"roles": ["release-approver"], "source": "action", "scope": "schedule.publish",
|
|
}
|
|
assert projected["history"] == {
|
|
"roles": ["p2-auditor"], "source": "power", "scope": "P2",
|
|
}
|
|
|
|
|
|
def test_missing_scoped_capabilities_use_existing_global_env_fallback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("APS_APPROVAL_INITIATOR_ROLES", "global-initiator")
|
|
monkeypatch.setenv("APS_APPROVAL_ROLES", "global-approver")
|
|
monkeypatch.setenv("APS_APPROVAL_VIEW_ROLES", "global-auditor")
|
|
_set_policy(monkeypatch, {
|
|
"actions": {
|
|
"schedule.publish": {"approve": ["release-approver"]},
|
|
},
|
|
})
|
|
|
|
confirm_id = _as(_identity(10, "global-initiator"), lambda: _stage("schedule.publish"))
|
|
approved = _as(
|
|
_identity(11, "release-approver"),
|
|
lambda: harness.take_confirmation(confirm_id, approve=True),
|
|
)
|
|
assert approved and approved["needsSecondConfirm"] is False
|
|
assert _as(_identity(12, "global-auditor"), harness.list_approval_history)
|
|
|
|
projected = _policy_entry("schedule.publish")["rolePolicy"]
|
|
assert projected["initiate"]["source"] == "global"
|
|
assert projected["initiate"]["scope"] == "APS_APPROVAL_INITIATOR_ROLES"
|
|
assert projected["history"]["roles"] == ["global-auditor"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw_policy",
|
|
[
|
|
"{not-json",
|
|
'{"unknown":{}}',
|
|
'{"actions":{"missing.action":{"approve":["admin"]}}}',
|
|
'{"powers":{"P3":{"approve":"admin"}}}',
|
|
'{"powers":{"P3":{"approve":["admin"],"approve":["auditor"]}}}',
|
|
],
|
|
)
|
|
def test_malformed_policy_fails_closed_and_projects_invalid_state(
|
|
raw_policy: str,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("APS_APPROVAL_ROLE_POLICIES", raw_policy)
|
|
|
|
with pytest.raises(PermissionError, match="角色权限"):
|
|
_as(_identity(20, "system"), lambda: _stage("mes.dispatch"))
|
|
assert harness._approval_store.pending == {}
|
|
|
|
projected = _policy_entry("mes.dispatch")["rolePolicy"]
|
|
assert projected["valid"] is False
|
|
assert projected["systemBypass"] is False
|
|
assert projected["approve"]["roles"] == []
|
|
assert projected["approve"]["source"] == "invalid"
|
|
|
|
|
|
def test_policy_becoming_invalid_blocks_approval_of_existing_pending(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
confirm_id = _as(_identity(25, "planner"), lambda: _stage("schedule.publish"))
|
|
monkeypatch.setenv("APS_APPROVAL_ROLE_POLICIES", '{"powers":{"P2":{"approve":"admin"}}}')
|
|
|
|
assert _as(
|
|
_identity(26, "system"),
|
|
lambda: harness.take_confirmation(confirm_id, approve=True),
|
|
) is None
|
|
assert confirm_id in harness._approval_store.pending
|
|
assert _as(_identity(26, "system"), harness.list_pending) == []
|
|
|
|
|
|
def test_power_policy_keeps_p3_two_person_separation(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_set_policy(monkeypatch, {
|
|
"powers": {
|
|
"P3": {
|
|
"initiate": ["dispatcher"],
|
|
"approve": ["dispatch-approver"],
|
|
"history": ["dispatch-auditor"],
|
|
},
|
|
},
|
|
})
|
|
confirm_id = _as(_identity(30, "dispatcher"), lambda: _stage("mes.dispatch"))
|
|
first_approver = _identity(31, "dispatch-approver")
|
|
first = _as(first_approver, lambda: harness.take_confirmation(confirm_id, approve=True))
|
|
denied = _as(first_approver, lambda: harness.take_confirmation(confirm_id, approve=True))
|
|
second = _as(
|
|
_identity(32, "dispatch-approver"),
|
|
lambda: harness.take_confirmation(confirm_id, approve=True),
|
|
)
|
|
|
|
assert first and first["needsSecondConfirm"] is True
|
|
assert denied and denied["approvalDenied"] is True and denied["separationRequired"] is True
|
|
assert second and second["approvalStep"] == 2 and second["executionGrant"]
|
|
history = _as(_identity(33, "dispatch-auditor"), harness.list_approval_history)
|
|
assert [item["status"] for item in history][-3:] == [
|
|
"PARTIALLY_APPROVED", "SOD_DENIED", "APPROVED",
|
|
]
|
|
|
|
|
|
def test_policy_api_keeps_existing_shape_and_adds_role_projection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
import server.gateway.app as gateway_module
|
|
from tests.auth_provider import install_test_auth
|
|
|
|
_set_policy(monkeypatch, {
|
|
"actions": {
|
|
"schedule.publish": {
|
|
"approve": ["planner"],
|
|
},
|
|
},
|
|
})
|
|
install_test_auth(monkeypatch, "tenant-role-policy-api")
|
|
client = TestClient(gateway_module.create_app())
|
|
assert client.post("/api/auth/login", json={"username": "planner"}).status_code == 200
|
|
|
|
response = client.get("/api/gov/policy")
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["defaultPower"] == "P3"
|
|
entry = next(item for item in payload["policy"] if item["action"] == "schedule.publish")
|
|
assert {"action", "power", "desc", "confirm"} <= set(entry)
|
|
assert entry["rolePolicy"]["approve"] == {
|
|
"roles": ["planner"], "source": "action", "scope": "schedule.publish",
|
|
}
|