Gitea-Tools/tests/test_recovery_playbook.py

"""Unit tests for the scoped recovery playbook (#669)."""

from __future__ import annotations

import recovery_playbook as rp
import restart_coordinator as rc


def test_ladder_covers_eleven_ordered_rungs():
    ranks = [r.rank for r in rp.RECOVERY_LADDER]
    assert ranks == list(range(len(rp.RECOVERY_LADDER)))
    assert len(rp.RECOVERY_LADDER) == 11
    assert rp.RECOVERY_LADDER[0].action is rp.RecoveryAction.CLIENT_RECONNECT
    assert rp.RECOVERY_LADDER[-1].action is rp.RecoveryAction.HOST_RESTART


def test_ladder_document_links_parent_issues():
    doc = rp.ladder_document()
    assert "#655" in doc["parent_issues"]
    assert "#652" in doc["parent_issues"]
    assert "#653" in doc["parent_issues"]
    assert doc["enforcement_issue"] == "#669"
    assert "full_mcp_restart" in doc["broad_restart_actions"]


def test_recommend_transport_eof_starts_at_client_reconnect():
    plan = rp.recommend_actions(symptoms=["transport_eof"])
    assert plan["recommended_actions"][0]["action"] == "client_reconnect"
    assert plan["recommended_actions"][0]["issue_links"]


def test_recommend_skips_successful_prior_attempts():
    attempts = [
        rp.build_attempt_record(
            "client_reconnect", outcome="success", reason="reconnected"
        )
    ]
    plan = rp.recommend_actions(
        symptoms=["transport_eof"], prior_recovery_attempts=attempts
    )
    actions = [a["action"] for a in plan["recommended_actions"]]
    assert "client_reconnect" not in actions
    assert actions[0] == "capability_refresh"


def test_escalation_denied_without_attempt_log():
    result = rp.assess_escalation("full_mcp_restart", prior_recovery_attempts=[])
    assert result.allowed is False
    assert result.require_attempt_log is True
    assert any("#669" in r for r in result.reasons)
    assert result.recommended_next  # soft recommendations still provided


def test_escalation_allowed_after_insufficient_narrower():
    attempts = [
        rp.build_attempt_record(
            "client_reconnect",
            outcome="insufficient",
            reason="still flapping",
        ),
        rp.build_attempt_record(
            "session_reconnect",
            outcome="failed",
            reason="namespace still dead",
        ),
    ]
    result = rp.assess_escalation(
        "full_mcp_restart", prior_recovery_attempts=attempts
    )
    assert result.allowed is True
    assert len(result.qualifying_attempts) == 2


def test_escalation_break_glass_bypasses_attempt_log():
    result = rp.assess_escalation(
        "host_restart", prior_recovery_attempts=[], break_glass=True
    )
    assert result.allowed is True
    assert result.break_glass is True


def test_narrow_action_does_not_require_attempt_log():
    result = rp.assess_escalation(
        "client_reconnect", prior_recovery_attempts=[]
    )
    assert result.allowed is True
    assert result.require_attempt_log is False


def test_same_rank_attempt_does_not_qualify_for_escalation():
    attempts = [
        rp.build_attempt_record(
            "full_mcp_restart", outcome="failed", reason="already failed full"
        )
    ]
    result = rp.assess_escalation(
        "full_mcp_restart", prior_recovery_attempts=attempts
    )
    assert result.allowed is False


def test_success_outcome_does_not_qualify_for_escalation():
    attempts = [
        rp.build_attempt_record(
            "client_reconnect", outcome="success", reason="fixed"
        )
    ]
    result = rp.assess_escalation(
        "full_mcp_restart", prior_recovery_attempts=attempts
    )
    assert result.allowed is False


def test_recovery_metrics_fraction_avoided():
    attempts = [
        rp.build_attempt_record("client_reconnect", outcome="success"),
        rp.build_attempt_record("session_reconnect", outcome="success"),
        rp.build_attempt_record("full_mcp_restart", outcome="success"),
    ]
    metrics = rp.recovery_metrics(attempts)
    assert metrics["successes_total"] == 3
    assert metrics["successes_avoided_full_restart"] == 2
    assert metrics["successes_full_or_host_restart"] == 1
    assert abs(metrics["fraction_avoided_full_restart"] - (2 / 3)) < 1e-9


def test_coordinator_denies_full_restart_without_attempt_log():
    inv = {
        "inventory_complete": True,
        "sessions": [],
        "leases": [],
        "prior_recovery_attempts": [],
    }
    report = rc.evaluate_restart_impact(
        inv,
        restart_class=rc.RestartClass.FULL_MCP_RESTART,
        requester_role="controller",
        requester_permissions=rc.permissions_for_role("controller"),
        controller_approved=True,
        operator_authorized=True,
    )
    assert report.allow_restart is False
    assert report.attempt_log_satisfied is False
    assert report.verdict == rc.VERDICT_UNSAFE
    blob = " ".join(report.reasons + report.authorization_reasons)
    assert "#669" in blob or "attempt log" in blob


def test_coordinator_allows_full_restart_with_attempt_log():
    inv = {
        "inventory_complete": True,
        "sessions": [],
        "leases": [],
        "prior_recovery_attempts": [
            {
                "action": "client_reconnect",
                "outcome": "insufficient",
                "reason": "still broken",
            }
        ],
    }
    report = rc.evaluate_restart_impact(
        inv,
        restart_class=rc.RestartClass.FULL_MCP_RESTART,
        requester_role="controller",
        requester_permissions=rc.permissions_for_role("controller"),
        controller_approved=True,
        operator_authorized=True,
    )
    assert report.attempt_log_satisfied is True
    assert report.allow_restart is True
    assert report.verdict == rc.VERDICT_SAFE


def test_coordinator_break_glass_allows_without_log():
    inv = {
        "inventory_complete": True,
        "sessions": [],
        "leases": [],
        "prior_recovery_attempts": [],
    }
    report = rc.evaluate_restart_impact(
        inv,
        restart_class=rc.RestartClass.FULL_MCP_RESTART,
        requester_role="controller",
        requester_permissions=rc.permissions_for_role("controller"),
        controller_approved=True,
        operator_authorized=True,
        break_glass=True,
    )
    assert report.break_glass is True
    assert report.attempt_log_satisfied is True
    assert report.allow_restart is True


def test_coordinator_client_reconnect_unaffected():
    inv = {
        "inventory_complete": True,
        "sessions": [],
        "leases": [],
        "prior_recovery_attempts": [],
    }
    report = rc.evaluate_restart_impact(
        inv,
        restart_class=rc.RestartClass.CLIENT_RECONNECT,
        requester_role="author",
        requester_permissions=rc.permissions_for_role("author"),
    )
    assert report.attempt_log_satisfied is True
    assert report.allow_restart is True


def test_restart_class_alias_accepted():
    assert (
        rp.resolve_action("full_mcp_restart")
        is rp.RecoveryAction.FULL_MCP_RESTART
    )