awoooi/apps/api/tests/test_current_owner_controlled_apply_policy.py

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import UTC, datetime
from types import SimpleNamespace
from uuid import uuid4

import pytest

from src.api.v1.webhooks import _controlled_ai_policy_allows
from src.core.trust_engine import classify_risk, get_required_signatures
from src.models.approval import ApprovalStatus, RiskLevel
from src.services import approval_db as approval_db_module
from src.services.approval_db import ApprovalDBService, approval_request_to_record_data
from src.services.telegram_gateway import TelegramGateway, TelegramMessage


@pytest.mark.parametrize(
    "risk_level",
    [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH],
)
def test_noncritical_risks_are_auto_authorized_for_controlled_apply(
    risk_level: RiskLevel,
) -> None:
    assert get_required_signatures(risk_level) == 0
    assert _controlled_ai_policy_allows(risk_level) is True


def test_critical_risk_remains_break_glass() -> None:
    assert get_required_signatures(RiskLevel.CRITICAL) == 1
    assert _controlled_ai_policy_allows(RiskLevel.CRITICAL) is False


@pytest.mark.parametrize(
    "action",
    [
        "host reboot 192.168.0.110",
        "kubectl get secret awoooi-secrets",
        "git force push main",
        "restore database from snapshot",
    ],
)
def test_critical_action_overrides_explicit_high_risk(action: str) -> None:
    assert classify_risk(action, explicit_level=RiskLevel.HIGH) == RiskLevel.CRITICAL


def test_medium_approval_record_is_created_auto_approved() -> None:
    request = SimpleNamespace(
        action="kubectl rollout restart deployment/api",
        description="bounded rollout",
        metadata={"source": "alertmanager"},
        blast_radius=None,
        dry_run_checks=[],
        requested_by="OpenClaw",
        expires_at=None,
        incident_id="INC-CONTROLLED",
        matched_playbook_id=None,
    )

    data = approval_request_to_record_data(
        request,
        RiskLevel.MEDIUM,
        get_required_signatures(RiskLevel.MEDIUM),
    )

    assert data["status"] == ApprovalStatus.APPROVED
    assert data["required_signatures"] == 0
    assert data["resolved_at"] is not None


def test_medium_telegram_card_is_automation_progress_not_action_required() -> None:
    message = TelegramMessage(
        status_emoji="⚠️",
        risk_level="MEDIUM",
        resource_name="awoooi-api",
        root_cause="restart required",
        suggested_action="kubectl rollout restart deployment/awoooi-api",
        estimated_downtime="30s",
        approval_id="approval-1",
        incident_id="INC-CONTROLLED",
        controlled_auto_authorized=True,
    )

    rendered = message.format()

    assert "AI CONTROLLED AUTOMATION" in rendered
    assert "ACTION REQUIRED" not in rendered
    assert "awaiting_approval" not in rendered
    assert "check>apply>verify>writeback" in rendered


def test_critical_telegram_card_stays_break_glass() -> None:
    message = TelegramMessage(
        status_emoji="🚨",
        risk_level="CRITICAL",
        resource_name="database",
        root_cause="destructive restore requested",
        suggested_action="restore database from snapshot",
        estimated_downtime="unknown",
        approval_id="approval-critical",
        incident_id="INC-CRITICAL",
    )

    rendered = message.format()

    assert "BREAK-GLASS REQUIRED" in rendered
    assert "AI CONTROLLED AUTOMATION" not in rendered


@pytest.mark.asyncio
async def test_controlled_automation_keyboard_has_no_approval_buttons() -> None:
    gateway = TelegramGateway()

    keyboard = await gateway._build_inline_keyboard(
        approval_id="approval-1",
        incident_id="INC-CONTROLLED",
        suggested_action="kubectl rollout restart deployment/awoooi-api",
        controlled_auto_authorized=True,
    )
    button_texts = {
        button["text"]
        for row in keyboard["inline_keyboard"]
        for button in row
    }

    assert "✅ 批准" not in button_texts
    assert "❌ 拒絕" not in button_texts
    assert "🧰 處置包" in button_texts
    assert "📊 歷史" in button_texts


def _approval_record(*, action: str, risk_level: RiskLevel) -> SimpleNamespace:
    now = datetime.now(UTC)
    return SimpleNamespace(
        id=str(uuid4()),
        action=action,
        description="legacy pending alert",
        status=ApprovalStatus.PENDING,
        risk_level=risk_level,
        required_signatures=1,
        current_signatures=0,
        signatures=[],
        blast_radius={
            "affected_pods": 1,
            "estimated_downtime": "30s",
            "related_services": ["awoooi-api"],
            "data_impact": "write",
        },
        dry_run_checks=[],
        requested_by="OpenClaw",
        created_at=now,
        expires_at=None,
        resolved_at=None,
        rejection_reason=None,
        extra_metadata={"source": "alertmanager"},
        fingerprint="fingerprint",
        hit_count=1,
        last_seen_at=now,
        incident_id="INC-LEGACY-PENDING",
        matched_playbook_id=None,
        telegram_message_id=None,
        telegram_chat_id=None,
    )


class _FakeResult:
    def __init__(self, record: SimpleNamespace) -> None:
        self._record = record

    def scalar_one_or_none(self) -> SimpleNamespace:
        return self._record


class _FakeDB:
    def __init__(self, record: SimpleNamespace) -> None:
        self.record = record
        self.events: list[object] = []

    async def execute(self, _statement: object) -> _FakeResult:
        return _FakeResult(self.record)

    def add(self, event: object) -> None:
        self.events.append(event)

    async def flush(self) -> None:
        return None

    async def refresh(self, _record: object) -> None:
        return None


@pytest.mark.asyncio
async def test_legacy_medium_pending_gate_is_durably_promoted(monkeypatch) -> None:
    record = _approval_record(
        action="kubectl rollout restart deployment/awoooi-api",
        risk_level=RiskLevel.MEDIUM,
    )
    db = _FakeDB(record)

    @asynccontextmanager
    async def fake_db_context():
        yield db

    monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context)

    approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id)

    assert promoted is True
    assert approval is not None
    assert approval.status == ApprovalStatus.APPROVED
    assert approval.required_signatures == 0
    assert approval.metadata["legacy_pending_promoted"] is True
    assert len(db.events) == 1


@pytest.mark.asyncio
async def test_legacy_critical_pending_gate_is_not_promoted(monkeypatch) -> None:
    record = _approval_record(
        action="host reboot 192.168.0.110",
        risk_level=RiskLevel.HIGH,
    )
    db = _FakeDB(record)

    @asynccontextmanager
    async def fake_db_context():
        yield db

    monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context)

    approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id)

    assert promoted is False
    assert approval is not None
    assert approval.status == ApprovalStatus.PENDING
    assert approval.required_signatures == 1
    assert db.events == []