67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
|
|
from app.task_plane.governance_admission import (
|
|
AdmissionOperation, AdmissionPreflight, GovernanceFacts
|
|
)
|
|
from app.task_plane.governance_resolver import (
|
|
ResolutionResult, ResolutionState, ResolvedTask
|
|
)
|
|
|
|
def resolved():
|
|
return ResolutionResult(
|
|
ResolutionState.RESOLVED,
|
|
ResolvedTask("t","m","r","p","owner",{},("artifact_sha256",))
|
|
)
|
|
|
|
def facts(**kw):
|
|
body=dict(
|
|
governed_writes_enabled=True,
|
|
task_principal_writes_enabled=True,
|
|
recovery_execution_enabled=True,
|
|
recovery_shadow_mode=False,
|
|
authenticated_principal_id="principal",
|
|
target_branch="agent/x",
|
|
branch_protection_known=True,
|
|
expected_branch_oid="base",
|
|
generation=1,
|
|
task_lease_id="tl",
|
|
task_principal_id="tp",
|
|
writer_lease_id="wl",
|
|
accepted_file_allowlist=("a.py",),
|
|
acceptance_revision="rev1",
|
|
)
|
|
body.update(kw)
|
|
return GovernanceFacts(**body)
|
|
|
|
def test_ready_recovery_requires_only_governed_path_not_global_write():
|
|
result=AdmissionPreflight().evaluate(
|
|
resolution=resolved(),facts=facts(),
|
|
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
|
|
assert result.ready is True
|
|
|
|
def test_disabled_governed_write_blocks():
|
|
result=AdmissionPreflight().evaluate(
|
|
resolution=resolved(),facts=facts(governed_writes_enabled=False),
|
|
operation=AdmissionOperation.GOVERNED_COMMIT)
|
|
assert result.ready is False
|
|
assert result.blockers[0].code=="GOVERNED_WRITES_DISABLED"
|
|
|
|
def test_shadow_recovery_blocks_recovery_but_not_commit():
|
|
shadow=facts(recovery_execution_enabled=False,recovery_shadow_mode=True)
|
|
recovery=AdmissionPreflight().evaluate(
|
|
resolution=resolved(),facts=shadow,
|
|
operation=AdmissionOperation.RECOVER_EXISTING_CHILD)
|
|
assert {"RECOVERY_EXECUTION_DISABLED","RECOVERY_SHADOW_MODE"}.issubset(
|
|
{b.code for b in recovery.blockers})
|
|
commit=AdmissionPreflight().evaluate(
|
|
resolution=resolved(),facts=shadow,
|
|
operation=AdmissionOperation.GOVERNED_COMMIT)
|
|
assert commit.ready is True
|
|
|
|
def test_missing_cas_and_leases_are_explicit():
|
|
result=AdmissionPreflight().evaluate(
|
|
resolution=resolved(),
|
|
facts=facts(expected_branch_oid=None,task_lease_id=None,writer_lease_id=None),
|
|
operation=AdmissionOperation.GOVERNED_COMMIT)
|
|
codes={b.code for b in result.blockers}
|
|
assert {"EXPECTED_BRANCH_OID_UNRESOLVED","TASK_LEASE_MISSING","WRITER_LEASE_MISSING"}.issubset(codes)
|