75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src.services.ai_providers.openclaw_nemo import OpenClawNemoProvider
|
|
|
|
|
|
def _transport(payload: dict) -> httpx.MockTransport:
|
|
def handler(_: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
return httpx.MockTransport(handler)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openclaw_degraded_response_is_fallbackable():
|
|
provider = OpenClawNemoProvider()
|
|
provider._http_client = httpx.AsyncClient(transport=_transport({
|
|
"action_title": "OpenClaw 降級調查",
|
|
"description": "OpenClaw 下游 LLM timeout",
|
|
"suggested_action": "investigate",
|
|
"kubectl_command": None,
|
|
"target_resource": "awoooi-api",
|
|
"namespace": "awoooi-prod",
|
|
"risk_level": "low",
|
|
"confidence": 0.2,
|
|
"provider": "openclaw_degraded",
|
|
"degraded": True,
|
|
"reasoning": "NVIDIA NIM timeout",
|
|
}))
|
|
|
|
result = await provider.analyze("diagnose", context={"incident_id": "inc-1"})
|
|
await provider.close()
|
|
|
|
assert result.success is False
|
|
assert result.provider == "openclaw_nemo"
|
|
assert "OpenClaw degraded" in (result.error or "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openclaw_low_confidence_response_is_fallbackable():
|
|
provider = OpenClawNemoProvider()
|
|
provider._http_client = httpx.AsyncClient(transport=_transport({
|
|
"action_title": "調查服務異常",
|
|
"risk_level": "low",
|
|
"confidence": 0.1,
|
|
"provider": "openclaw_nvidia_nim",
|
|
"reasoning": "low confidence",
|
|
}))
|
|
|
|
result = await provider.analyze("diagnose", context={"incident_id": "inc-2"})
|
|
await provider.close()
|
|
|
|
assert result.success is False
|
|
assert "OpenClaw degraded" in (result.error or "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openclaw_normal_response_stays_successful():
|
|
provider = OpenClawNemoProvider()
|
|
provider._http_client = httpx.AsyncClient(transport=_transport({
|
|
"action_title": "重啟服務",
|
|
"risk_level": "medium",
|
|
"confidence": 0.8,
|
|
"provider": "openclaw_nvidia_nim",
|
|
"reasoning": "valid proposal",
|
|
}))
|
|
|
|
result = await provider.analyze("diagnose", context={"incident_id": "inc-3"})
|
|
await provider.close()
|
|
|
|
assert result.success is True
|
|
assert result.provider == "openclaw_nemo"
|