312 lines
10 KiB
Python
312 lines
10 KiB
Python
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 解決"
|
|
)
|