92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
"""AIRouter integration with Ollama health evidence.
|
|
|
|
The failover manager may report the currently healthiest candidate, but that
|
|
observation cannot mutate the production execution contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.services.ai_router import AIProviderEnum, AIRouter, reset_ai_router
|
|
|
|
EXPECTED_FALLBACK = [
|
|
AIProviderEnum.OLLAMA_GCP_B,
|
|
AIProviderEnum.OLLAMA_LOCAL,
|
|
AIProviderEnum.GEMINI,
|
|
]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_router() -> None:
|
|
yield
|
|
reset_ai_router()
|
|
|
|
|
|
def _router_with_observed_fallback(provider_name: str) -> tuple[AIRouter, MagicMock]:
|
|
observed_fallback = (
|
|
None
|
|
if provider_name == "ollama_gcp_a"
|
|
else SimpleNamespace(provider_name=provider_name)
|
|
)
|
|
manager = MagicMock()
|
|
manager.select_provider = AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
primary=SimpleNamespace(provider_name="ollama_gcp_a"),
|
|
observed_fallback=observed_fallback,
|
|
fallback_chain=[],
|
|
)
|
|
)
|
|
router = AIRouter()
|
|
router._failover_manager = manager
|
|
return router, manager
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"observed_primary",
|
|
["ollama_gcp_a", "ollama_gcp_b", "ollama_local", "gemini"],
|
|
)
|
|
async def test_health_observation_never_reorders_execution_contract(
|
|
observed_primary: str,
|
|
) -> None:
|
|
router, manager = _router_with_observed_fallback(observed_primary)
|
|
|
|
decision = await router.route(
|
|
"diagnose an alert",
|
|
context={"intent_hint": "diagnose"},
|
|
)
|
|
|
|
assert decision.selected_provider == AIProviderEnum.OLLAMA_GCP_A
|
|
assert [provider for provider, _model in decision.fallback_chain] == (
|
|
EXPECTED_FALLBACK
|
|
)
|
|
expected_observation = (
|
|
None if observed_primary == "ollama_gcp_a" else observed_primary
|
|
)
|
|
assert decision.observed_fallback_provider == expected_observation
|
|
assert "selected primary=ollama_gcp_a" in decision.routing_reason
|
|
manager.select_provider.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_probe_failure_cannot_authorize_cloud_jump() -> None:
|
|
manager = MagicMock()
|
|
manager.select_provider = AsyncMock(side_effect=RuntimeError("probe unavailable"))
|
|
router = AIRouter()
|
|
router._failover_manager = manager
|
|
|
|
decision = await router.route(
|
|
"delete resource",
|
|
context={"intent_hint": "delete"},
|
|
)
|
|
|
|
assert decision.selected_provider == AIProviderEnum.OLLAMA_GCP_A
|
|
assert [provider for provider, _model in decision.fallback_chain] == (
|
|
EXPECTED_FALLBACK
|
|
)
|
|
manager.select_provider.assert_awaited_once()
|