370 lines
13 KiB
Python
370 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
|
|
|
|
from silent_release_controller import ( # noqa: E402
|
|
AcceptanceEvidence,
|
|
ApprovalEnvelope,
|
|
ApprovalError,
|
|
ApprovalLedger,
|
|
DeploymentController,
|
|
DuplicateTagConflict,
|
|
ReleaseAcceptanceEngine,
|
|
ReleaseCandidate,
|
|
ReleaseCandidateStore,
|
|
ReleaseState,
|
|
SilentReleaseController,
|
|
SilentReleasePolicy,
|
|
TagApprovalEnvelope,
|
|
TagController,
|
|
redact,
|
|
)
|
|
|
|
|
|
COMMIT = "1" * 40
|
|
TREE = "2" * 40
|
|
ARTIFACT = "3" * 64
|
|
MANIFEST = "4" * 64
|
|
TAG_OID = "5" * 40
|
|
TARGET = "6" * 40
|
|
|
|
|
|
def candidate() -> ReleaseCandidate:
|
|
return ReleaseCandidate(
|
|
candidate_id="rc-20260723-001",
|
|
source_commit=COMMIT,
|
|
tree_oid=TREE,
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
)
|
|
|
|
|
|
def passing_evidence() -> AcceptanceEvidence:
|
|
return AcceptanceEvidence(
|
|
local_main_oid=COMMIT,
|
|
bare_main_oid=COMMIT,
|
|
gitea_main_oid=COMMIT,
|
|
repository_identity="aios/aios",
|
|
artifact_sha256=ARTIFACT,
|
|
manifest_sha256=MANIFEST,
|
|
manifest_source_commit=COMMIT,
|
|
manifest_tree_oid=TREE,
|
|
tests={"passed": True, "summary": "81 passed"},
|
|
canary={
|
|
"passed": True,
|
|
"runtime_without_git": True,
|
|
"writes_enabled": False,
|
|
},
|
|
rollback_ready=True,
|
|
authoritative_git_dir="/data/aios-gitops/repo.git",
|
|
authoritative_git_readable=True,
|
|
)
|
|
|
|
|
|
class FakeProbe:
|
|
def __init__(self, evidence=None, error=None):
|
|
self.evidence = evidence or passing_evidence()
|
|
self.error = error
|
|
self.calls = 0
|
|
|
|
def collect(self, release_candidate):
|
|
self.calls += 1
|
|
if self.error:
|
|
raise self.error
|
|
return self.evidence
|
|
|
|
|
|
class FakeBuilder:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def build(self, source_commit):
|
|
self.calls += 1
|
|
self.source_commit = source_commit
|
|
return candidate()
|
|
|
|
|
|
class FakeDeployment:
|
|
def __init__(self, health=True, restart_error=None):
|
|
self.health = health
|
|
self.restart_error = restart_error
|
|
self.switched = 0
|
|
self.restarted = 0
|
|
self.rolled_back = 0
|
|
|
|
def atomic_switch(self, release_candidate):
|
|
self.switched += 1
|
|
return {"previous_target": "old", "new_target": release_candidate.candidate_id}
|
|
|
|
def restart(self):
|
|
self.restarted += 1
|
|
if self.restart_error:
|
|
raise self.restart_error
|
|
|
|
def health_check(self, release_candidate):
|
|
return {
|
|
"passed": self.health,
|
|
"runtime_commit": release_candidate.source_commit,
|
|
"writes_enabled": False,
|
|
"authoritative_head_oid": release_candidate.source_commit,
|
|
"repository_identity": "aios/aios",
|
|
}
|
|
|
|
def rollback(self, switch_evidence):
|
|
self.rolled_back += 1
|
|
return {"passed": True, "writes_enabled": False}
|
|
|
|
|
|
class FakeTagAdapter:
|
|
def __init__(self):
|
|
self.current = None
|
|
self.creates = 0
|
|
|
|
def lookup(self, version):
|
|
return self.current
|
|
|
|
def create(self, version, tag_oid, target_oid):
|
|
self.creates += 1
|
|
self.current = (tag_oid, target_oid)
|
|
|
|
def verify(self, version):
|
|
return self.current
|
|
|
|
|
|
class SilentReleaseControllerTests(unittest.TestCase):
|
|
def test_top_level_controller_automates_prepare_then_waits_for_owner(self):
|
|
builder = FakeBuilder()
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = ReleaseCandidateStore(Path(td) / "candidates")
|
|
controller = SilentReleaseController(
|
|
builder=builder,
|
|
acceptance=ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
),
|
|
deployment=DeploymentController(
|
|
FakeDeployment(),
|
|
ApprovalLedger(Path(td) / "approvals"),
|
|
SilentReleasePolicy.default(),
|
|
store,
|
|
allow_legacy_unsigned_approval=True,
|
|
),
|
|
store=store,
|
|
policy=SilentReleasePolicy.default(),
|
|
)
|
|
prepared = controller.prepare(COMMIT)
|
|
self.assertEqual(builder.calls, 1)
|
|
self.assertEqual(prepared.state, ReleaseState.READY_FOR_APPROVAL)
|
|
self.assertEqual(prepared.approval_state, "PENDING")
|
|
self.assertEqual(
|
|
store.load(prepared.candidate_id).state,
|
|
ReleaseState.READY_FOR_APPROVAL,
|
|
)
|
|
|
|
def test_candidate_model_round_trips_and_enforces_state_machine(self):
|
|
value = candidate()
|
|
self.assertEqual(value.state, ReleaseState.BUILDING)
|
|
value.transition(ReleaseState.VERIFYING)
|
|
restored = ReleaseCandidate.from_dict(value.to_dict())
|
|
self.assertEqual(restored.state, ReleaseState.VERIFYING)
|
|
with self.assertRaisesRegex(ValueError, "invalid release transition"):
|
|
restored.transition(ReleaseState.VERIFIED)
|
|
|
|
def test_acceptance_engine_moves_verified_candidate_to_ready(self):
|
|
value = candidate()
|
|
result = ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
self.assertTrue(result.ready)
|
|
self.assertEqual(value.state, ReleaseState.READY_FOR_APPROVAL)
|
|
self.assertTrue(value.rollback_ready)
|
|
self.assertTrue(value.tests["passed"])
|
|
self.assertTrue(value.canary["runtime_without_git"])
|
|
self.assertEqual(value.git_consistency["bare_main_oid"], COMMIT)
|
|
self.assertEqual(value.gitea_consistency["main_oid"], COMMIT)
|
|
|
|
def test_artifact_mismatch_fails_closed(self):
|
|
evidence = replace(passing_evidence(), artifact_sha256="9" * 64)
|
|
value = candidate()
|
|
result = ReleaseAcceptanceEngine(
|
|
FakeProbe(evidence), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
self.assertFalse(result.ready)
|
|
self.assertEqual(value.state, ReleaseState.FAILED)
|
|
self.assertIn("artifact_sha256", result.failed_checks)
|
|
|
|
def test_gitea_unavailable_fails_closed(self):
|
|
value = candidate()
|
|
result = ReleaseAcceptanceEngine(
|
|
FakeProbe(error=RuntimeError("GITEA_UNAVAILABLE")),
|
|
expected_repository="aios/aios",
|
|
).evaluate(value)
|
|
self.assertFalse(result.ready)
|
|
self.assertEqual(value.state, ReleaseState.FAILED)
|
|
self.assertIn("GITEA_UNAVAILABLE", value.failure_reason)
|
|
|
|
def test_git_unavailable_fails_closed(self):
|
|
value = candidate()
|
|
result = ReleaseAcceptanceEngine(
|
|
FakeProbe(error=RuntimeError("AUTHORITATIVE_GIT_UNAVAILABLE")),
|
|
expected_repository="aios/aios",
|
|
).evaluate(value)
|
|
self.assertFalse(result.ready)
|
|
self.assertEqual(value.state, ReleaseState.FAILED)
|
|
self.assertIn("AUTHORITATIVE_GIT_UNAVAILABLE", value.failure_reason)
|
|
|
|
def test_runtime_with_git_metadata_is_rejected(self):
|
|
evidence = replace(
|
|
passing_evidence(),
|
|
canary={"passed": True, "runtime_without_git": False},
|
|
)
|
|
value = candidate()
|
|
result = ReleaseAcceptanceEngine(
|
|
FakeProbe(evidence), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
self.assertFalse(result.ready)
|
|
self.assertIn("runtime_without_git", result.failed_checks)
|
|
|
|
def test_runtime_switch_requires_one_bound_owner_approval(self):
|
|
value = candidate()
|
|
ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
envelope = ApprovalEnvelope.for_candidate(
|
|
value,
|
|
approval_id="approval-001",
|
|
owner_id="owner",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
with tempfile.TemporaryDirectory() as td:
|
|
deployment = FakeDeployment()
|
|
controller = DeploymentController(
|
|
deployment,
|
|
ApprovalLedger(Path(td) / "approvals"),
|
|
SilentReleasePolicy.default(),
|
|
allow_legacy_unsigned_approval=True,
|
|
)
|
|
result = controller.deploy(value, envelope)
|
|
self.assertEqual(value.state, ReleaseState.VERIFIED)
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(deployment.switched, 1)
|
|
self.assertEqual(deployment.restarted, 1)
|
|
self.assertEqual(deployment.rolled_back, 0)
|
|
|
|
def test_failed_health_check_rolls_back(self):
|
|
value = candidate()
|
|
ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
envelope = ApprovalEnvelope.for_candidate(
|
|
value,
|
|
approval_id="approval-rollback",
|
|
owner_id="owner",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
with tempfile.TemporaryDirectory() as td:
|
|
deployment = FakeDeployment(health=False)
|
|
result = DeploymentController(
|
|
deployment,
|
|
ApprovalLedger(Path(td) / "approvals"),
|
|
SilentReleasePolicy.default(),
|
|
allow_legacy_unsigned_approval=True,
|
|
).deploy(value, envelope)
|
|
self.assertFalse(result["verified"])
|
|
self.assertEqual(value.state, ReleaseState.ROLLED_BACK)
|
|
self.assertEqual(deployment.rolled_back, 1)
|
|
|
|
def test_approval_replay_is_rejected(self):
|
|
value = candidate()
|
|
ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
).evaluate(value)
|
|
envelope = ApprovalEnvelope.for_candidate(
|
|
value,
|
|
approval_id="approval-replay",
|
|
owner_id="owner",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
with tempfile.TemporaryDirectory() as td:
|
|
ledger = ApprovalLedger(Path(td) / "approvals")
|
|
controller = DeploymentController(
|
|
FakeDeployment(),
|
|
ledger,
|
|
SilentReleasePolicy.default(),
|
|
allow_legacy_unsigned_approval=True,
|
|
)
|
|
controller.deploy(value, envelope)
|
|
second = candidate()
|
|
ReleaseAcceptanceEngine(
|
|
FakeProbe(), expected_repository="aios/aios"
|
|
).evaluate(second)
|
|
with self.assertRaisesRegex(ApprovalError, "already consumed"):
|
|
controller.deploy(second, envelope)
|
|
|
|
def test_tag_controller_is_separate_and_rejects_duplicate_conflict(self):
|
|
adapter = FakeTagAdapter()
|
|
adapter.current = (TAG_OID, TARGET)
|
|
controller = TagController(
|
|
adapter,
|
|
SilentReleasePolicy.default(),
|
|
ledger=None,
|
|
allow_legacy_unsigned_approval=True,
|
|
)
|
|
envelope = TagApprovalEnvelope(
|
|
approval_id="tag-approval-001",
|
|
owner_id="owner",
|
|
version="V9R9C9",
|
|
tag_oid="7" * 40,
|
|
target_oid=TARGET,
|
|
action="production_tag",
|
|
expires_at="2099-01-01T00:00:00Z",
|
|
)
|
|
approved = controller.approve(envelope)
|
|
with self.assertRaises(DuplicateTagConflict):
|
|
controller.execute(approved)
|
|
self.assertEqual(adapter.creates, 0)
|
|
|
|
def test_policy_separates_automatic_owner_and_forbidden_actions(self):
|
|
policy = SilentReleasePolicy.default()
|
|
self.assertTrue(policy.is_automatic("canary"))
|
|
self.assertTrue(policy.requires_owner("production_runtime_switch"))
|
|
self.assertTrue(policy.requires_owner("production_tag"))
|
|
self.assertTrue(policy.is_forbidden("force_push"))
|
|
self.assertTrue(policy.is_forbidden("tag_rewrite"))
|
|
self.assertTrue(policy.is_forbidden("direct_main_push"))
|
|
self.assertTrue(policy.is_forbidden("baseline_mutation"))
|
|
self.assertTrue(policy.is_forbidden("registry_mutation"))
|
|
|
|
def test_redaction_removes_credentials_from_errors(self):
|
|
message = (
|
|
"Authorization: Bearer super-secret-token "
|
|
"https://bot:password@example.invalid/repo.git token=another-secret"
|
|
)
|
|
safe = redact(message)
|
|
self.assertNotIn("super-secret-token", safe)
|
|
self.assertNotIn("password", safe)
|
|
self.assertNotIn("another-secret", safe)
|
|
self.assertIn("REDACTED", safe)
|
|
|
|
def test_candidate_store_persists_without_secret_fields(self):
|
|
value = candidate()
|
|
value.tests = {"passed": True, "Authorization": "Bearer hidden"}
|
|
with tempfile.TemporaryDirectory() as td:
|
|
store = ReleaseCandidateStore(Path(td))
|
|
store.save(value)
|
|
raw = store.path(value.candidate_id).read_text(encoding="utf-8")
|
|
self.assertNotIn("Bearer hidden", raw)
|
|
loaded = store.load(value.candidate_id)
|
|
self.assertEqual(loaded.candidate_id, value.candidate_id)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|