96 lines
3.2 KiB
Python
96 lines
3.2 KiB
Python
from types import SimpleNamespace
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
|
|
|
import release_campaign_authorization as a
|
|
import release_campaign_bootstrap as b
|
|
import campaign_release_acceptance as c
|
|
import release_campaign_orchestrator as o
|
|
|
|
|
|
ANCHOR = "69b45e81dcd2c3707c367003b08c55fd361faae8"
|
|
|
|
|
|
def policy(state="ACTIVE", active=True):
|
|
value = b.campaign_policy("AIOS-AUTONOMOUS-RELEASE-V1", ANCHOR)
|
|
value["activation_state"] = state
|
|
value["active"] = active
|
|
return value
|
|
|
|
|
|
def test_campaign_policy_exact():
|
|
assert a.validate_policy(policy())["normal_owner_interaction_required"] is False
|
|
|
|
|
|
def test_denied_operation_fails():
|
|
obj = a.CampaignAuthorization(Path("/tmp/x"), policy(), "0"*64)
|
|
with pytest.raises(a.CampaignError):
|
|
obj.assert_operation("force_push")
|
|
|
|
|
|
def test_release_acceptance_has_no_second_owner_interaction():
|
|
obj = a.CampaignAuthorization(Path("/tmp/x"), policy(), "0"*64)
|
|
binding = {
|
|
"version": "V2R1C8",
|
|
"target_oid": "a"*40,
|
|
"evidence_closure_oid": "b"*40,
|
|
"artifact_sha256": "c"*64,
|
|
"artifact_size": 123,
|
|
"approval_plan_id": "sha256:" + "d"*64,
|
|
}
|
|
receipt = c.evaluate(
|
|
obj,
|
|
binding,
|
|
{"object_type": "commit", "target_is_ancestor": True, "closure_oid": "b"*40},
|
|
{"sha256": "c"*64, "size": 123, "retained": True},
|
|
{"install": "PASS", "health": "PASS", "restart": "PASS", "rollback_readiness": "PASS"},
|
|
{"repository": "aios/aios", "target_in_main_lineage": True},
|
|
{"conflict": False},
|
|
)
|
|
assert receipt["accepted"] is True
|
|
assert receipt["owner_interaction_required"] is False
|
|
|
|
|
|
def test_changed_binding_rejected():
|
|
obj = a.CampaignAuthorization(Path("/tmp/x"), policy(), "0"*64)
|
|
with pytest.raises(c.AcceptanceError):
|
|
c.evaluate(
|
|
obj,
|
|
{
|
|
"version": "V2R1C8",
|
|
"target_oid": "a"*40,
|
|
"evidence_closure_oid": "b"*40,
|
|
"artifact_sha256": "c"*64,
|
|
"artifact_size": 123,
|
|
"approval_plan_id": "sha256:" + "d"*64,
|
|
},
|
|
{"object_type": "commit", "target_is_ancestor": True, "closure_oid": "b"*40},
|
|
{"sha256": "e"*64, "size": 123, "retained": True},
|
|
{"install": "PASS", "health": "PASS", "restart": "PASS", "rollback_readiness": "PASS"},
|
|
{"repository": "aios/aios", "target_in_main_lineage": True},
|
|
{"conflict": False},
|
|
)
|
|
|
|
|
|
def test_orchestrator_recoverable_failure_resumes_without_owner():
|
|
state = o.ReleaseState("V2R1C8", state="STAGING", last_stable_state="STAGING")
|
|
state.recoverable_failure("transient_502")
|
|
assert state.as_dict()["owner_interaction_required"] is False
|
|
state.resume()
|
|
assert state.state == "STAGING"
|
|
|
|
|
|
def test_only_red_line_can_enter_manual_review():
|
|
state = o.ReleaseState("V2R1C8")
|
|
with pytest.raises(o.OrchestratorError):
|
|
state.manual_review("pytest_failed")
|
|
state.manual_review("formal_same_name_tag_conflict")
|
|
assert state.as_dict()["owner_interaction_required"] is True
|