wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(ai): restore bounded alert automation fallback

Browse Source
This commit is contained in:
ogt
2026-07-11 10:31:33 +08:00
parent 88f8be1378
commit f64174106b
16 changed files with 449 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
apps/api/tests/test_ai_router_cache_provider_policy.py
View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import json
from typing import Any
@@ -71,6 +72,21 @@ class _CloudProvider:
return AIResult(raw_response='{"provider":"gemini"}', success=True, provider=self.name)
class _SlowLocalProvider(_FailingLocalProvider):
async def analyze(
self,
prompt: str,
context: dict[str, Any] | None = None,
) -> AIResult:
self.calls += 1
await asyncio.sleep(1)
return AIResult(
raw_response='{"provider":"late"}',
success=True,
provider=self.name,
)
@pytest.mark.asyncio
async def test_executor_skips_cached_cloud_provider_when_ollama_lane_is_required(
monkeypatch: pytest.MonkeyPatch,
@@ -178,3 +194,34 @@ async def test_alert_cloud_backup_allowed_after_ollama_local_attempt(
assert gcp_a.calls == 1
assert local.calls == 1
assert gemini.calls == 1
@pytest.mark.asyncio
async def test_alert_provider_timeout_allows_next_fallback(
monkeypatch: pytest.MonkeyPatch,
) -> None:
fake_redis = _FakeRedis(cached_provider="none")
local = _SlowLocalProvider("ollama_local")
gemini = _CloudProvider()
registry = AIProviderRegistry()
registry.register(local)
registry.register(gemini)
monkeypatch.setattr(ai_router_module._settings, "MOCK_MODE", False)
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: fake_redis)
result = await AIRouterExecutor(registry).execute(
prompt="diagnose alert",
provider_order=["ollama_local", "gemini"],
context={
"intent_hint": "alert_triage",
"alert_type": "HostHighCpuLoad",
"alert_requires_ollama_before_cloud": True,
"provider_timeout_seconds": 0.01,
},
)
assert result.success is True
assert result.provider == "gemini"
assert local.calls == 1
assert gemini.calls == 1

23
apps/api/tests/test_alertmanager_rule_bypass.py
View File

@@ -1,13 +1,17 @@
import ast
import asyncio
import inspect
from datetime import datetime
import pytest
from src.api.v1 import webhooks as webhooks_module
from src.api.v1.webhooks import (
_analyze_alertmanager_with_timeout,
_should_bypass_alertmanager_llm,
_should_use_alertmanager_rule_first,
)
from src.repositories.alert_operation_log_repository import ALERT_EVENT_TYPES
from src.services.alertmanager_llm_guard import (
ALERTMANAGER_LLM_INFLIGHT_LOCK_TTL_SECONDS,
alertmanager_llm_inflight_key,
@@ -20,6 +24,25 @@ from src.services.decision_manager import (
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",

41
apps/api/tests/test_heartbeat_dedup_p0_4.py
View File

@@ -10,7 +10,7 @@ P0 #4 heartbeat 噪音降頻測試
直接測 telegram_gateway.send_heartbeat()，mock 掉 redis + report + send_to_group。
"""
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import AsyncMock, patch
import pytest
@@ -53,10 +53,12 @@ class FakeRedis:
def _make_report(warnings: list[str] | None = None):
"""構造 fake HeartbeatReport"""
from datetime import datetime, timezone
from datetime import UTC, datetime
from src.services.heartbeat_report_service import HeartbeatReport
return HeartbeatReport(
timestamp=datetime.now(timezone.utc),
timestamp=datetime.now(UTC),
warnings=warnings or [],
)
@@ -186,6 +188,39 @@ class TestHeartbeatDedup:
gw.send_to_group.assert_not_called()
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_ollama_fallback_warning_details_do_not_repeat(
self,
gateway_with_fake_redis,
sre_group_configured,
):
"""備援正常時，單一端點的 volatile 降級細節不重複推送。"""
gw, fake_redis = gateway_with_fake_redis
from src.services.telegram_gateway import _heartbeat_warnings_hash
fake_redis.preset(
"heartbeat:warnings_hash",
_heartbeat_warnings_hash(
["Ollama GCP-A 降級: ❌ timeout；111 備援執行中"]
),
)
with patch("src.core.redis_client.get_redis", return_value=fake_redis), \
patch("src.services.heartbeat_report_service.HeartbeatReportService") as MockSvc, \
patch("src.services.heartbeat_report_service.report_to_telegram_html",
return_value="<b>warnings</b>"):
MockSvc.return_value.collect = AsyncMock(
return_value=_make_report(
["Ollama GCP-A 降級: ❌ connect timeout；111 備援執行中"]
)
)
result = await gw.send_heartbeat()
assert result is True
gw.send_to_group.assert_not_called()
gw.send_notification.assert_not_called()
@pytest.mark.asyncio
async def test_warnings_changed_pushes(
self,

25
apps/api/tests/test_heartbeat_ollama_endpoints.py
View File

@@ -21,7 +21,7 @@ class _FakeAsyncClient:
def __init__(self, *_args: Any, **_kwargs: Any) -> None:
pass
async def __aenter__(self) -> "_FakeAsyncClient":
async def __aenter__(self) -> _FakeAsyncClient:
return self
async def __aexit__(self, *_args: Any) -> None:
@@ -53,6 +53,24 @@ async def test_probe_ollama_reports_each_endpoint(monkeypatch) -> None:
assert result["endpoints"]["GCP-A"].ok is True
assert result["endpoints"]["GCP-B"].ok is True
assert result["endpoints"]["111"].ok is False
assert result["active_endpoint"] == "GCP-A"
@pytest.mark.asyncio
async def test_probe_ollama_uses_healthy_fallback_when_gcp_a_is_down(monkeypatch) -> None:
monkeypatch.setattr(heartbeat.httpx, "AsyncClient", _FakeAsyncClient)
monkeypatch.setattr(heartbeat.settings, "OLLAMA_URL", "http://unreachable:11434")
monkeypatch.setattr(heartbeat.settings, "OLLAMA_SECONDARY_URL", "http://gcp-b:11434")
monkeypatch.setattr(heartbeat.settings, "OLLAMA_FALLBACK_URL", "http://local-111:11434")
monkeypatch.setattr(heartbeat.settings, "OLLAMA_REQUIRED_MODELS", ["gemma3:4b"])
result = await heartbeat.HeartbeatReportService()._probe_ollama()
assert result["probe"].ok is True
assert result["probe"].status == "✅ 備援中 (GCP-B)"
assert result["active_endpoint"] == "GCP-B"
assert result["models"] == {"gemma3:4b": True}
assert result["endpoints"]["GCP-A"].ok is False
def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None:
@@ -67,6 +85,7 @@ def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None:
"GCP-B": heartbeat.ProbeResult(True, "✅ 正常", 1100),
"111": heartbeat.ProbeResult(False, "❌ connect failed"),
}
report.ollama_active_endpoint = "GCP-A"
report.warnings = heartbeat.HeartbeatReportService()._build_warnings(report)
text = heartbeat.report_to_telegram_html(report)
@@ -74,4 +93,6 @@ def test_report_to_telegram_html_renders_ollama_endpoint_statuses() -> None:
assert "GCP-A: ✅ 正常" in text
assert "GCP-B: ✅ 正常" in text
assert "111: ❌ connect failed" in text
assert "Ollama 111 異常" in "\n".join(report.warnings)
assert "Ollama 111 降級" in "\n".join(report.warnings)
assert "GCP-A 備援執行中" in "\n".join(report.warnings)
assert "ollama 服務異常" not in "\n".join(report.warnings)

19
apps/api/tests/test_ollama_endpoint_resolver.py
View File

@@ -47,7 +47,7 @@ def test_all_workloads_prefer_gcp_a_lane() -> None:
assert selection.reason == "global_primary_gcp_a"
def test_all_workloads_share_global_ollama_order() -> None:
def test_general_workloads_share_global_ollama_order() -> None:
cfg = _settings()
for workload in ("interactive", "deep_rca", "local_required", "privacy_sensitive", "dr"):
@@ -64,6 +64,23 @@ def test_all_workloads_share_global_ollama_order() -> None:
]
def test_alert_fast_prefers_local_111_before_gcp_b() -> None:
order = resolve_ollama_order("alert_fast", config=_settings())
assert [selection.provider_name for selection in order] == [
"ollama_gcp_a",
"ollama_local",
"ollama_gcp_b",
]
def test_alert_fast_uses_local_when_gcp_a_is_not_configured() -> None:
selection = resolve_ollama_selection("alert_fast", config=_settings(primary=""))
assert selection.provider_name == "ollama_local"
assert selection.url == "http://192.168.0.110:11437"
def test_non_sensitive_workloads_fall_back_to_gcp_b_when_primary_missing() -> None:
cfg = _settings(primary="")

22
apps/api/tests/test_ollama_failover_manager.py
View File

@@ -84,7 +84,7 @@ def _offline_health(url: str = URL_GCP_A) -> HealthReport:
class TestDecideRoute:
"""_decide_route 路由邏輯純函數測試（ADR-110 三層容災：GCP-A → GCP-B → Local → Gemini）"""
"""_decide_route 路由邏輯純函數測試。"""
def _setup(self) -> OllamaFailoverManager:
return _make_manager()
@@ -237,6 +237,26 @@ class TestDecideRoute:
assert "nemotron" in provider_names
assert "claude" in provider_names
def test_alert_fast_prefers_healthy_local_before_gcp_b(self):
manager = self._setup()
result = manager._decide_route(
health_gcp_a=_make_health(HealthStatus.OFFLINE, URL_GCP_A),
health_gcp_b=_make_health(HealthStatus.HEALTHY, URL_GCP_B),
health_local=_make_health(HealthStatus.HEALTHY, URL_LOCAL),
url_gcp_a=URL_GCP_A,
url_gcp_b=URL_GCP_B,
url_local=URL_LOCAL,
task_type="alert_fast",
)
assert result.primary.provider_name == "ollama_local"
assert result.primary.url == URL_LOCAL
assert [endpoint.provider_name for endpoint in result.fallback_chain] == [
"ollama_gcp_b",
"gemini",
]
# ------------------------------------------------------------------
# routing_reason 記錄
# ------------------------------------------------------------------

36
apps/api/tests/test_openclaw_alert_cloud_fallback_gate.py
View File

@@ -53,9 +53,14 @@ class _UnorderedFailoverManager:
class _FakeRouter:
def __init__(self) -> None:
self.contexts: list[dict[str, Any]] = []
async def route(self, prompt: str, context: dict[str, Any]) -> SimpleNamespace:
self.contexts.append(context)
return SimpleNamespace(
selected_provider=AIProviderEnum.GEMINI,
selected_model="gemini-cloud-only-model",
fallback_chain=[
(AIProviderEnum.OPENCLAW_NEMO, "nvidia"),
(AIProviderEnum.CLAUDE, "claude"),
@@ -69,6 +74,7 @@ class _FakeRouter:
class _FakeExecutor:
def __init__(self) -> None:
self.provider_order: list[str] | None = None
self.context: dict[str, Any] | None = None
async def execute(
self,
@@ -80,6 +86,7 @@ class _FakeExecutor:
require_local: bool,
) -> SimpleNamespace:
self.provider_order = provider_order
self.context = context
return SimpleNamespace(
raw_response='{"root_cause":"ok","suggested_action":"NO_ACTION"}',
provider=provider_order[0],
@@ -96,6 +103,7 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini(
) -> None:
fake_executor = _FakeExecutor()
fake_failover = _FakeFailoverManager()
fake_router = _FakeRouter()
monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True)
monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
@@ -104,7 +112,8 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini(
monkeypatch.setattr(ai_control_module, "get_ai_router_enabled", AsyncMock(return_value=None))
monkeypatch.setattr(ai_control_module, "get_primary_provider", AsyncMock(return_value=None))
monkeypatch.setattr(ai_control_module, "is_provider_disabled", AsyncMock(return_value=False))
monkeypatch.setattr(ai_router_module, "get_ai_router", lambda: _FakeRouter())
monkeypatch.setattr(openclaw_module.settings, "OLLAMA_HEALTH_CHECK_MODEL", "gemma3:4b")
monkeypatch.setattr(ai_router_module, "get_ai_router", lambda: fake_router)
monkeypatch.setattr(ai_router_module, "get_ai_executor", lambda: fake_executor)
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
@@ -127,12 +136,18 @@ async def test_alert_context_uses_ollama_lane_then_openclaw_nemo_before_gemini(
)
assert fake_executor.provider_order == [
"ollama_gcp_a",
"ollama_gcp_b",
"ollama_local",
"ollama_gcp_b",
"openclaw_nemo",
"gemini",
]
assert fake_failover.task_types == ["diagnose"]
assert fake_failover.task_types == ["alert_fast"]
assert fake_router.contexts[0]["intent_hint"] == "alert_triage"
assert fake_executor.context is not None
assert fake_executor.context["task_type"] == "alert_fast"
assert fake_executor.context["ollama_model"] == "gemma3:4b"
assert fake_executor.context["allow_gcp_heavy_model"] is False
assert fake_executor.context["provider_timeout_seconds"] == 18.0
@pytest.mark.asyncio
@@ -159,7 +174,7 @@ async def test_alert_context_can_disable_cloud_backup_for_cost_stop(
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
)
assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"]
@pytest.mark.asyncio
@@ -225,7 +240,7 @@ async def test_explicit_ai_governance_context_uses_ollama_first(
@pytest.mark.asyncio
async def test_alert_context_uses_gcp_a_gcp_b_then_111_order(
async def test_alert_context_uses_gcp_a_111_then_gcp_b_order(
monkeypatch: pytest.MonkeyPatch,
) -> None:
fake_failover = _FakeFailoverManager()
@@ -236,11 +251,12 @@ async def test_alert_context_uses_gcp_a_gcp_b_then_111_order(
service = object.__new__(OpenClawService)
provider_order = await service._resolve_alert_provider_order(
task_type="diagnose",
task_type="alert_fast",
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
)
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"]
assert fake_failover.task_types == ["alert_fast"]
@pytest.mark.asyncio
@@ -257,7 +273,7 @@ async def test_alert_context_sorts_ollama_lane_and_drops_cloud_providers(
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
)
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b"]
@pytest.mark.asyncio
@@ -276,7 +292,7 @@ async def test_alert_context_sorts_ollama_lane_before_openclaw_nemo_backup(
cloud_provider_order=["claude", "openclaw_nemo", "gemini", "ollama"],
)
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "openclaw_nemo", "gemini"]
assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b", "openclaw_nemo", "gemini"]
@pytest.mark.asyncio
@@ -295,4 +311,4 @@ async def test_alert_context_respects_disabled_openclaw_nemo_backup(
cloud_provider_order=["openclaw_nemo", "gemini"],
)
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"]
assert provider_order == ["ollama_gcp_a", "ollama_local", "ollama_gcp_b", "gemini"]

60
apps/api/tests/test_signoz_fail_closed.py Normal file
View File

@@ -0,0 +1,60 @@
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from src.services import signoz_client as signoz_module
from src.services.openclaw import OpenClawService
from src.services.signoz_client import SignOzClient
@pytest.mark.asyncio
async def test_clickhouse_query_can_fail_closed_on_transport_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
client = SignOzClient()
http_client = SimpleNamespace(
post=AsyncMock(side_effect=RuntimeError("connection refused"))
)
monkeypatch.setattr(
signoz_module,
"get_clickhouse_client",
AsyncMock(return_value=http_client),
)
assert await client._query_clickhouse("SELECT 1") == []
with pytest.raises(RuntimeError, match="connection refused"):
await client._query_clickhouse("SELECT 1", fail_closed=True)
@pytest.mark.asyncio
async def test_gold_metrics_requires_successful_clickhouse_queries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
client = SignOzClient()
query = AsyncMock(side_effect=RuntimeError("clickhouse unavailable"))
monkeypatch.setattr(client, "_query_clickhouse", query)
with pytest.raises(RuntimeError, match="clickhouse unavailable"):
await client.get_gold_metrics("node-exporter-110")
assert query.await_args.kwargs["fail_closed"] is True
@pytest.mark.asyncio
async def test_openclaw_marks_signoz_unavailable_when_metrics_query_fails(
monkeypatch: pytest.MonkeyPatch,
) -> None:
service = object.__new__(OpenClawService)
service._signoz = SimpleNamespace(
get_gold_metrics=AsyncMock(side_effect=RuntimeError("clickhouse unavailable")),
generate_trace_url=lambda **_kwargs: "https://unused.example/traces",
)
monkeypatch.setattr(signoz_module.settings, "SIGNOZ_URL", "https://signoz.example")
metrics, trace_url = await service.get_signoz_context("node-exporter-110")
assert metrics is None
assert trace_url == "https://signoz.example/traces?service=node-exporter-110"