diff --git a/apps/api/tests/test_p0_diagnose_routing.py b/apps/api/tests/test_p0_diagnose_routing.py index 2b10424b..fd0cbca1 100644 --- a/apps/api/tests/test_p0_diagnose_routing.py +++ b/apps/api/tests/test_p0_diagnose_routing.py @@ -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"