163 lines
5.0 KiB
Python
163 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from src.services import decision_manager as decision_module
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, response: str) -> None:
|
|
self._response = response
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
def json(self) -> dict[str, str]:
|
|
return {"response": self._response}
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
posted_urls: list[str] = []
|
|
posted_payloads: list[dict[str, Any]] = []
|
|
fail_urls: set[str] = set()
|
|
response: str = ""
|
|
|
|
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
|
self.args = args
|
|
self.kwargs = kwargs
|
|
|
|
async def __aenter__(self) -> _FakeAsyncClient:
|
|
return self
|
|
|
|
async def __aexit__(self, *args: Any) -> None:
|
|
return None
|
|
|
|
async def post(self, url: str, *, json: dict[str, Any]) -> _FakeResponse:
|
|
self.posted_urls.append(url)
|
|
self.posted_payloads.append(json)
|
|
if url in self.fail_urls:
|
|
raise RuntimeError("endpoint unavailable")
|
|
return _FakeResponse(self.response)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_fake_client() -> None:
|
|
_FakeAsyncClient.posted_urls = []
|
|
_FakeAsyncClient.posted_payloads = []
|
|
_FakeAsyncClient.fail_urls = set()
|
|
_FakeAsyncClient.response = ""
|
|
|
|
|
|
def _incident() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
incident_id="INC-ROUTE-001",
|
|
affected_services=["awoooi-api"],
|
|
signals=[
|
|
SimpleNamespace(
|
|
labels={"alertname": "AwoooPRouteTest", "severity": "warning"},
|
|
alert_name="AwoooPRouteTest",
|
|
annotations={},
|
|
)
|
|
],
|
|
)
|
|
|
|
|
|
def _ollama_order(_workload_type: str) -> tuple[SimpleNamespace, ...]:
|
|
return (
|
|
SimpleNamespace(
|
|
url="http://gcp-a:11435",
|
|
provider_name="ollama_gcp_a",
|
|
reason="global_primary_gcp_a",
|
|
),
|
|
SimpleNamespace(
|
|
url="http://gcp-b:11436",
|
|
provider_name="ollama_gcp_b",
|
|
reason="global_secondary_gcp_b",
|
|
),
|
|
SimpleNamespace(
|
|
url="http://local-111:11434",
|
|
provider_name="ollama_local",
|
|
reason="global_local_111",
|
|
),
|
|
)
|
|
|
|
|
|
def _patch_ollama_dependencies(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import httpx
|
|
|
|
from src.services import model_registry
|
|
|
|
monkeypatch.setattr(httpx, "AsyncClient", _FakeAsyncClient)
|
|
monkeypatch.setattr(decision_module, "resolve_ollama_order", _ollama_order)
|
|
monkeypatch.setattr(
|
|
model_registry,
|
|
"get_model",
|
|
lambda _provider, model_key: f"{model_key}-model",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nemoclaw_second_opinion_tries_gcp_b_after_gcp_a_failure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_ollama_dependencies(monkeypatch)
|
|
_FakeAsyncClient.fail_urls = {"http://gcp-a:11435/api/generate"}
|
|
_FakeAsyncClient.response = "<think>scratchpad</think>GCP-B advisory"
|
|
|
|
advisory = await decision_module._nemoclaw_second_opinion(
|
|
_incident(),
|
|
{"action": "restart", "confidence": 0.4, "reasoning": "primary reasoning"},
|
|
)
|
|
|
|
assert advisory == "GCP-B advisory"
|
|
assert _FakeAsyncClient.posted_urls == [
|
|
"http://gcp-a:11435/api/generate",
|
|
"http://gcp-b:11436/api/generate",
|
|
]
|
|
assert all(payload["think"] is False for payload in _FakeAsyncClient.posted_payloads)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_playbook_draft_tries_gcp_b_after_gcp_a_failure(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
_patch_ollama_dependencies(monkeypatch)
|
|
_FakeAsyncClient.fail_urls = {"http://gcp-a:11435/api/generate"}
|
|
_FakeAsyncClient.response = (
|
|
"## 症狀\nGCP-B 生成的 Playbook 草稿內容。\n"
|
|
"## 根因假設\n主要服務短暫不可用。\n"
|
|
"## 診斷步驟\n確認告警、查詢 Pod、檢查近期部署。\n"
|
|
"## 修復動作\n依標準流程處理。\n"
|
|
)
|
|
|
|
from src.repositories import alert_operation_log_repository
|
|
from src.services import knowledge_service
|
|
|
|
knowledge = SimpleNamespace(
|
|
semantic_search=AsyncMock(return_value=[]),
|
|
create_entry=AsyncMock(return_value=SimpleNamespace(entry_id="KB-ROUTE-001")),
|
|
)
|
|
op_repo = SimpleNamespace(append=AsyncMock())
|
|
monkeypatch.setattr(knowledge_service, "get_knowledge_service", lambda: knowledge)
|
|
monkeypatch.setattr(
|
|
alert_operation_log_repository,
|
|
"get_alert_operation_log_repository",
|
|
lambda: op_repo,
|
|
)
|
|
|
|
await decision_module._generate_playbook_draft_if_new(_incident())
|
|
|
|
assert _FakeAsyncClient.posted_urls == [
|
|
"http://gcp-a:11435/api/generate",
|
|
"http://gcp-b:11436/api/generate",
|
|
]
|
|
knowledge.create_entry.assert_awaited_once()
|
|
created_entry = knowledge.create_entry.await_args.args[0]
|
|
assert created_entry.related_incident_id == "INC-ROUTE-001"
|
|
assert "GCP-B 生成" in created_entry.content
|
|
op_repo.append.assert_awaited_once()
|