363 lines
13 KiB
Python
363 lines
13 KiB
Python
# apps/api/tests/test_failover_e2e_dispatch.py | 2026-05-06 @ Asia/Taipei
|
||
# 2026-05-06 Codex — 188 不再作為 Ollama Provider；驗證 ollama_local dispatch。
|
||
"""
|
||
E2E：executor dispatch 層驗證
|
||
===============================
|
||
測試覆蓋（補全 B4 — 整合測試只驗決策層，未驗執行層）：
|
||
|
||
1. registry 確實有 ollama_local provider，且沒有 ollama_188 provider
|
||
2. OllamaLocalProvider.is_enabled 在有 OLLAMA_FALLBACK_URL 時為 True
|
||
3. OllamaLocalProvider.is_enabled 在 OLLAMA_FALLBACK_URL 空字串時為 False
|
||
4. OllamaLocalProvider.analyze() 真的把 HTTP 打到 OLLAMA_FALLBACK_URL（攔截 httpx）
|
||
5. executor.execute(provider_order=["ollama_local"]) 真的路由到 local URL
|
||
6. Gemini quota pipeline 並行 5 次不超發（B3 atomic 驗證）
|
||
7. Gemini quota TTL 第一次呼叫即設定
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
|
||
# =============================================================================
|
||
# B1：registry 健全性
|
||
# =============================================================================
|
||
|
||
|
||
def test_registry_has_ollama_local_provider_without_ollama_188():
|
||
"""_init_registry() 後 registry 必須有 ollama_local，且不得有 ollama_188"""
|
||
from src.services.ai_router import _init_registry
|
||
|
||
registry = _init_registry()
|
||
# registry.get() 只返回 is_enabled=True 的 provider
|
||
# 用 _providers dict 直接檢查（不管 is_enabled）
|
||
assert "ollama_local" in registry._providers
|
||
assert "ollama_188" not in registry._providers
|
||
|
||
|
||
def test_ollama_local_provider_name():
|
||
"""OllamaLocalProvider.name == 'ollama_local'"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
p = OllamaLocalProvider()
|
||
assert p.name == "ollama_local"
|
||
|
||
|
||
def test_ollama_local_provider_privacy_level():
|
||
"""OllamaLocalProvider.privacy_level == 'local'（本地推理，可接機密資料）"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
p = OllamaLocalProvider()
|
||
assert p.privacy_level == "local"
|
||
|
||
|
||
# =============================================================================
|
||
# B1：is_enabled 邏輯
|
||
# =============================================================================
|
||
|
||
|
||
def test_ollama_local_is_enabled_with_fallback_url(monkeypatch):
|
||
"""OLLAMA_FALLBACK_URL 有值 + ENABLE_OLLAMA_LOCAL 未設 → is_enabled == True"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
monkeypatch.setenv("ENABLE_OLLAMA_LOCAL", "true")
|
||
# patch settings 的 OLLAMA_FALLBACK_URL
|
||
mock_settings = MagicMock()
|
||
mock_settings.OLLAMA_FALLBACK_URL = "http://192.168.0.111:11434"
|
||
mock_settings.OPENCLAW_TIMEOUT = "60"
|
||
|
||
p = OllamaLocalProvider()
|
||
# 直接 patch module-level settings 物件
|
||
with patch("src.services.ai_providers.ollama.settings", mock_settings):
|
||
assert p.is_enabled is True
|
||
|
||
|
||
def test_ollama_local_is_disabled_without_fallback_url(monkeypatch):
|
||
"""OLLAMA_FALLBACK_URL 空字串 → is_enabled == False（local 節點未設定）"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
monkeypatch.setenv("ENABLE_OLLAMA_LOCAL", "true")
|
||
mock_settings = MagicMock()
|
||
mock_settings.OLLAMA_FALLBACK_URL = ""
|
||
|
||
p = OllamaLocalProvider()
|
||
with patch("src.services.ai_providers.ollama.settings", mock_settings):
|
||
assert p.is_enabled is False
|
||
|
||
|
||
def test_ollama_local_is_disabled_by_env_flag(monkeypatch):
|
||
"""ENABLE_OLLAMA_LOCAL=false → is_enabled == False（即使有 URL）"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
monkeypatch.setenv("ENABLE_OLLAMA_LOCAL", "false")
|
||
mock_settings = MagicMock()
|
||
mock_settings.OLLAMA_FALLBACK_URL = "http://192.168.0.111:11434"
|
||
|
||
p = OllamaLocalProvider()
|
||
with patch("src.services.ai_providers.ollama.settings", mock_settings):
|
||
assert p.is_enabled is False
|
||
|
||
|
||
# =============================================================================
|
||
# B4 核心：HTTP dispatch 驗證
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ollama_local_analyze_dispatches_to_fallback_url():
|
||
"""
|
||
B4 核心：OllamaLocalProvider.analyze() 必須把 HTTP 打到 OLLAMA_FALLBACK_URL。
|
||
攔截 httpx.AsyncClient.post，記錄實際呼叫 URL，斷言包含本地 fallback IP。
|
||
"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
FALLBACK_URL = "http://192.168.0.111:11434"
|
||
captured_urls: list[str] = []
|
||
|
||
mock_response = MagicMock()
|
||
mock_response.status_code = 200
|
||
mock_response.raise_for_status = MagicMock()
|
||
mock_response.json = MagicMock(return_value={
|
||
"response": '{"action_title": "test", "confidence": 0.9}',
|
||
"eval_count": 10,
|
||
"prompt_eval_count": 5,
|
||
})
|
||
|
||
# httpx.AsyncClient.post 是 instance method，mock 需要接受 self
|
||
async def mock_post(self_client, url, **kwargs):
|
||
captured_urls.append(url)
|
||
return mock_response
|
||
|
||
mock_settings = MagicMock()
|
||
mock_settings.OLLAMA_FALLBACK_URL = FALLBACK_URL
|
||
mock_settings.OLLAMA_HEALTH_CHECK_MODEL = "qwen2.5:7b-instruct"
|
||
mock_settings.OPENCLAW_TIMEOUT = "60"
|
||
mock_settings.OLLAMA_DIAGNOSE_TIMEOUT_SECONDS = 200
|
||
|
||
# mock model_registry
|
||
mock_registry = MagicMock()
|
||
mock_registry.get_model = MagicMock(return_value="qwen2.5:7b-instruct")
|
||
mock_registry.get_provider_options = MagicMock(return_value={
|
||
"num_predict": 1024,
|
||
"temperature": 0.1,
|
||
"top_p": 0.9,
|
||
})
|
||
|
||
provider = OllamaLocalProvider()
|
||
|
||
with patch("src.services.ai_providers.ollama.settings", mock_settings):
|
||
with patch("src.services.ai_providers.ollama.get_model_registry", return_value=mock_registry):
|
||
import httpx
|
||
# patch httpx.AsyncClient.post（class-level，適用所有 instance）
|
||
with patch.object(httpx.AsyncClient, "post", new=mock_post):
|
||
result = await provider.analyze("test prompt", context={})
|
||
|
||
assert len(captured_urls) > 0, "analyze() 未發出任何 HTTP 請求"
|
||
assert any("192.168.0.111" in url for url in captured_urls), (
|
||
f"HTTP 請求未打到 local fallback，實際 URL: {captured_urls}"
|
||
)
|
||
assert result.provider == "ollama_local"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ollama_local_analyze_returns_error_when_no_fallback_url():
|
||
"""OLLAMA_FALLBACK_URL 未設定 → analyze() 應返回 success=False，不發 HTTP"""
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
|
||
mock_settings = MagicMock()
|
||
mock_settings.OLLAMA_FALLBACK_URL = ""
|
||
|
||
provider = OllamaLocalProvider()
|
||
with patch("src.services.ai_providers.ollama.settings", mock_settings):
|
||
result = await provider.analyze("test prompt")
|
||
|
||
assert result.success is False
|
||
assert result.provider == "ollama_local"
|
||
assert "OLLAMA_FALLBACK_URL" in (result.error or "")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_executor_dispatches_ollama_local_to_fallback_url():
|
||
"""
|
||
B4 執行層：AIRouterExecutor.execute(provider_order=["ollama_local"])
|
||
應路由到 OllamaLocalProvider，且 HTTP 打到 OLLAMA_FALLBACK_URL。
|
||
"""
|
||
from src.services.ai_router import AIProviderRegistry, AIRouterExecutor, reset_ai_router
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
from src.services.ai_providers.interfaces import AIResult
|
||
|
||
reset_ai_router()
|
||
|
||
FALLBACK_URL = "http://192.168.0.111:11434"
|
||
captured_urls: list[str] = []
|
||
|
||
# 建立真實 registry，只登錄 ollama_local
|
||
registry = AIProviderRegistry()
|
||
|
||
# mock analyze 讓它回傳成功，但驗 URL 路徑
|
||
async def fake_analyze(prompt, context=None):
|
||
captured_urls.append(f"{FALLBACK_URL}/api/generate")
|
||
return AIResult(
|
||
raw_response='{"action_title":"ok","confidence":0.9}',
|
||
success=True,
|
||
provider="ollama_local",
|
||
tokens=10,
|
||
)
|
||
|
||
mock_settings_global = MagicMock()
|
||
mock_settings_global.OLLAMA_FALLBACK_URL = FALLBACK_URL
|
||
|
||
# 建立 OllamaLocalProvider，mock 其 analyze + is_enabled
|
||
provider = OllamaLocalProvider()
|
||
provider.analyze = fake_analyze # type: ignore[method-assign]
|
||
|
||
# 強制 is_enabled = True（繞過 settings patch 的複雜度）
|
||
type(provider).is_enabled = property(lambda self: True)
|
||
|
||
registry.register(provider)
|
||
executor = AIRouterExecutor(registry)
|
||
|
||
# mock Redis（不依賴真實 Redis）
|
||
mock_redis = AsyncMock()
|
||
mock_redis.get = AsyncMock(return_value=None)
|
||
mock_redis.set = AsyncMock(return_value=True)
|
||
|
||
with patch("src.core.redis_client.get_redis", return_value=mock_redis):
|
||
with patch("src.services.ai_router._settings") as mock_settings:
|
||
mock_settings.MOCK_MODE = False
|
||
result = await executor.execute(
|
||
prompt="test alert",
|
||
provider_order=["ollama_local"],
|
||
context={},
|
||
)
|
||
|
||
assert result.success is True, f"execute 失敗: {result.error}"
|
||
assert result.provider == "ollama_local", f"provider 不是 ollama_local: {result.provider}"
|
||
assert any("192.168.0.111" in u for u in captured_urls), (
|
||
f"HTTP 未打到 local fallback，captured: {captured_urls}"
|
||
)
|
||
|
||
|
||
# =============================================================================
|
||
# B3：Gemini quota atomic pipeline 驗證
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_gemini_quota_concurrent_no_overshoot():
|
||
"""
|
||
B3 atomic 驗證：5 個並行呼叫 _check_gemini_quota()，quota=5。
|
||
pipeline 原子遞增 → counter 嚴格等於 5（不超發）。
|
||
第 6 次呼叫應返回 False。
|
||
"""
|
||
from src.services.ollama_failover_manager import OllamaFailoverManager
|
||
from src.services.ollama_health_monitor import OllamaHealthMonitor
|
||
|
||
# 用真正的 in-memory counter 模擬 Redis pipeline
|
||
_store: dict[str, int] = {}
|
||
|
||
def make_mock_redis():
|
||
redis = MagicMock()
|
||
|
||
class FakePipeline:
|
||
def __init__(self):
|
||
self._key = None
|
||
self._nx_val = 0
|
||
self._ex = None
|
||
|
||
def set(self, key, val, ex=None, nx=False):
|
||
self._key = key
|
||
self._nx_val = val
|
||
self._ex = ex
|
||
return self
|
||
|
||
def incr(self, key):
|
||
self._key = key
|
||
return self
|
||
|
||
async def execute(self):
|
||
key = self._key
|
||
# NX set: only if not exists
|
||
if key not in _store:
|
||
_store[key] = self._nx_val
|
||
# INCR
|
||
_store[key] = _store.get(key, 0) + 1
|
||
new_val = _store[key]
|
||
return [True, new_val]
|
||
|
||
redis.pipeline = MagicMock(return_value=FakePipeline())
|
||
return redis
|
||
|
||
mock_settings = MagicMock()
|
||
mock_settings.GEMINI_DAILY_QUOTA = 5
|
||
|
||
mock_monitor = MagicMock(spec=OllamaHealthMonitor)
|
||
manager = OllamaFailoverManager(health_monitor=mock_monitor)
|
||
manager._settings = mock_settings
|
||
|
||
call_count = 0
|
||
|
||
async def patched_check():
|
||
nonlocal call_count
|
||
mock_redis = make_mock_redis()
|
||
with patch("src.core.redis_client.get_redis", return_value=mock_redis):
|
||
return await manager._check_gemini_quota()
|
||
|
||
# 5 個並行呼叫，quota=5，每個都應返回 True
|
||
results = await asyncio.gather(*[patched_check() for _ in range(5)])
|
||
assert all(results), f"5 個並行呼叫中有失敗: {results}"
|
||
|
||
# 第 6 次（超出 quota）應返回 False
|
||
# 重置 store 到 quota 值，模擬已滿
|
||
_store.clear()
|
||
for _ in range(5):
|
||
await patched_check()
|
||
|
||
result_6 = await patched_check()
|
||
assert result_6 is False, f"第 6 次超出 quota 應返回 False，實際: {result_6}"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_gemini_quota_ttl_set_atomically():
|
||
"""
|
||
B3 TTL 驗證：第一次呼叫 _check_gemini_quota() 後，
|
||
pipeline 的 SET NX 應已設定 TTL（不依賴分開的 EXPIRE）。
|
||
"""
|
||
from src.services.ollama_failover_manager import OllamaFailoverManager
|
||
from src.services.ollama_health_monitor import OllamaHealthMonitor
|
||
|
||
set_calls: list[dict] = []
|
||
|
||
class CapturingPipeline:
|
||
def set(self, key, val, ex=None, nx=False):
|
||
set_calls.append({"key": key, "val": val, "ex": ex, "nx": nx})
|
||
return self
|
||
|
||
def incr(self, key):
|
||
return self
|
||
|
||
async def execute(self):
|
||
return [True, 1]
|
||
|
||
mock_redis = MagicMock()
|
||
mock_redis.pipeline = MagicMock(return_value=CapturingPipeline())
|
||
|
||
mock_settings = MagicMock()
|
||
mock_settings.GEMINI_DAILY_QUOTA = 1000
|
||
|
||
mock_monitor = MagicMock(spec=OllamaHealthMonitor)
|
||
manager = OllamaFailoverManager(health_monitor=mock_monitor)
|
||
manager._settings = mock_settings
|
||
|
||
with patch("src.core.redis_client.get_redis", return_value=mock_redis):
|
||
await manager._check_gemini_quota()
|
||
|
||
assert len(set_calls) == 1, f"pipeline.set() 應被呼叫一次，實際: {len(set_calls)}"
|
||
call = set_calls[0]
|
||
assert call["nx"] is True, "SET 必須帶 NX=True（只首次設定）"
|
||
assert call["ex"] == 86400, f"TTL 必須 86400s，實際: {call['ex']}"
|
||
assert call["ex"] is not None, "TTL 必須在 SET 時設定，不能分開 EXPIRE（B3 修復驗證）"
|