126 lines
3.6 KiB
Python
126 lines
3.6 KiB
Python
"""Runbook generation must use the global provider and cost-guard contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.services import ai_router
|
|
from src.services.ai_provider_policy import PRODUCTION_PROVIDER_ORDER
|
|
from src.services.ai_providers.interfaces import AIResult
|
|
from src.services.runbook_generator import (
|
|
NemotronRunbookGenerator,
|
|
_generated_document,
|
|
)
|
|
|
|
|
|
def _objects():
|
|
incident = SimpleNamespace(
|
|
incident_id="INC-20260714-RUNBOOK1",
|
|
affected_services=["api"],
|
|
severity=SimpleNamespace(value="P1"),
|
|
)
|
|
playbook = SimpleNamespace(
|
|
playbook_id="PB-RUNBOOK-1",
|
|
name="bounded repair",
|
|
)
|
|
result = SimpleNamespace(
|
|
executed_steps=["check", "apply", "verify"],
|
|
execution_time_ms=1200,
|
|
error=None,
|
|
)
|
|
return incident, playbook, result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runbook_generation_uses_exact_chain_and_receipt_correlations(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
captured: dict = {}
|
|
|
|
class Executor:
|
|
async def execute(self, **kwargs):
|
|
captured.update(kwargs)
|
|
return AIResult(
|
|
raw_response=json.dumps({"content": "## 症狀描述\nverified"}),
|
|
success=True,
|
|
provider="ollama",
|
|
)
|
|
|
|
monkeypatch.setattr(ai_router, "get_ai_executor", lambda: Executor())
|
|
incident, playbook, result = _objects()
|
|
|
|
content = await NemotronRunbookGenerator()._generate_runbook_content(
|
|
incident,
|
|
playbook,
|
|
result,
|
|
)
|
|
|
|
assert content == "## 症狀描述\nverified"
|
|
assert captured["provider_order"] == list(PRODUCTION_PROVIDER_ORDER)
|
|
assert captured["provider_order"] == [
|
|
"ollama_gcp_a",
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
]
|
|
assert captured["context"] == {
|
|
"task_type": "runbook_generation",
|
|
"intent_hint": "maintenance",
|
|
"enforce_ollama_first": True,
|
|
"alert_requires_ollama_before_cloud": True,
|
|
"trace_id": "incident:INC-20260714-RUNBOOK1",
|
|
"run_id": "runbook:INC-20260714-RUNBOOK1:runbook",
|
|
"work_item_id": "playbook:PB-RUNBOOK-1",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cost_guard_block_returns_local_minimal_document_only(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
calls = 0
|
|
|
|
class Executor:
|
|
async def execute(self, **_kwargs):
|
|
nonlocal calls
|
|
calls += 1
|
|
return AIResult(
|
|
raw_response="",
|
|
success=False,
|
|
provider="gemini",
|
|
error="cost_guard_blocked:redis_unavailable",
|
|
)
|
|
|
|
monkeypatch.setattr(ai_router, "get_ai_executor", lambda: Executor())
|
|
incident, playbook, result = _objects()
|
|
|
|
content = await NemotronRunbookGenerator()._generate_runbook_content(
|
|
incident,
|
|
playbook,
|
|
result,
|
|
)
|
|
|
|
assert calls == 1
|
|
assert "## 症狀描述" in content
|
|
assert "AI route fallback" in content
|
|
|
|
|
|
def test_runbook_source_has_no_direct_legacy_or_paid_provider_call() -> None:
|
|
source = inspect.getsource(NemotronRunbookGenerator)
|
|
|
|
assert "get_nvidia_provider" not in source
|
|
assert "GeminiProvider(" not in source
|
|
assert "_call_gemini" not in source
|
|
assert "get_ai_executor" in source
|
|
|
|
|
|
def test_generated_document_rejects_unstructured_provider_prose() -> None:
|
|
assert _generated_document("unstructured answer") is None
|
|
assert _generated_document(json.dumps({"content": "## 驗證\npass"})) == (
|
|
"## 驗證\npass"
|
|
)
|