fix(flywheel): fallback on OpenClaw degraded responses
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 9m56s
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 9m56s
This commit is contained in:
74
apps/api/tests/test_openclaw_nemo_provider.py
Normal file
74
apps/api/tests/test_openclaw_nemo_provider.py
Normal file
@@ -0,0 +1,74 @@
|
||||
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"
|
||||
Reference in New Issue
Block a user