awoooi/apps/api/tests/test_alert_approval_guard.py

from __future__ import annotations

import pytest

from src.services.alert_approval_guard import guard_alert_approval_action
from src.services.resource_resolver import ResolveResult, set_resource_resolver
from src.utils.k8s_naming import ResourceType


class StubResolver:
    def __init__(self, *, success: bool, candidates: list[str] | None = None) -> None:
        self.success = success
        self.candidates = candidates or []
        self.calls: list[dict[str, str]] = []

    async def resolve(
        self,
        raw_resource: str,
        namespace: str = "awoooi-prod",
        resource_kind: str = "deployment",
    ) -> ResolveResult:
        self.calls.append(
            {
                "raw_resource": raw_resource,
                "namespace": namespace,
                "resource_kind": resource_kind,
            }
        )
        return ResolveResult(
            success=self.success,
            resource_name=raw_resource if self.success else None,
            namespace=namespace,
            resource_type=ResourceType(resource_kind)
            if resource_kind in ResourceType._value2member_map_
            else ResourceType.UNKNOWN,
            confidence=1.0 if self.success else 0.0,
            candidates=self.candidates,
            original_input=raw_resource,
        )


@pytest.mark.asyncio
async def test_blocks_llm_kubectl_default_namespace_for_prod_alert() -> None:
    result = await guard_alert_approval_action(
        action="kubectl logs deployment/sentry-self-hosted-snuba-metrics-consumer-1 -n default",
        alert_namespace="awoooi-prod",
        alertname="SentryRpsZero",
        alert_category="infrastructure",
    )

    assert result.blocked is True
    assert result.action.startswith("NO_ACTION - INVALID_TARGET")
    assert result.reason == "namespace_not_allowed:default"


@pytest.mark.asyncio
async def test_cold_start_invalid_k8s_target_becomes_fail_closed_agent99_candidate() -> None:
    result = await guard_alert_approval_action(
        action="kubectl scale deployment/cold-start-gate --replicas=2 -n default",
        alert_namespace="default",
        alertname="ColdStartGateBlocked",
        alert_category="general",
        target_resource="cold-start-gate",
    )

    assert result.blocked is True
    assert result.reason == "namespace_not_allowed:default"
    assert result.action == (
        "DRAFT_READY - CONTROLLED_AGENT99_RECOVERY: "
        "agent99-mode Recover controlledApply=true"
    )
    assert result.metadata["blocked_action"].startswith("kubectl scale")
    assert result.metadata["controlled_target_reroute"] is True
    assert result.metadata["controlled_target_executor"] == "Agent99"

    contract = result.metadata["repair_candidate_promotion_contract"]
    assert contract["route_id"] == "agent99_recover_after_owner_review"
    assert contract["status"] == "controlled_route_candidate_pending_receipts"
    assert contract["runtime_execution_authorized"] is False
    assert contract["runtime_write_allowed"] is False
    assert contract["completion_status"] == "partial"
    assert contract["blocked_count"] > 0
    assert "dispatch_receipt" in contract["blocked_fields"]
    assert "post_apply_verifier" in contract["blocked_fields"]
    assert "km_writeback" in contract["blocked_fields"]
    assert "playbook_trust_writeback" in contract["blocked_fields"]
    assert "kubectl_for_host_control_plane_target" in contract["forbidden_operations"]
    assert "generic_ansible_keyword_fallback" in contract["forbidden_operations"]

    closure = contract["controlled_automation_closure"]
    stages = {row["stage"]: row["status"] for row in closure["stages"]}
    assert stages == {
        "sensor": "pending",
        "normalize": "passed",
        "check": "pending",
        "controlled_apply": "pending",
        "verify": "pending",
        "learn_writeback": "pending",
    }
    assert {row["asset_type"] for row in closure["automation_asset_ledger"]} == {
        "Sensor",
        "PlayBook",
        "Verifier",
        "KM",
    }


@pytest.mark.asyncio
async def test_cold_start_valid_namespace_still_cannot_become_fake_k8s_action() -> None:
    result = await guard_alert_approval_action(
        action=(
            "kubectl rollout restart deployment/cold-start-gate "
            "-n awoooi-prod"
        ),
        alert_namespace="awoooi-prod",
        alertname="ColdStartGateBlocked",
        alert_category="general",
        target_resource="cold-start-gate",
    )

    assert result.blocked is True
    assert result.reason == "kubernetes_action_not_applicable:control_plane_recovery"
    assert result.metadata["controlled_target_executor"] == "Agent99"
    assert result.action.startswith("DRAFT_READY - CONTROLLED_AGENT99_RECOVERY")


@pytest.mark.asyncio
async def test_blocks_hallucinated_mutating_k8s_resource() -> None:
    resolver = StubResolver(success=False, candidates=["awoooi-api"])
    set_resource_resolver(resolver)
    try:
        result = await guard_alert_approval_action(
            action="kubectl scale deployment flywheelexecutionratemissing --replicas=5 -n awoooi-prod",
            alert_namespace="awoooi-prod",
            alertname="FlywheelExecutionRateMissing",
            alert_category="infrastructure",
        )
    finally:
        set_resource_resolver(None)

    assert result.blocked is True
    assert result.reason == "k8s_resource_not_found:deployment/flywheelexecutionratemissing"
    assert result.metadata["candidates"] == ["awoooi-api"]
    assert resolver.calls == [
        {
            "raw_resource": "flywheelexecutionratemissing",
            "namespace": "awoooi-prod",
            "resource_kind": "deployment",
        }
    ]


@pytest.mark.asyncio
async def test_allows_sane_readonly_cluster_inventory() -> None:
    result = await guard_alert_approval_action(
        action="kubectl get pods -n awoooi-prod",
        alert_namespace="awoooi-prod",
        alertname="ColdStartCheck",
        alert_category="infrastructure",
    )

    assert result.blocked is False
    assert result.action == "kubectl get pods -n awoooi-prod"