369 lines
15 KiB
Python
369 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import inspect
|
|
from dataclasses import dataclass
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.services import ai_control as ai_control_module
|
|
from src.services import ai_router as ai_router_module
|
|
from src.services import openclaw as openclaw_module
|
|
from src.services.ai_router import AIProviderEnum
|
|
from src.services.intent_classifier import IntentType
|
|
from src.services.openclaw import OpenClawService
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _explicitly_enable_gemini_environment(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setenv("ENABLE_GEMINI", "true")
|
|
|
|
|
|
@dataclass
|
|
class _FakeEndpoint:
|
|
provider_name: str
|
|
url: str = "http://example.test"
|
|
|
|
|
|
class _FakeRoute:
|
|
def all_endpoints_in_order(self) -> list[_FakeEndpoint]:
|
|
return [
|
|
_FakeEndpoint("ollama_gcp_a"),
|
|
_FakeEndpoint("ollama_gcp_b"),
|
|
_FakeEndpoint("ollama_local"),
|
|
_FakeEndpoint("gemini", ""),
|
|
]
|
|
|
|
|
|
class _FakeFailoverManager:
|
|
def __init__(self) -> None:
|
|
self.task_types: list[str] = []
|
|
|
|
async def select_provider(self, task_type: str = "general") -> _FakeRoute:
|
|
self.task_types.append(task_type)
|
|
return _FakeRoute()
|
|
|
|
|
|
class _UnorderedFailoverManager:
|
|
async def select_provider(self, task_type: str = "general") -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
all_endpoints_in_order=lambda: [
|
|
_FakeEndpoint("ollama_local"),
|
|
_FakeEndpoint("gemini"),
|
|
_FakeEndpoint("ollama_gcp_b"),
|
|
_FakeEndpoint("ollama_gcp_a"),
|
|
],
|
|
)
|
|
|
|
|
|
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"),
|
|
(AIProviderEnum.OLLAMA, "qwen2.5:7b-instruct"),
|
|
],
|
|
intent=IntentType.DIAGNOSE,
|
|
routing_reason="high complexity would normally prefer cloud",
|
|
)
|
|
|
|
|
|
class _FakeExecutor:
|
|
def __init__(self) -> None:
|
|
self.provider_order: list[str] | None = None
|
|
self.context: dict[str, Any] | None = None
|
|
|
|
async def execute(
|
|
self,
|
|
*,
|
|
prompt: str,
|
|
provider_order: list[str],
|
|
context: dict[str, Any],
|
|
cache_ttl: int,
|
|
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],
|
|
success=True,
|
|
tokens=42,
|
|
cost_usd=0.0,
|
|
latency_ms=10.0,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_context_uses_exact_ollama_lane_then_gemini(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> 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)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "GEMINI_API_KEY", "configured-test-value")
|
|
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(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)
|
|
|
|
service = object.__new__(OpenClawService)
|
|
result = await service._call_with_fallback(
|
|
"diagnose alert",
|
|
alert_context={
|
|
"incident_id": "INC-1",
|
|
"alertname": "HostHighCpuLoad",
|
|
"target_resource": "node-exporter-110",
|
|
},
|
|
)
|
|
|
|
assert result == (
|
|
'{"root_cause":"ok","suggested_action":"NO_ACTION"}',
|
|
"ollama_gcp_a",
|
|
True,
|
|
42,
|
|
0.0,
|
|
)
|
|
assert fake_executor.provider_order == [
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
]
|
|
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
|
|
assert all(
|
|
fake_executor.context[key]
|
|
for key in ("trace_id", "run_id", "work_item_id")
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_context_can_disable_cloud_backup_for_cost_stop(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_executor = _FakeExecutor()
|
|
fake_failover = _FakeFailoverManager()
|
|
|
|
monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
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(ai_router_module, "get_ai_executor", lambda: fake_executor)
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
|
|
|
|
service = object.__new__(OpenClawService)
|
|
await service._call_with_fallback(
|
|
"diagnose alert",
|
|
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
|
|
)
|
|
|
|
assert fake_executor.provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_alert_context_uses_global_production_order(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_executor = _FakeExecutor()
|
|
fake_failover = _FakeFailoverManager()
|
|
|
|
monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "GEMINI_API_KEY", "configured-test-value")
|
|
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(ai_router_module, "get_ai_executor", lambda: fake_executor)
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
|
|
|
|
service = object.__new__(OpenClawService)
|
|
await service._call_with_fallback("general question", alert_context={"intent_hint": "query"})
|
|
|
|
assert fake_executor.provider_order == [
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_explicit_ai_governance_context_uses_ollama_first(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_executor = _FakeExecutor()
|
|
fake_failover = _FakeFailoverManager()
|
|
|
|
monkeypatch.setattr(openclaw_module.settings, "USE_AI_ROUTER", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "MOCK_MODE", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "GEMINI_API_KEY", "configured-test-value")
|
|
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(ai_router_module, "get_ai_executor", lambda: fake_executor)
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
|
|
|
|
service = object.__new__(OpenClawService)
|
|
await service._call_with_fallback(
|
|
"analyze config drift",
|
|
alert_context={
|
|
"intent_hint": "config",
|
|
"task_type": "diagnose",
|
|
"enforce_ollama_first": True,
|
|
"allow_gcp_heavy_model": True,
|
|
"target_resource": "config-drift",
|
|
},
|
|
)
|
|
|
|
assert fake_executor.provider_order == [
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
]
|
|
assert fake_failover.task_types == ["diagnose"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_context_uses_gcp_a_gcp_b_then_111_order(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_failover = _FakeFailoverManager()
|
|
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: fake_failover)
|
|
|
|
service = object.__new__(OpenClawService)
|
|
provider_order = await service._resolve_alert_provider_order(
|
|
task_type="alert_fast",
|
|
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
|
|
)
|
|
|
|
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
|
|
assert fake_failover.task_types == ["alert_fast"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_context_sorts_ollama_lane_and_drops_cloud_providers(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", False)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: _UnorderedFailoverManager())
|
|
|
|
service = object.__new__(OpenClawService)
|
|
provider_order = await service._resolve_alert_provider_order(
|
|
task_type="diagnose",
|
|
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
|
|
)
|
|
|
|
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_alert_context_drops_cloud_insertions_before_gemini_backup(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "GEMINI_API_KEY", "configured-test-value")
|
|
monkeypatch.setattr(ai_control_module, "is_provider_disabled", AsyncMock(return_value=False))
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: _UnorderedFailoverManager())
|
|
|
|
service = object.__new__(OpenClawService)
|
|
provider_order = await service._resolve_alert_provider_order(
|
|
task_type="diagnose",
|
|
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
|
|
cloud_provider_order=["claude", "openclaw_nemo", "gemini", "ollama"],
|
|
)
|
|
|
|
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_gemini_is_not_readded_to_alert_route(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ALLOW_CLOUD_FALLBACK", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_ENFORCE_OLLAMA_FIRST", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "GEMINI_API_KEY", "configured-test-value")
|
|
monkeypatch.setattr(ai_control_module, "is_provider_disabled", AsyncMock(return_value=True))
|
|
monkeypatch.setattr(openclaw_module, "get_ollama_failover_manager", lambda: _UnorderedFailoverManager())
|
|
|
|
service = object.__new__(OpenClawService)
|
|
provider_order = await service._resolve_alert_provider_order(
|
|
task_type="diagnose",
|
|
alert_context={"incident_id": "INC-1", "alertname": "HostHighCpuLoad"},
|
|
cloud_provider_order=["openclaw_nemo", "gemini"],
|
|
)
|
|
|
|
assert provider_order == ["ollama_gcp_a", "ollama_gcp_b", "ollama_local"]
|
|
|
|
|
|
def test_incident_proposal_has_no_direct_openclaw_nemo_preemption() -> None:
|
|
source = inspect.getsource(OpenClawService.generate_incident_proposal)
|
|
|
|
assert "_call_openclaw_analyze(" not in source
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_strict_incident_chain_skips_legacy_nemotron_collaboration(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
service = object.__new__(OpenClawService)
|
|
proposal = {"risk_level": "high", "reasoning": "test"}
|
|
generate = AsyncMock(return_value=(proposal, "ollama_gcp_a", True))
|
|
nemotron = AsyncMock()
|
|
monkeypatch.setattr(service, "generate_incident_proposal", generate)
|
|
monkeypatch.setattr(service, "_call_nemotron_tools", nemotron)
|
|
monkeypatch.setattr(openclaw_module.settings, "ALERT_AI_STRICT_PROVIDER_CHAIN", True)
|
|
monkeypatch.setattr(openclaw_module.settings, "ENABLE_NEMOTRON_COLLABORATION", True)
|
|
|
|
result = await service.generate_incident_proposal_with_tools(
|
|
"INC-1",
|
|
"P1",
|
|
[{"alert_name": "HostHighCpuLoad"}],
|
|
["awoooi-api"],
|
|
)
|
|
|
|
assert result == (proposal, "ollama_gcp_a", True)
|
|
assert proposal["nemotron_enabled"] is False
|
|
assert proposal["nemotron_validation"] == "strict_provider_chain"
|
|
nemotron.assert_not_awaited()
|