aios/tests/test_approved_release_executor.py

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))

from approved_release_executor import (
    Binding,
    DurableState,
    ManualOwnerReviewRequired,
    ReconciliationRequired,
    deploy_step,
    execute,
    tag_step,
    verify_authorization,
)


TARGET = "c7adece1cc8beffe7c718d53c775898b45d6c7f8"
EVIDENCE = "8c2898baf07195b9861ed2ff74f03a71674ec9e9"
ARTIFACT = "ce09a46f28b453634784abceda4597fd51c3aa0838e72a5ad5fefd03d58b949e"
PLAN = "sha256:c8199d638c2bb2e4896ca9b219ca52db79713bee007df05653ab348905d6cf28"
TAG_OID = "1" * 40


def binding(tmp_path: Path) -> Binding:
    artifact = tmp_path / "release.tar.zst"
    artifact.write_bytes(b"x")
    return Binding(
        version="VTEST",
        target_oid=TARGET,
        evidence_closure_oid=EVIDENCE,
        artifact_sha256=ARTIFACT,
        approval_plan_id=PLAN,
        artifact_path=str(artifact),
        artifact_size=1,
        production_service="test.service",
        production_port=18927,
    )


def authorization(value: Binding) -> dict:
    return {
        "accepted": True,
        "binding_replaced": False,
        "version": value.version,
        "target_oid": value.target_oid,
        "evidence_closure_oid": value.evidence_closure_oid,
        "artifact_sha256": value.artifact_sha256,
        "approval_plan_id": value.approval_plan_id,
    }


def test_owner_authorization_is_exact(tmp_path):
    value = binding(tmp_path)
    assert verify_authorization(authorization(value), value)["accepted"] is True
    wrong = authorization(value)
    wrong["artifact_sha256"] = "0" * 64
    with pytest.raises(ManualOwnerReviewRequired):
        verify_authorization(wrong, value)


def test_durable_state_rejects_different_binding(tmp_path):
    first = binding(tmp_path)
    DurableState(tmp_path / "state", first)
    second = Binding(
        **{
            **first.__dict__,
            "approval_plan_id": "sha256:" + "2" * 64,
        }
    )
    with pytest.raises(ManualOwnerReviewRequired):
        DurableState(tmp_path / "state", second)


def test_tag_conflict_never_applies(monkeypatch, tmp_path):
    value = binding(tmp_path)
    calls = []

    def fake(command, request):
        calls.append(request["operation"])
        return {
            "status": "CONFLICT",
            "version": value.version,
            "target_oid": value.target_oid,
            "artifact_sha256": value.artifact_sha256,
        }

    monkeypatch.setattr("approved_release_executor.run_adapter", fake)
    with pytest.raises(ManualOwnerReviewRequired):
        tag_step(["/tag"], value)
    assert calls == ["probe"]


def test_runtime_failure_attempts_rollback_after_apply(monkeypatch, tmp_path):
    value = binding(tmp_path)
    calls = []
    probes = 0

    def fake(command, request):
        nonlocal probes
        calls.append(request["operation"])
        if request["operation"] == "probe":
            probes += 1
            if probes == 1:
                return {
                    "status": "NOT_APPLIED",
                    "version": value.version,
                    "target_oid": value.target_oid,
                    "artifact_sha256": value.artifact_sha256,
                }
            return {
                "status": "NOT_APPLIED",
                "version": value.version,
                "target_oid": value.target_oid,
                "artifact_sha256": value.artifact_sha256,
            }
        if request["operation"] == "rollback":
            return {
                "status": "ROLLED_BACK",
                "version": value.version,
                "target_oid": value.target_oid,
                "artifact_sha256": value.artifact_sha256,
            }
        return {
            "status": "APPLY_STARTED",
            "version": value.version,
            "target_oid": value.target_oid,
            "artifact_sha256": value.artifact_sha256,
        }

    monkeypatch.setattr("approved_release_executor.run_adapter", fake)
    with pytest.raises(ReconciliationRequired):
        deploy_step(["/deploy"], value)
    assert calls == ["probe", "apply", "probe", "rollback"]


def test_execute_orders_registry_before_baseline_and_baseline_last(
    monkeypatch, tmp_path
):
    value = binding(tmp_path)
    auth_path = tmp_path / "authorization.json"
    auth_path.write_text(json.dumps(authorization(value)))
    auth_path.chmod(0o600)

    monkeypatch.setattr(
        "approved_release_executor.secure_json",
        lambda path: json.loads(path.read_text(encoding="utf-8")),
    )
    monkeypatch.setattr(
        "approved_release_executor.verify_git_lineage",
        lambda repo, binding: {"verified": True},
    )
    monkeypatch.setattr(
        "approved_release_executor.verify_artifact",
        lambda binding: {"verified": True},
    )

    calls = []
    tag_applied = False
    runtime_applied = False
    registry_applied = False
    baseline_applied = False

    def fake(command, request):
        nonlocal tag_applied, runtime_applied, registry_applied, baseline_applied
        name = command[0]
        operation = request["operation"]
        calls.append((name, operation))

        common = {
            "version": value.version,
            "target_oid": value.target_oid,
            "artifact_sha256": value.artifact_sha256,
        }
        if name == "/tag":
            if operation == "apply":
                tag_applied = True
            if tag_applied:
                return {
                    **common,
                    "status": "APPLIED_VERIFIED",
                    "tag_oid": TAG_OID,
                    "peeled_target_oid": value.target_oid,
                }
            return {**common, "status": "NOT_APPLIED"}

        if name == "/deploy":
            if operation == "apply":
                runtime_applied = True
            if operation == "rollback":
                return {**common, "status": "ROLLED_BACK"}
            if runtime_applied:
                return {
                    **common,
                    "status": "ACTIVE_VERIFIED",
                    "deployment_state": "ACTIVE",
                    "production_deployed": True,
                }
            return {**common, "status": "NOT_APPLIED"}

        if name == "/registry":
            if operation == "apply":
                registry_applied = True
            return {
                **common,
                "status": "VERIFIED" if registry_applied else "NOT_APPLIED",
                "tag_oid": TAG_OID if registry_applied else None,
            }

        if name == "/baseline":
            if operation == "apply":
                assert registry_applied is True
                assert runtime_applied is True
                baseline_applied = True
            return {
                **common,
                "status": "VERIFIED" if baseline_applied else "NOT_APPLIED",
                "tag_oid": TAG_OID if baseline_applied else None,
            }
        raise AssertionError(name)

    monkeypatch.setattr("approved_release_executor.run_adapter", fake)

    result = execute(
        binding=value,
        authorization_path=auth_path,
        repository=tmp_path / "repo.git",
        state_root=tmp_path / "state",
        tag_command=["/tag"],
        deploy_command=["/deploy"],
        registry_command=["/registry"],
        baseline_command=["/baseline"],
    )

    assert result["status"] == "PRODUCTION_ACCEPTED"
    assert calls.index(("/registry", "apply")) < calls.index(("/baseline", "apply"))
    assert calls.index(("/deploy", "apply")) < calls.index(("/registry", "apply"))
    state = json.loads((tmp_path / "state" / "execution.json").read_text())
    assert state["state"] == "COMPLETED"