54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
"""Release state machine with explicit fail-closed production gates."""
|
|
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
from typing import Mapping
|
|
|
|
|
|
class InvalidTransition(RuntimeError):
|
|
pass
|
|
|
|
|
|
class ReleaseState(str, Enum):
|
|
SOURCE_READY = "SOURCE_READY"
|
|
TESTED = "TESTED"
|
|
PR_OPEN = "PR_OPEN"
|
|
PR_APPROVED = "PR_APPROVED"
|
|
MERGED = "MERGED"
|
|
ARTIFACT_VERIFIED = "ARTIFACT_VERIFIED"
|
|
RELEASE_APPROVED = "RELEASE_APPROVED"
|
|
TAG_VERIFIED = "TAG_VERIFIED"
|
|
DEPLOYED = "DEPLOYED"
|
|
PRODUCTION_ACCEPTED = "PRODUCTION_ACCEPTED"
|
|
|
|
|
|
TRANSITIONS = {
|
|
ReleaseState.SOURCE_READY: (ReleaseState.TESTED,),
|
|
ReleaseState.TESTED: (ReleaseState.PR_OPEN,),
|
|
ReleaseState.PR_OPEN: (ReleaseState.PR_APPROVED,),
|
|
ReleaseState.PR_APPROVED: (ReleaseState.MERGED,),
|
|
ReleaseState.MERGED: (ReleaseState.ARTIFACT_VERIFIED,),
|
|
ReleaseState.ARTIFACT_VERIFIED: (ReleaseState.RELEASE_APPROVED,),
|
|
ReleaseState.RELEASE_APPROVED: (ReleaseState.TAG_VERIFIED,),
|
|
ReleaseState.TAG_VERIFIED: (ReleaseState.DEPLOYED,),
|
|
ReleaseState.DEPLOYED: (ReleaseState.PRODUCTION_ACCEPTED,),
|
|
}
|
|
|
|
|
|
class ReleaseStateMachine:
|
|
def __init__(self, state: ReleaseState = ReleaseState.SOURCE_READY) -> None:
|
|
self.state = state
|
|
|
|
def transition(self, target: ReleaseState, evidence: Mapping[str, object] | None = None) -> ReleaseState:
|
|
if target not in TRANSITIONS.get(self.state, ()):
|
|
raise InvalidTransition(f"{self.state.value}_to_{target.value}_denied")
|
|
evidence = evidence or {}
|
|
if target is ReleaseState.PRODUCTION_ACCEPTED:
|
|
if evidence.get("tag_verified") is not True or evidence.get("remote_tag_readback") is not True or evidence.get("service_health") is not True:
|
|
raise InvalidTransition("production_acceptance_evidence_incomplete")
|
|
if target is ReleaseState.DEPLOYED and evidence.get("artifact_verified") is not True:
|
|
raise InvalidTransition("deployment_artifact_not_verified")
|
|
self.state = target
|
|
return self.state
|
|
|