285 lines
11 KiB
Python
285 lines
11 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_providers.interfaces import AIResult
|
||
from src.services.ai_providers.ollama import OllamaLocalProvider
|
||
from src.services.ai_router import (
|
||
AIProviderRegistry,
|
||
AIRouterExecutor,
|
||
reset_ai_router,
|
||
)
|
||
|
||
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 selection preflight is read-only; generation owns atomic quota
|
||
# =============================================================================
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_gemini_quota_concurrent_no_overshoot():
|
||
"""Concurrent route checks read unified stats and never reserve paid quota."""
|
||
from src.services.ollama_failover_manager import OllamaFailoverManager
|
||
from src.services.ollama_health_monitor import OllamaHealthMonitor
|
||
|
||
mock_monitor = MagicMock(spec=OllamaHealthMonitor)
|
||
manager = OllamaFailoverManager(health_monitor=mock_monitor)
|
||
limiter = MagicMock()
|
||
limiter.get_usage_stats = AsyncMock(return_value={
|
||
"daily_requests": {"current": 4, "limit": 5},
|
||
"daily_tokens": {"current": 100, "reserved": 0, "limit": 1000},
|
||
"daily_cost_usd": {"current": 0.01, "reserved": 0.0, "limit": 1.0},
|
||
"cost_exceeded": False,
|
||
})
|
||
|
||
with patch("src.services.ai_rate_limiter.get_ai_rate_limiter", return_value=limiter):
|
||
results = await asyncio.gather(*[manager._check_gemini_quota() for _ in range(5)])
|
||
|
||
assert all(results)
|
||
assert limiter.get_usage_stats.await_count == 5
|
||
assert not hasattr(limiter, "reserve_generation") or limiter.reserve_generation.call_count == 0
|
||
|
||
|
||
def test_gemini_reservation_receipt_and_identity_are_durable():
|
||
"""The atomic reservation must not reopen a paid call through TTL expiry."""
|
||
from src.services.ai_rate_limiter import _RESERVE_GENERATION_LUA
|
||
|
||
assert "redis.call('HSET', KEYS[9]" in _RESERVE_GENERATION_LUA
|
||
assert "redis.call('EXPIRE', KEYS[9]" not in _RESERVE_GENERATION_LUA
|
||
assert "redis.call('EXPIRE', KEYS[10]" not in _RESERVE_GENERATION_LUA
|
||
assert "redis.call('INCR', KEYS[1])" in _RESERVE_GENERATION_LUA
|
||
assert "redis.call('INCR', KEYS[2])" in _RESERVE_GENERATION_LUA
|