import ast import asyncio import inspect from datetime import datetime from types import SimpleNamespace from unittest.mock import AsyncMock, Mock import pytest from src.api.v1 import webhooks as webhooks_module from src.api.v1.webhooks import ( _analyze_alertmanager_with_timeout, _process_new_alert_background, _should_bypass_alertmanager_llm, _should_use_alertmanager_rule_first, ) from src.repositories.alert_operation_log_repository import ALERT_EVENT_TYPES from src.services.alert_approval_guard import ApprovalActionGuardResult from src.services.alertmanager_llm_guard import ( ALERTMANAGER_LLM_INFLIGHT_LOCK_TTL_SECONDS, alertmanager_llm_inflight_key, ) from src.services.decision_manager import ( _is_host_layer_ssh_category, _is_non_k8s_host_category, _should_escalate_auto_approve_rejection, ) from src.services.telegram_gateway import _format_resolved_guard_stamp def test_repair_candidate_operation_log_events_use_database_enum() -> None: tree = ast.parse(inspect.getsource(webhooks_module)) emitted = { call.args[0].value for call in ast.walk(tree) if isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) and call.func.attr == "append" and isinstance(call.func.value, ast.Name) and call.func.value.id == "_op_log_fallback" and call.args and isinstance(call.args[0], ast.Constant) and isinstance(call.args[0].value, str) } assert emitted == {"PLAYBOOK_DRAFT_CREATED", "STATE_GUARD_BLOCKED"} assert emitted <= ALERT_EVENT_TYPES def test_host_resource_yaml_no_action_bypasses_llm(): rule_response = { "rule_id": "host_resource_alert", "suggested_action": "NO_ACTION", "kubectl_command": "", } assert _should_bypass_alertmanager_llm(rule_response, "host_resource") is True def test_generic_fallback_does_not_bypass_llm(): rule_response = { "rule_id": "generic_fallback", "suggested_action": "NO_ACTION", "kubectl_command": "", } assert _should_bypass_alertmanager_llm(rule_response, "host_resource") is False def test_non_host_category_does_not_bypass_llm(): rule_response = { "rule_id": "host_resource_alert", "suggested_action": "NO_ACTION", "kubectl_command": "", } assert _should_bypass_alertmanager_llm(rule_response, "kubernetes") is False def test_backup_failure_yaml_no_action_bypasses_llm(): rule_response = { "rule_id": "host_backup_failed", "suggested_action": "NO_ACTION", "kubectl_command": "", } assert _should_bypass_alertmanager_llm(rule_response, "backup_failure") is True def test_host_resource_ssh_rule_uses_rule_first(): rule_response = { "rule_id": "host_resource_alert", "suggested_action": "SSH_DIAGNOSE", "kubectl_command": "ssh {host} 'df -h'", } assert _should_use_alertmanager_rule_first(rule_response, "host_resource") is True def test_backup_failure_ssh_rule_uses_rule_first(): rule_response = { "rule_id": "host_backup_failed", "suggested_action": "SSH_DIAGNOSE", "kubectl_command": "ssh {host} 'tail -80 backup.log'", } assert _should_use_alertmanager_rule_first(rule_response, "backup_failure") is True def test_generic_fallback_does_not_use_rule_first(): rule_response = { "rule_id": "generic_fallback", "suggested_action": "SSH_DIAGNOSE", "kubectl_command": "ssh {host} 'df -h'", } assert _should_use_alertmanager_rule_first(rule_response, "host_resource") is False def test_manual_gate_reasons_escalate_to_emergency_intervention(): assert _should_escalate_auto_approve_rejection("no_executable_action") is True assert _should_escalate_auto_approve_rejection("no_playbook") is True assert _should_escalate_auto_approve_rejection("critical_operation") is False def test_backup_failure_routes_to_decision_ssh_before_kubectl_parser(): assert _is_host_layer_ssh_category("backup_failure") is True assert _is_host_layer_ssh_category("host_resource") is True assert _is_host_layer_ssh_category("kubernetes") is False def test_backup_failure_blocks_k8s_auto_execute(): assert _is_non_k8s_host_category("backup_failure") is True assert _is_non_k8s_host_category("host_resource") is True assert _is_non_k8s_host_category("infrastructure") is False def test_alertmanager_llm_inflight_lock_key_is_fingerprint_scoped(): fingerprint = "abc123" assert alertmanager_llm_inflight_key(fingerprint) == "alertmanager:llm_inflight:abc123" assert ALERTMANAGER_LLM_INFLIGHT_LOCK_TTL_SECONDS == 600 @pytest.mark.asyncio async def test_alertmanager_analysis_timeout_returns_fallback(monkeypatch): from src.api.v1 import webhooks as webhooks_module class SlowOpenClaw: async def analyze_alert(self, alert_context): await asyncio.sleep(1) return "unexpected" monkeypatch.setattr(webhooks_module, "ALERTMANAGER_BACKGROUND_AI_TIMEOUT_SECONDS", 0.01) result = await _analyze_alertmanager_with_timeout( SlowOpenClaw(), {"alertname": "AwoooPTimeoutCanary"}, alert_id="alert-timeout", alertname="AwoooPTimeoutCanary", ) assert result == (None, "fallback_timeout", "", None, "", 0, 0.0) @pytest.mark.asyncio async def test_alertmanager_analysis_error_returns_fallback(): class BrokenOpenClaw: async def analyze_alert(self, alert_context): raise RuntimeError("provider chain failed") result = await _analyze_alertmanager_with_timeout( BrokenOpenClaw(), {"alertname": "AwoooPErrorCanary"}, alert_id="alert-error", alertname="AwoooPErrorCanary", ) assert result == (None, "fallback_error", "", None, "", 0, 0.0) @pytest.mark.asyncio async def test_agent99_durable_route_dispatches_before_llm(monkeypatch): class ApprovalService: created_request = None async def create_approval_with_fingerprint(self, *, request, fingerprint): self.created_request = request assert fingerprint == "fp-agent99-durable-route" return SimpleNamespace(id="approval-agent99-fast-path", incident_id=None) async def update_incident_id(self, approval_id, incident_id): assert str(approval_id) == "approval-agent99-fast-path" assert incident_id == "INC-AGENT99-FAST-PATH" service = ApprovalService() openclaw_factory = Mock( side_effect=AssertionError("durable Agent99 route must not create OpenClaw") ) handoff = AsyncMock( return_value={ "status": "agent99_dispatch_accepted_verifier_pending", "queued": True, } ) telegram = AsyncMock() monkeypatch.setattr(webhooks_module, "get_approval_service", lambda: service) monkeypatch.setattr(webhooks_module, "get_openclaw", openclaw_factory) monkeypatch.setattr( webhooks_module, "get_trusted_alert_canonical_route_context", lambda *_args, **_kwargs: {}, ) monkeypatch.setattr( webhooks_module, "match_rule", lambda _context: { "rule_id": "generic_fallback", "risk_level": "critical", "blast_radius": {}, "kubectl_command": "", "description": "generic fallback must not block Agent99", "confidence": 0.0, }, ) monkeypatch.setattr( webhooks_module, "resolve_playbook_id_for_alert", AsyncMock(return_value=None), ) monkeypatch.setattr( webhooks_module, "guard_alert_approval_action", AsyncMock( side_effect=lambda **kwargs: ApprovalActionGuardResult( action=kwargs["action"] ) ), ) monkeypatch.setattr( webhooks_module, "get_auto_approve_policy", lambda: SimpleNamespace( evaluate=lambda _proposal: SimpleNamespace( should_auto_approve=False, reason=SimpleNamespace(value="bounded_durable_route"), ) ), ) monkeypatch.setattr( webhooks_module, "create_incident_for_approval", AsyncMock(return_value="INC-AGENT99-FAST-PATH"), ) monkeypatch.setattr( webhooks_module, "record_alertmanager_event", AsyncMock(), ) monkeypatch.setattr(webhooks_module, "_try_auto_repair_background", handoff) monkeypatch.setattr(webhooks_module, "_push_to_telegram_background", telegram) monkeypatch.setattr(webhooks_module, "record_alert_chain_success", Mock()) await _process_new_alert_background( alert_context={ "alertname": "RebootAutoRecoveryActiveBlocker", "annotations": {"summary": "cold-start gate blocked"}, "source_url": "http://prometheus.invalid/graph", }, alert_id="alert-agent99-fast-path", fingerprint="fp-agent99-durable-route", target_resource="reboot-auto-recovery-slo", namespace="default", alert_type="custom", message="cold-start-gate service recovery verifier", alertname="RebootAutoRecoveryActiveBlocker", severity="critical", alert_labels={"service": "cold-start-gate"}, notification_type="TYPE-3", alert_category="general", can_auto_repair=True, ) openclaw_factory.assert_not_called() handoff.assert_awaited_once() assert handoff.await_args.kwargs["source_alertname"] == ( "RebootAutoRecoveryActiveBlocker" ) assert service.created_request.risk_level.value == "medium" assert service.created_request.action == ( "AGENT99_CONTROLLED_ROUTE agent99:host_recovery:Recover" ) assert service.created_request.metadata["source"] == "agent99_durable_route" telegram.assert_awaited_once() assert telegram.await_args.kwargs["automation_state"] == ( "agent99_dispatch_accepted_verifier_pending" ) def test_resolved_guard_stamp_without_timestamp_is_clean(): assert _format_resolved_guard_stamp(None) == "✅ 此事件已解決" def test_resolved_guard_stamp_with_timestamp_formats_time(): resolved_at = datetime(2026, 4, 25, 0, 2) assert ( _format_resolved_guard_stamp(resolved_at) == "✅ 此事件已於 2026-04-25 00:02 解決" )