70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
|
|
from .models import ProtectionSnapshot
|
|
from .protection import MemoryProtectionStore, ProtectionPlan, snapshot_sha256
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RollbackConfiguration:
|
|
branch: str
|
|
previous_sha256: str
|
|
target_sha256: str
|
|
payload: dict
|
|
|
|
|
|
class RollbackConfigurationManager:
|
|
def build(self, plan: ProtectionPlan) -> RollbackConfiguration:
|
|
return RollbackConfiguration(
|
|
plan.branch,
|
|
plan.expected_previous_sha256,
|
|
plan.target_sha256,
|
|
{"previous": plan.previous.as_dict(), "target": plan.target.as_dict()},
|
|
)
|
|
|
|
def install_fixture(self, root: Path, config: RollbackConfiguration) -> Path:
|
|
if not root.name.startswith("aios-governance-fixture-"):
|
|
raise PermissionError("rollback configuration may only be installed in fixture root")
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
destination = root / "rollback-configuration.json"
|
|
raw = json.dumps(config.payload, sort_keys=True, indent=2).encode() + b"\n"
|
|
with tempfile.NamedTemporaryFile(dir=root, prefix=".rollback.", delete=False) as handle:
|
|
handle.write(raw)
|
|
handle.flush()
|
|
temporary = Path(handle.name)
|
|
temporary.chmod(0o600)
|
|
temporary.replace(destination)
|
|
return destination
|
|
|
|
def verify(self, config: RollbackConfiguration, plan: ProtectionPlan) -> bool:
|
|
return (
|
|
config.previous_sha256 == snapshot_sha256(plan.previous)
|
|
and config.target_sha256 == snapshot_sha256(plan.target)
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ProtectionCompensationSaga:
|
|
adapter: object
|
|
store: MemoryProtectionStore
|
|
plan: ProtectionPlan
|
|
applied: bool = False
|
|
rolled_back: bool = False
|
|
|
|
def execute_fixture(self) -> ProtectionSnapshot:
|
|
result = self.adapter.apply_fixture(self.store, self.plan)
|
|
self.applied = True
|
|
return result
|
|
|
|
def compensate_fixture(self) -> ProtectionSnapshot:
|
|
if not self.applied:
|
|
return self.store.read(self.plan.branch)
|
|
result = self.adapter.rollback_fixture(self.store, self.plan)
|
|
self.rolled_back = True
|
|
return result
|