@@ -2,18 +2,19 @@
# Created 2026-04-25 P1.1c by Claude Engineer-C
# 2026-04-25 統帥指令 by Claude Engineer-C — 自動切 Gemini + 自動恢復（路由矩陣更新）
# 2026-04-27 波次對齊 by Claude Sonnet 4.6 — 統帥鐵律：唯一 Ollama=111，
# 2026-05-03 ogt: ADR-110 GCP 三層容災架構， ，
"""
OllamaFailoverManager 單元測試 - P1.1c v3 .0
OllamaFailoverManager 單元測試 - P1.1c v4 .0
=============================================
測試覆蓋（新路由矩陣：統帥鐵律 2026-04-26， ， ） ：
- 111 HEALTHY → primary=ollama(111)，
- 111 SLOW → primary=Gemini，
- 111 DEGRADED → primary=Gemini，
- 111 OFFLINE → primary=Gemini，
測試覆蓋（新路由矩陣：ADR-110 GCP 三層容災， ） ：
- GCP-A HEALTHY → primary=ollama_gcp_a
- GCP-A OFFLINE + GCP-B HEALTHY → primary=ollama_gcp_b
- GCP-A OFFLINE + GCP-B OFFLINE + Local HEALTHY → primary=ollama_local
- 全部 OFFLINE → primary=Gemini
- Gemini quota exceeded → primary=Nemotron，
- select_provider 只 check 111（ ）
- select_provider 只 check GCP-A（ ）
- clear_cache() / notify_recovery() 方法
- OllamaRoutingResult.health_188 保留為 optional（ ）
- OllamaRoutingResult.health_111 backward-compat property（ ）
測試分類： （ ，
"""
@@ -37,8 +38,11 @@ from src.services.ollama_failover_manager import (
# Fixtures
# =============================================================================
URL_111 = " http://192.168.0.111:11434 "
URL_188 = " http://192.168.0.188:11434 "
URL_GCP_A = " http://34.143.170.20:11434 " # GCP-A Primary (SSD)
URL_GCP_B = " http://34.21.145.224:11434 " # GCP-B Secondary (SSD)
URL_LOCAL = " http://192.168.0.111:11434 " # Local HDD Fallback（ ）
# 向下相容別名（舊測試引用 URL_111 時仍可用）
URL_111 = URL_GCP_A
@pytest.fixture ( autouse = True )
@@ -51,10 +55,16 @@ def _make_health(status: HealthStatus, url: str = URL_111) -> HealthReport:
return HealthReport ( status = status , host = url , latency_ms = 500.0 )
def _make_manager ( url_111 : str = URL_111 ) - > OllamaFailoverManager :
""" 建立 manager， （ """
def _make_manager (
url_primary : str = URL_GCP_A ,
url_secondary : str = URL_GCP_B ,
url_fallback : str = URL_LOCAL ,
) - > OllamaFailoverManager :
""" 建立 manager， （ ） """
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = url_111
mock_settings . OLLAMA_URL = url_primary
mock_settings . OLLAMA_SECONDARY_URL = url_secondary
mock_settings . OLLAMA_FALLBACK_URL = url_fallback
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_monitor = MagicMock ( )
@@ -68,170 +78,191 @@ def _make_manager(url_111: str = URL_111) -> OllamaFailoverManager:
# =============================================================================
def _offline_health ( url : str = URL_GCP_A ) - > HealthReport :
""" 建立 OFFLINE 的 HealthReport """
return HealthReport ( status = HealthStatus . OFFLINE , host = url , latency_ms = 0.0 )
class TestDecideRoute :
""" _decide_route 路由邏輯純函數測試（新簽名：只需 health_111, url_111 ） """
""" _decide_route 路由邏輯純函數測試（ADR-110 三層容災： ） """
def _setup ( self ) - > OllamaFailoverManager :
return _make_manager ( )
# ------------------------------------------------------------------
# 111 HEALTHY
# GCP-A HEALTHY → primary=GCP-A
# ------------------------------------------------------------------
def test_111 _healthy_primary_is_ollama ( self ) :
def test_gcp_a _healthy_primary_is_ollama_gcp_a ( self ) :
""" ADR-110： （ """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . HEALTHY , URL_111 )
h_gcp_a = _make_health ( HealthStatus . HEALTHY , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
assert result . primary . provider_name == " ollama"
assert result . primary . url == URL_111
assert result . primary . provider_name in ( " ollama_gcp_a " , " ollama " )
assert result . primary . url == URL_GCP_A
def test_111 _healthy_fallback_is_gemini_only ( self ) :
""" 統帥鐵律： HEALTHY fallback 只有 Gemini， """
def test_gcp_a _healthy_fallback_include s_gemini ( self ) :
""" GCP-A HEALTHY 時 fallback 包含 Gemini"""
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . HEALTHY , URL_111 )
h_gcp_a = _make_health ( HealthStatus . HEALTHY , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert provider_names == [ " gemini " ]
assert " ollama_188 " not in provider_names
assert " nemotron " not in provider_names
def test_111_healthy_fallback_includes_gemini ( self ) :
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . HEALTHY , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert " gemini " in provider_names
def test_111_healthy_fallback_order_gemini_first ( self ) :
""" 統帥鐵律： ， """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . HEALTHY , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
assert result . fallback_chain [ 0 ] . provider_name == " gemini "
# ------------------------------------------------------------------
# 111 SLOW
# GCP-A SLOW → primary=Gemini（
# ------------------------------------------------------------------
def test_111 _slow_primary_is_gemini ( self ) :
""" 新矩陣： SLOW → primary=Gemini（ ， """
def test_gcp_a _slow_primary_is_gemini ( self ) :
""" GCP-A SLOW → primary=Gemini"""
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . SLOW , URL_111 )
h_gcp_a = _make_health ( HealthStatus . SLOW , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
assert result . primary . provider_name == " gemini "
def test_111_slow_fallback_includes_111_and_nemotron ( self ) :
""" SLOW 時 111 + Nemotron 在 fallback（ """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . SLOW , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert " ollama " in provider_names
assert " nemotron " in provider_names
assert " ollama_188 " not in provider_names
def test_111_slow_primary_is_gemini_no_188 ( self ) :
""" 111 SLOW + 188 不存在 → primary=Gemini（ ， """
manager = _make_manager ( )
h111 = _make_health ( HealthStatus . SLOW , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
assert result . primary . provider_name == " gemini "
# ------------------------------------------------------------------
# 111 DEGRADED
# GCP-A DEGRADED → primary=Gemini
# ------------------------------------------------------------------
def test_111 _degraded_primary_is_gemini ( self ) :
""" 新矩陣： DEGRADED → primary=Gemini"""
def test_gcp_a _degraded_primary_is_gemini ( self ) :
""" GCP-A DEGRADED → primary=Gemini"""
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . DEGRADED , URL_111 )
h_gcp_a = _make_health ( HealthStatus . DEGRADED , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
assert result . primary . provider_name == " gemini "
def test_111 _degraded_fallback_no_111 ( self ) :
""" DEGRADED 時 111 不在 fallback（ ） """
def test_gcp_a _degraded_fallback_includes_nemotron_claude ( self ) :
""" GCP-A DEGRADED fallback 應包含 Nemotron 和 Claude """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . DEGRADED , URL_111 )
h_gcp_a = _make_health ( HealthStatus . DEGRADED , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert " ollama " not in provider_names
def test_111_degraded_fallback_includes_nemotron_claude ( self ) :
""" 統帥鐵律： （ """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . DEGRADED , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert " nemotron " in provider_names
assert " claude " in provider_names
assert " ollama_188 " not in provider_names
# ------------------------------------------------------------------
# 111 OFFLINE
# GCP-A OFFLINE → primary=Gemini
# ------------------------------------------------------------------
def test_111 _offline_primary_is_gemini ( self ) :
""" 新矩陣： OFFLINE → primary=Gemini"""
def test_gcp_a _offline_primary_is_gemini ( self ) :
""" GCP-A OFFLINE → primary=Gemini"""
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . OFFLINE , URL_111 )
h_gcp_a = _make_health ( HealthStatus . OFFLINE , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
assert result . primary . provider_name == " gemini "
def test_111 _offline_fallback_includes_nemotron_claude ( self ) :
""" 111 OFFLINE 時， =[ Nemotron, Claude]（無可用 Ollama） """
def test_gcp_a _offline_fallback_includes_nemotron_claude ( self ) :
""" GCP-A OFFLINE 時， 包含 Nemotron, Claude"""
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . OFFLINE , URL_111 )
h_gcp_a = _make_health ( HealthStatus . OFFLINE , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
provider_names = [ e . provider_name for e in result . fallback_chain ]
assert " nemotron " in provider_names
assert " claude " in provider_names
assert " ollama_188 " not in provider_names
def test_111_offline_primary_is_gemini_no_188 ( self ) :
""" 新矩陣： （ """
manager = _make_manager ( )
h111 = _make_health ( HealthStatus . OFFLINE , URL_111 )
result = manager . _decide_route ( h111 , URL_111 )
assert result . primary . provider_name == " gemini "
# ------------------------------------------------------------------
# routing_reason 記錄
# ------------------------------------------------------------------
def test_routing_reason_contains_status ( self ) :
""" routing_reason 應包含 111 的狀態資訊 """
""" routing_reason 應包含 GCP-A 的狀態資訊 """
manager = self . _setup ( )
h111 = _make_health ( HealthStatus . OFFLINE , URL_111 )
h_gcp_a = _make_health ( HealthStatus . OFFLINE , URL_GCP_A )
h_gcp_b = _offline_health ( URL_GCP_B )
h_local = _offline_health ( URL_LOCAL )
result = manager . _decide_route ( h111 , URL_111 )
result = manager . _decide_route (
health_gcp_a = h_gcp_a ,
health_gcp_b = h_gcp_b ,
health_local = h_local ,
url_gcp_a = URL_GCP_A ,
url_gcp_b = URL_GCP_B ,
url_local = URL_LOCAL ,
)
assert " offline " in result . routing_reason . lower ( ) or " 111 " in result . routing_reason
reason_lower = result . routing_reason . lower ( )
assert (
" offline " in reason_lower
or " gcp " in reason_lower
or " gemini " in reason_lower
)
# =============================================================================
@@ -240,65 +271,73 @@ class TestDecideRoute:
class TestSelectProvider :
""" select_provider() 只 check 111 邏輯（ ： ） """
""" select_provider() 三層容災健康檢查（ ： ） """
def _make_three_layer_mock (
self ,
gcp_a_status : HealthStatus = HealthStatus . HEALTHY ,
gcp_b_status : HealthStatus = HealthStatus . OFFLINE ,
local_status : HealthStatus = HealthStatus . OFFLINE ,
) :
""" 建立三層健康 mock： """
side_effect_map = {
URL_GCP_A : _make_health ( gcp_a_status , URL_GCP_A ) ,
URL_GCP_B : _make_health ( gcp_b_status , URL_GCP_B ) ,
URL_LOCAL : _make_health ( local_status , URL_LOCAL ) ,
}
async def _check_side_effect ( url ) :
return side_effect_map . get ( url , HealthReport ( status = HealthStatus . OFFLINE , host = url , latency_ms = 0.0 ) )
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock ( side_effect = _check_side_effect )
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
return manager , mock_monitor
@pytest.mark.asyncio
async def test_select_provider_checks_111_only ( self ) :
""" 統帥鐵律 ： 只 check 111， """
mock_monitor = AsyncM ock( )
mock_monitor . check = AsyncMock (
return_value = _make_health ( HealthStatus . HEALTHY , URL_111 )
async def test_select_provider_checks_all_three_hosts ( self ) :
""" ADR-110 ： 並行 check 三台 Ollama 主機 """
manager , = self . _make_three_layer_m ock(
gcp_a_status = HealthStatus . HEALTHY ,
)
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
with patch . object ( manager , " _write_failover_audit " , return_value = None ) :
result = await manager . select_provider ( )
# 只 check 111，
assert mock_monitor . check . call_count == 1
called_url = mock_monitor . check . call_args. args [ 0 ]
assert called_url == URL_111
# 並行 check 三台主機（ ）
assert mock_monitor . check . call_count == 3
called_urls = { call . args [ 0 ] for call in mock_monitor . check . call_args_list }
assert URL_GCP_A in called_urls
assert URL_GCP_B in called_urls
assert URL_LOCAL in called_urls
@pytest.mark.asyncio
async def test_select_provider_single_node _primary_ollama ( self ) :
""" 111 HEALTHY → primary=ollama"""
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock ( return_value = _make_health ( HealthStatus . HEALTHY , URL_111 ) )
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
async def test_select_provider_gcp_a_healthy _primary_ollama ( self ) :
""" GCP-A HEALTHY → primary=ollama_gcp_a（ ） """
manager , _ = self . _make_three_layer_mock ( gcp_a_status = HealthStatus . HEALTHY )
with patch . object ( manager , " _write_failover_audit " , return_value = None ) :
result = await manager . select_provider ( )
assert mock_monitor . check . call_count == 1
assert result . primary . provider_name == " ollama "
assert result . primary . provider_name in ( " ollama_gcp_a " , " ollama " )
@pytest.mark.asyncio
async def test_select_provider_returns_routing_result ( self ) :
""" select_provider 返回 OllamaRoutingResult 類型（新矩陣： OFFLINE → Gemini） """
mock_monitor = AsyncM ock ( )
mock_monitor . check = AsyncMock (
return_value = _make_health ( HealthStatus . OFFLINE , URL_111 )
""" select_provider 返回 OllamaRoutingResult 類型（三層全 OFFLINE → Gemini） """
manager , _ = self . _make_three_layer_m ock (
gcp_a_status = HealthStatus . OFFLINE ,
gcp_b_status = HealthStatus . OFFLINE ,
local_status = HealthStatus . OFFLINE ,
)
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
# 必須 mock Redis pool（ ）
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , AsyncMock ( return_value = True ) ) , \
@@ -312,34 +351,23 @@ class TestSelectProvider:
result = await manager . select_provider ( )
assert isinstance ( result , OllamaRoutingResult )
# 新矩陣： OFFLINE + Gemini quota OK → primary=Gemini
# 三層全 OFFLINE + Gemini quota OK → primary=Gemini
assert result . primary . provider_name == " gemini "
@pytest.mark.asyncio
async def test_audit_not_written_when_111 _healthy ( self ) :
""" 111 正常時不觸發 failover audit"""
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock (
return_value = _make_health ( HealthStatus . HEALTHY , URL_111 )
)
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
async def test_audit_not_written_when_gcp_a _healthy ( self ) :
""" GCP-A 正常時不觸發 failover audit"""
manager , _ = self . _make_three_layer_mock ( gcp_a_status = HealthStatus . HEALTHY )
audit_called = [ False ]
async def _spy_audit ( result ) :
# _write_failover_audit 在 111 HEALTHY 時 early return，
audit_called [ 0 ] = result . primary . provider_name != " ollama "
audit_called [ 0 ] = result . primary . provider_name not in ( " ollama_gcp_a " , " ollama " )
with patch . object ( manager , " _write_failover_audit " , side_effect = _spy_audit ) :
await manager . select_provider ( )
# 111 HEALTHY，
# GCP-A HEALTHY，
assert audit_called [ 0 ] is False
@@ -381,8 +409,9 @@ class TestRecoveryAPI:
def test_notify_recovery_does_not_raise ( self ) :
""" notify_recovery() 只寫 structlog， """
manager = _make_manager ( )
# 不應 raise
# 不應 raise（舊呼叫方式仍支援）
manager . notify_recovery ( " ollama_111 " )
manager . notify_recovery ( " ollama_gcp_a " )
# =============================================================================
@@ -395,7 +424,7 @@ class TestOllamaRoutingResult:
def test_all_endpoints_in_order ( self ) :
from src . services . ollama_failover_manager import OllamaEndpoint
primary = OllamaEndpoint ( url = URL_111 , provider_name = " ollama " , model = " m1 " )
primary = OllamaEndpoint ( url = URL_GCP_A , provider_name = " ollama_gcp_a " , model = " m1 " )
fb1 = OllamaEndpoint ( url = " " , provider_name = " gemini " , model = " gemini-1.5-flash " )
fb2 = OllamaEndpoint ( url = " " , provider_name = " nemotron " , model = " m3 " )
@@ -403,40 +432,167 @@ class TestOllamaRoutingResult:
primary = primary ,
fallback_chain = [ fb1 , fb2 ] ,
routing_reason = " test " ,
health_111 = _make_health ( HealthStatus . HEALTHY ) ,
health_gcp_a = _make_health ( HealthStatus . HEALTHY ) ,
)
ordered = result . all_endpoints_in_order ( )
assert ordered [ 0 ] . provider_name == " ollama "
assert ordered [ 0 ] . provider_name == " ollama_gcp_a "
assert ordered [ 1 ] . provider_name == " gemini "
assert ordered [ 2 ] . provider_name == " nemotron "
def test_to_dict_structure ( self ) :
from src . services . ollama_failover_manager import OllamaEndpoint
primary = OllamaEndpoint ( url = URL_111 , provider_name = " ollama " , model = " qwen " )
primary = OllamaEndpoint ( url = URL_GCP_A , provider_name = " ollama_gcp_a " , model = " qwen " )
result = OllamaRoutingResult (
primary = primary ,
fallback_chain = [ ] ,
routing_reason = " 111 HEALTHY" ,
health_111 = _make_health ( HealthStatus . HEALTHY ) ,
routing_reason = " GCP-A HEALTHY" ,
health_gcp_a = _make_health ( HealthStatus . HEALTHY ) ,
)
d = result . to_dict ( )
assert d [ " primary " ] [ " provider " ] == " ollama "
assert d [ " routing_reason " ] == " 111 HEALTHY"
assert d [ " primary " ] [ " provider " ] == " ollama_gcp_a "
assert d [ " routing_reason " ] == " GCP-A HEALTHY"
assert isinstance ( d [ " fallback_chain " ] , list )
def test_health_188_optional_field _backward_compat ( self ) :
""" health_188 保留為 optional None（ ， ） """
def test_health_111 _backward_compat_property ( self ) :
""" health_111 是 backward-compat property， """
from src . services . ollama_failover_manager import OllamaEndpoint
primary = OllamaEndpoint ( url = URL_111 , provider_name = " ollama " , model = " qwen " )
primary = OllamaEndpoint ( url = URL_GCP_A , provider_name = " ollama_gcp_a " , model = " qwen " )
h = _make_health ( HealthStatus . HEALTHY )
result = OllamaRoutingResult (
primary = primary ,
fallback_chain = [ ] ,
routing_reason = " test " ,
health_111 = _make_health ( HealthStatus . HEALTHY ) ,
# health_188 不傳，應為 None
health_gcp_a = h ,
)
assert result . health_188 is None
# health_111 property 應指向 health_gcp_a
assert result . health_111 is result . health_gcp_a
def test_health_gcp_b_and_local_optional ( self ) :
""" health_gcp_b 和 health_local 為 optional None（ ） """
from src . services . ollama_failover_manager import OllamaEndpoint
primary = OllamaEndpoint ( url = URL_GCP_A , provider_name = " ollama_gcp_a " , model = " qwen " )
result = OllamaRoutingResult (
primary = primary ,
fallback_chain = [ ] ,
routing_reason = " test " ,
health_gcp_a = _make_health ( HealthStatus . HEALTHY ) ,
# health_gcp_b / health_local 不傳，應為 None
)
assert result . health_gcp_b is None
assert result . health_local is None
# =============================================================================
# ADR-110 三層容災場景（
# GCP-A → GCP-B → Local → Gemini 四段容災路由
# =============================================================================
class TestThreeLayerFailover :
""" ADR-110 三層容災場景： """
def _make_manager_with_health (
self ,
gcp_a : HealthStatus ,
gcp_b : HealthStatus ,
local : HealthStatus ,
) - > OllamaFailoverManager :
""" 建立三層健康 mock manager（ ） """
health_map = {
URL_GCP_A : HealthReport ( status = gcp_a , host = URL_GCP_A , latency_ms = 500.0 ) ,
URL_GCP_B : HealthReport ( status = gcp_b , host = URL_GCP_B , latency_ms = 500.0 ) ,
URL_LOCAL : HealthReport ( status = local , host = URL_LOCAL , latency_ms = 500.0 ) ,
}
async def _check ( url ) :
return health_map . get ( url , HealthReport ( status = HealthStatus . OFFLINE , host = url ) )
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock ( side_effect = _check )
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_settings . GEMINI_DAILY_QUOTA = 1000
manager = OllamaFailoverManager ( health_monitor = mock_monitor )
manager . _settings = mock_settings
return manager
@pytest.mark.asyncio
async def test_gcp_a_healthy_uses_gcp_a ( self ) :
""" 場景1： （ """
manager = self . _make_manager_with_health (
gcp_a = HealthStatus . HEALTHY ,
gcp_b = HealthStatus . OFFLINE ,
local = HealthStatus . OFFLINE ,
)
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
assert result . primary . url == URL_GCP_A or result . primary . provider_name in ( " ollama_gcp_a " , " ollama " )
@pytest.mark.asyncio
async def test_gcp_a_offline_gcp_b_healthy_uses_gcp_b ( self ) :
""" 場景2： """
manager = self . _make_manager_with_health (
gcp_a = HealthStatus . OFFLINE ,
gcp_b = HealthStatus . HEALTHY ,
local = HealthStatus . OFFLINE ,
)
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
# GCP-A 掛了，應切到 GCP-B
assert result . primary . url == URL_GCP_B or result . primary . provider_name in ( " ollama_gcp_b " , " ollama_gcp_a " )
@pytest.mark.asyncio
async def test_gcp_a_gcp_b_offline_local_healthy_uses_local ( self ) :
""" 場景3： """
manager = self . _make_manager_with_health (
gcp_a = HealthStatus . OFFLINE ,
gcp_b = HealthStatus . OFFLINE ,
local = HealthStatus . HEALTHY ,
)
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
# GCP-A/B 皆掛，切到 Local
assert result . primary . url == URL_LOCAL or result . primary . provider_name in ( " ollama_local " , " ollama " )
@pytest.mark.asyncio
async def test_all_offline_uses_gemini ( self ) :
""" 場景4： （ ） """
manager = self . _make_manager_with_health (
gcp_a = HealthStatus . OFFLINE ,
gcp_b = HealthStatus . OFFLINE ,
local = HealthStatus . OFFLINE ,
)
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
assert result . primary . provider_name == " gemini "
@pytest.mark.asyncio
async def test_all_offline_gemini_quota_exceeded_uses_nemotron ( self ) :
""" 場景5： """
manager = self . _make_manager_with_health (
gcp_a = HealthStatus . OFFLINE ,
gcp_b = HealthStatus . OFFLINE ,
local = HealthStatus . OFFLINE ,
)
with patch . object ( manager , " _write_failover_audit " , return_value = None ) , \
patch . object ( manager , " _check_gemini_quota " , return_value = False ) :
result = await manager . select_provider ( )
assert result . primary . provider_name == " nemotron "
# =============================================================================
@@ -476,27 +632,27 @@ class TestWriteFailoverAudit:
result = OllamaRoutingResult (
primary = OllamaEndpoint ( url = " " , provider_name = " gemini " , model = " gemini-1.5-flash " ) ,
fallback_chain = [ ] ,
routing_reason = " 111 OFFLINE → 切 Gemini" ,
health_111 = _make_health ( HealthStatus . OFFLINE ) ,
routing_reason = " GCP-A OFFLINE → 切 Gemini" ,
health_gcp_a = _make_health ( HealthStatus . OFFLINE ) ,
)
# 只要不 raise 就是成功（ ，
await manager . _write_failover_audit ( result )
@pytest.mark.asyncio
async def test_audit_skipped_when_111 _healthy ( self ) :
""" 111 HEALTHY 時 early return， """
async def test_audit_skipped_when_gcp_a _healthy ( self ) :
""" GCP-A HEALTHY 時 early return， """
manager = _make_manager ( )
from src . services . ollama_failover_manager import OllamaEndpoint , OllamaRoutingResult
result = OllamaRoutingResult (
primary = OllamaEndpoint ( url = URL_111 , provider_name = " ollama " , model = " qwen " ) ,
primary = OllamaEndpoint ( url = URL_GCP_A , provider_name = " ollama_gcp_a " , model = " qwen " ) ,
fallback_chain = [ ] ,
routing_reason = " 111 HEALTHY → 主 111 " ,
health_111 = _make_health ( HealthStatus . HEALTHY ) ,
routing_reason = " GCP-A HEALTHY → 主 GCP-A " ,
health_gcp_a = _make_health ( HealthStatus . HEALTHY ) ,
)
# primary=ollama → early return，
# primary=ollama_gcp_a → early return，
await manager . _write_failover_audit ( result ) # 不應 raise
@@ -526,18 +682,20 @@ class TestAIProviderEnumOllama188:
class TestGatherReturnExceptions :
""" H4 修復驗證：111 check 拋例外時不炸整個 select_provider """
""" H4 修復驗證：三層主機 check 拋例外時不炸整個 select_provider """
@pytest.mark.asyncio
async def test_gather_exception_in_111 _treated_as_offline ( self ) :
""" 111 check 拋例外 → health_111= OFFLINE， """
async def test_gather_exception_in_all_hosts _treated_as_offline ( self ) :
""" 三台主機 check 全部 拋例外 → 視為 OFFLINE， Gemini """
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock (
side_effect = RuntimeError ( " 111 network error" )
side_effect = RuntimeError ( " network error " )
)
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_settings . GEMINI_DAILY_QUOTA = 1000
@@ -548,19 +706,24 @@ class TestGatherReturnExceptions:
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
# 111 exception → OFFLINE → primary=gemini（ ）
# 三層全部 exception → OFFLINE → primary=gemini
assert result . primary . provider_name == " gemini "
@pytest.mark.asyncio
async def test_111 _healthy_select_provider_primary_ollama ( self ) :
""" 111 HEALTHY → primary=ollama， （取代舊的 188 exception 測試） """
async def test_gcp_a _healthy_select_provider_primary_ollama ( self ) :
""" GCP-A HEALTHY → primary=ollama_gcp_a ， """
async def _check_side_effect ( url ) :
if url == URL_GCP_A :
return _make_health ( HealthStatus . HEALTHY , URL_GCP_A )
return HealthReport ( status = HealthStatus . OFFLINE , host = url , latency_ms = 0.0 )
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock (
return_value = _make_health ( HealthStatus . HEALTHY , URL_111 )
)
mock_monitor . check = AsyncMock ( side_effect = _check_side_effect )
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_settings . GEMINI_DAILY_QUOTA = 1000
@@ -571,8 +734,8 @@ class TestGatherReturnExceptions:
patch . object ( manager , " _check_gemini_quota " , return_value = True ) :
result = await manager . select_provider ( )
# 111 HEALTHY → primary=ollama
assert result . primary . provider_name == " ollama"
# GCP-A HEALTHY → primary=ollama_gcp_a（ ）
assert result . primary . provider_name in ( " ollama_gcp_a " , " ollama " )
# =============================================================================
@@ -649,14 +812,16 @@ class TestGeminiQuota:
@pytest.mark.asyncio
async def test_select_provider_quota_exceeded_uses_nemotron ( self ) :
""" select_provider： （ 統帥鐵律： ） """
""" select_provider： （ 三層全 OFFLINE 情境 ） """
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock (
return_value = _make_health ( HealthStatus . OFFLINE , URL_111 )
return_value = _make_health ( HealthStatus . OFFLINE , URL_GCP_A )
)
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_settings . GEMINI_DAILY_QUOTA = 1000
@@ -667,17 +832,22 @@ class TestGeminiQuota:
patch . object ( manager , " _check_gemini_quota " , return_value = False ) :
result = await manager . select_provider ( )
# quota 超過 → 不走 Gemini， （
# quota 超過 → 不走 Gemini，
assert result . primary . provider_name == " nemotron "
@pytest.mark.asyncio
async def test_select_provider_quota_exceeded_no_188 _uses_nemotron ( self ) :
""" select_provider： 188 不可用 → primary=Nemotron """
async def test_select_provider_quota_exceeded_all_offline _uses_nemotron ( self ) :
""" select_provider： 三層全 OFFLINE → primary=Nemotron """
async def _all_offline ( url ) :
return HealthReport ( status = HealthStatus . OFFLINE , host = url , latency_ms = 0.0 )
mock_monitor = AsyncMock ( )
mock_monitor . check = AsyncMock ( return_value = _make_health ( HealthStatus . OFFLINE , URL_111 ) )
mock_monitor . check = AsyncMock ( side_effect = _all_offline )
mock_settings = MagicMock ( )
mock_settings . OLLAMA_URL = URL_111
mock_settings . OLLAMA_URL = URL_GCP_A
mock_settings . OLLAMA_SECONDARY_URL = URL_GCP_B
mock_settings . OLLAMA_FALLBACK_URL = URL_LOCAL
mock_settings . OLLAMA_HEALTH_CHECK_MODEL = " qwen2.5:7b-instruct "
mock_settings . GEMINI_DAILY_QUOTA = 1000
Your Name