89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
"""AI route readback separates execution selection from health observation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from src.services import platform_operator_service
|
|
from src.services.ollama_failover_manager import (
|
|
OllamaEndpoint,
|
|
OllamaRoutingResult,
|
|
)
|
|
from src.services.ollama_health_monitor import HealthReport, HealthStatus
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_ollama_offline_never_labels_gemini_selected_primary(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
gcp_a = OllamaEndpoint(
|
|
url="http://gcp-a:11434",
|
|
provider_name="ollama_gcp_a",
|
|
model="qwen3:14b",
|
|
)
|
|
gcp_b = OllamaEndpoint(
|
|
url="http://gcp-b:11434",
|
|
provider_name="ollama_gcp_b",
|
|
model="qwen3:14b",
|
|
)
|
|
local = OllamaEndpoint(
|
|
url="http://host111:11434",
|
|
provider_name="ollama_local",
|
|
model="qwen3:14b",
|
|
)
|
|
gemini = OllamaEndpoint(
|
|
url="",
|
|
provider_name="gemini",
|
|
model="gemini-2.5-flash-lite",
|
|
)
|
|
offline = HealthReport(status=HealthStatus.OFFLINE, reason="offline")
|
|
route = OllamaRoutingResult(
|
|
primary=gcp_a,
|
|
fallback_chain=[gcp_b, local, gemini],
|
|
routing_reason=(
|
|
"selected primary remains GCP-A; observed Gemini policy candidate"
|
|
),
|
|
health_gcp_a=offline,
|
|
health_gcp_b=offline,
|
|
health_local=offline,
|
|
observed_fallback=gemini,
|
|
)
|
|
manager = MagicMock()
|
|
manager.select_provider = AsyncMock(return_value=route)
|
|
monkeypatch.setattr(
|
|
platform_operator_service,
|
|
"get_ollama_failover_manager",
|
|
lambda: manager,
|
|
)
|
|
monkeypatch.setattr(
|
|
platform_operator_service,
|
|
"_latest_ai_route_repair_evidence",
|
|
AsyncMock(return_value=None),
|
|
)
|
|
|
|
response = await platform_operator_service.get_ai_route_status("deep_rca")
|
|
|
|
assert response["schema_version"] == "awooop_ai_route_status_v2"
|
|
assert response["selection_mode"] == "fixed_production_order"
|
|
assert response["selected_provider"] == "ollama_gcp_a"
|
|
assert response["selected_primary"]["provider_name"] == "ollama_gcp_a"
|
|
assert response["observed_fallback"] == {
|
|
"priority": 4,
|
|
"provider_name": "gemini",
|
|
"url": None,
|
|
"model": "gemini-2.5-flash-lite",
|
|
"runtime": "cloud",
|
|
"observation_only": True,
|
|
"requires_generation_cost_guard": True,
|
|
}
|
|
assert [item["provider_name"] for item in response["fallback_chain"]] == [
|
|
"ollama_gcp_b",
|
|
"ollama_local",
|
|
"gemini",
|
|
]
|
|
assert response["gemini_activation_evidence"] is None
|
|
assert response["lane_mode"] == "unavailable"
|
|
assert response["active_lane"] is None
|