test(ai-router): TestLocalFallbackChain — require_local 隱私邊界驗證 (P0)
Some checks failed
CD Pipeline / build-and-deploy (push) Failing after 43s
Some checks failed
CD Pipeline / build-and-deploy (push) Failing after 43s
新增兩個測試:cloud provider 被跳過 + 全失敗回傳 local_providers_unavailable。 實作邏輯已存在於 AIRouterExecutor.execute()(2026-04-04 ogt Phase 25 P0)。 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -38,3 +38,101 @@ class TestNemotronPerTaskTimeout:
|
||||
|
||||
assert result.success is True
|
||||
mock_nvidia.tool_call.assert_called_once()
|
||||
|
||||
|
||||
class TestLocalFallbackChain:
|
||||
"""require_local=True 時只走 local chain,全部失敗 → REJECT,不觸碰雲端"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_require_local_skips_cloud_providers(self):
|
||||
"""require_local=True 時,cloud provider 不被呼叫"""
|
||||
import os
|
||||
from src.services.ai_router import AIRouterExecutor, AIProviderRegistry
|
||||
from src.services.ai_providers.interfaces import AIResult
|
||||
|
||||
registry = AIProviderRegistry()
|
||||
|
||||
# Mock: Ollama 成功
|
||||
mock_ollama = AsyncMock()
|
||||
mock_ollama.name = "ollama"
|
||||
mock_ollama.privacy_level = "local"
|
||||
mock_ollama.is_enabled = True
|
||||
mock_ollama.capabilities = {"rca", "chat"}
|
||||
mock_ollama.analyze = AsyncMock(return_value=AIResult(
|
||||
raw_response="本地診斷結果",
|
||||
success=True,
|
||||
provider="ollama",
|
||||
))
|
||||
mock_ollama.health_check = AsyncMock(return_value=True)
|
||||
|
||||
# Mock: Gemini(不應該被呼叫)
|
||||
mock_gemini = AsyncMock()
|
||||
mock_gemini.name = "gemini"
|
||||
mock_gemini.privacy_level = "cloud"
|
||||
mock_gemini.is_enabled = True
|
||||
mock_gemini.analyze = AsyncMock(return_value=AIResult(
|
||||
raw_response="雲端結果",
|
||||
success=True,
|
||||
provider="gemini",
|
||||
))
|
||||
|
||||
registry._providers = {
|
||||
"ollama": mock_ollama,
|
||||
"gemini": mock_gemini,
|
||||
}
|
||||
|
||||
executor = AIRouterExecutor(registry)
|
||||
|
||||
# 暫時關閉 MOCK_MODE,測試真實執行路徑
|
||||
with patch("src.services.ai_router._settings") as mock_settings:
|
||||
mock_settings.MOCK_MODE = False
|
||||
result = await executor.execute(
|
||||
prompt="診斷這個問題",
|
||||
provider_order=["ollama", "gemini"],
|
||||
require_local=True,
|
||||
)
|
||||
|
||||
assert result.success is True
|
||||
assert result.provider == "ollama"
|
||||
mock_gemini.analyze.assert_not_called()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_require_local_all_fail_returns_reject(self):
|
||||
"""require_local=True 且所有 local provider 失敗 → 回傳明確錯誤"""
|
||||
import os
|
||||
from src.services.ai_router import AIRouterExecutor, AIProviderRegistry
|
||||
from src.services.ai_providers.interfaces import AIResult
|
||||
|
||||
registry = AIProviderRegistry()
|
||||
|
||||
# Mock: Ollama 失敗
|
||||
mock_ollama = AsyncMock()
|
||||
mock_ollama.name = "ollama"
|
||||
mock_ollama.privacy_level = "local"
|
||||
mock_ollama.is_enabled = True
|
||||
mock_ollama.capabilities = {"rca", "chat"}
|
||||
mock_ollama.analyze = AsyncMock(return_value=AIResult(
|
||||
raw_response="",
|
||||
success=False,
|
||||
provider="ollama",
|
||||
error="timeout",
|
||||
))
|
||||
mock_ollama.health_check = AsyncMock(return_value=False)
|
||||
|
||||
registry._providers = {
|
||||
"ollama": mock_ollama,
|
||||
}
|
||||
|
||||
executor = AIRouterExecutor(registry)
|
||||
|
||||
# 暫時關閉 MOCK_MODE + 讓 telegram import 失敗(不影響主流程)
|
||||
with patch("src.services.ai_router._settings") as mock_settings:
|
||||
mock_settings.MOCK_MODE = False
|
||||
result = await executor.execute(
|
||||
prompt="診斷這個問題",
|
||||
provider_order=["ollama"],
|
||||
require_local=True,
|
||||
)
|
||||
|
||||
assert result.success is False
|
||||
assert result.error == "local_providers_unavailable"
|
||||
|
||||
Reference in New Issue
Block a user