#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ tests/test_code_review_claude_routing.py CodeReviewPipeline._openclaw_assess Ollama-first / cloud fallback 路由測試 驗收項目： - Ollama 成功 → 不呼叫 Claude / Gemini / Elephant - Ollama 失敗 + flag=true + Claude 可用 → 走 Claude - Ollama 失敗 + Claude 不可用/失敗 → Gemini 只作備援 - Ollama + Gemini 都失敗 → 最終 Elephant 接手 """ from __future__ import annotations import importlib import os import sys import types from unittest.mock import MagicMock import pytest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # ───────────────────────────────────────────────────────────────────────────── # 共用工具 # ───────────────────────────────────────────────────────────────────────────── @pytest.fixture(autouse=True) def _disable_real_code_review_ollama_preflight(monkeypatch): monkeypatch.setenv("CODE_REVIEW_OLLAMA_HOST_PREFLIGHT_ENABLED", "false") def _reload_pipeline(): """重新載入 pipeline 模組（讓 module-level CODE_REVIEW_USE_CLAUDE flag 即時生效）""" import services.code_review_pipeline_service as svc_mod importlib.reload(svc_mod) return svc_mod def _make_pipeline(svc_mod, commit="abc12345"): return svc_mod.CodeReviewPipeline( commit_sha=commit, changed_files=["services/foo.py"], branch="main", deploy_type="sync", ) def _stub_anthropic(monkeypatch, svc_mod, *, available: bool, success: bool = True, content: str = "CLAUDE-RESULT", error: str = None): """注入假的 anthropic_service 全域單例到 services.anthropic_service。 目標：避免 monkeypatch sys.modules 失敗（pipeline 內是 from services.anthropic_service import anthropic_service） """ fake_resp = MagicMock() fake_resp.success = success fake_resp.content = content if success else "" fake_resp.input_tokens = 200 fake_resp.output_tokens = 100 fake_resp.cache_creation_tokens = 50 fake_resp.cache_read_tokens = 150 fake_resp.cache_hit = True fake_resp.error = error fake_svc = MagicMock() fake_svc.is_available.return_value = available fake_svc.generate.return_value = fake_resp # 動態造一個假 module 並寫入 sys.modules（覆蓋既有 import 結果） fake_module = types.ModuleType('services.anthropic_service') fake_module.anthropic_service = fake_svc monkeypatch.setitem(sys.modules, 'services.anthropic_service', fake_module) return fake_svc def _stub_logger(monkeypatch): """避免 log_ai_call 真寫 DB""" import services.ai_call_logger as logger_mod monkeypatch.setattr(logger_mod, '_write_to_db', lambda state: None) monkeypatch.setenv('AI_CALL_LOGGING_ENABLED', 'false') def _capture_ai_call_states(monkeypatch): """攔截 ai_call_logger 寫入，保留 caller/provider/fallback_to 供路由斷言。""" import services.ai_call_logger as logger_mod states = [] def _capture(state): states.append({ "caller": state.caller, "provider": state.provider, "model": state.model, "status": state.status, "fallback_to": state.fallback_to, "error": state.error, }) monkeypatch.setattr(logger_mod, '_async_write', _capture) monkeypatch.setenv('AI_CALL_LOGGING_ENABLED', 'true') return states def _stub_ollama(monkeypatch, *, success: bool = True, content: str = "OLLAMA-RESULT", error: str = "all hosts failed"): import services.ollama_service as ollama_mod fake_resp = MagicMock() fake_resp.success = success fake_resp.content = content if success else "" fake_resp.model = "qwen3:14b" fake_resp.error = None if success else error fake_resp.host = "http://34.87.90.216:11434" fake_resp.input_tokens = 30 if success else 0 fake_resp.output_tokens = 12 if success else 0 fake_service = MagicMock() fake_service.generate.return_value = fake_resp fake_cls = MagicMock(return_value=fake_service) monkeypatch.setattr(ollama_mod, "OllamaService", fake_cls) return fake_cls, fake_service def _stub_gemini_and_elephant(monkeypatch, *, gemini_text: str = "GEMINI-RESULT", elephant_text: str = "ELEPHANT-RESULT", gemini_works: bool = True): """攔截 _openclaw_assess 內的 import google.generativeai / elephant_service。 pipeline 內是 lazy import，所以注入到 sys.modules 即可生效。 """ # 1) Fake google.generativeai fake_genai = types.ModuleType('google.generativeai') fake_types = types.SimpleNamespace(GenerationConfig=lambda **kw: MagicMock()) fake_genai.types = fake_types fake_genai.configure = lambda **kw: None if gemini_works: fake_resp = MagicMock() fake_resp.text = gemini_text fake_resp.usage_metadata = MagicMock(prompt_token_count=10, candidates_token_count=5) fake_model = MagicMock() fake_model.generate_content.return_value = fake_resp fake_genai.GenerativeModel = MagicMock(return_value=fake_model) else: fake_genai.GenerativeModel = MagicMock(side_effect=RuntimeError("gemini broken")) # google.generativeai 是子模組；注入它和父模組 fake_google = types.ModuleType('google') fake_google.generativeai = fake_genai monkeypatch.setitem(sys.modules, 'google', fake_google) monkeypatch.setitem(sys.modules, 'google.generativeai', fake_genai) # 2) Fake elephant_service fake_eresp = MagicMock(success=True, content=elephant_text, input_tokens=20, output_tokens=10, error=None) fake_elephant = MagicMock() fake_elephant.generate.return_value = fake_eresp fake_eservice_mod = types.ModuleType('services.elephant_service') fake_eservice_mod.elephant_service = fake_elephant monkeypatch.setitem(sys.modules, 'services.elephant_service', fake_eservice_mod) return fake_genai, fake_elephant # ───────────────────────────────────────────────────────────────────────────── # Tests # ───────────────────────────────────────────────────────────────────────────── def test_default_uses_ollama_first(monkeypatch): """CODE_REVIEW_USE_CLAUDE=false（預設）→ Ollama 成功時 Gemini 不被呼叫""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _ollama_cls, fake_ollama = _stub_ollama(monkeypatch) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "OLLAMA-RESULT" fake_ollama.generate.assert_called_once() fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_code_review_ollama_defaults_use_fast_local_model(monkeypatch): """未設定 env 時，Code Review 用 coder 7B 與短 timeout，避免 120s*3 後才備援。""" for key in ( "CODE_REVIEW_OLLAMA_MODEL", "OPENCLAW_OLLAMA_MODEL", "CODE_REVIEW_OLLAMA_TIMEOUT", "CODE_REVIEW_OLLAMA_SECONDARY_MODEL", "CODE_REVIEW_OLLAMA_SECONDARY_TIMEOUT", "CODE_REVIEW_OLLAMA_FALLBACK_MODEL", "CODE_REVIEW_OLLAMA_FALLBACK_TIMEOUT", "CODE_REVIEW_OLLAMA_NUM_PREDICT", "CODE_REVIEW_OLLAMA_KEEP_ALIVE", "CODE_REVIEW_ALLOW_111_FALLBACK", "CODE_REVIEW_HERMES_TIMEOUT", "CODE_REVIEW_HERMES_PRIMARY_MODEL", "CODE_REVIEW_HERMES_PRIMARY_TIMEOUT", "CODE_REVIEW_HERMES_SECONDARY_MODEL", "CODE_REVIEW_HERMES_SECONDARY_TIMEOUT", "CODE_REVIEW_HERMES_FALLBACK_MODEL", "CODE_REVIEW_HERMES_FALLBACK_TIMEOUT", "CODE_REVIEW_HERMES_NUM_PREDICT", "CODE_REVIEW_HERMES_MAX_FILES", "CODE_REVIEW_HERMES_MAX_CHARS", "CODE_REVIEW_HERMES_LLM_SCAN_ENABLED", ): monkeypatch.delenv(key, raising=False) svc_mod = _reload_pipeline() assert svc_mod.CODE_REVIEW_OLLAMA_MODEL == "qwen2.5-coder:7b" assert svc_mod.CODE_REVIEW_OLLAMA_TIMEOUT == 15 assert svc_mod.CODE_REVIEW_OLLAMA_SECONDARY_MODEL == "gemma3:4b" assert svc_mod.CODE_REVIEW_OLLAMA_SECONDARY_TIMEOUT == 25 assert svc_mod.CODE_REVIEW_OLLAMA_FALLBACK_MODEL == "hermes3:latest" assert svc_mod.CODE_REVIEW_OLLAMA_FALLBACK_TIMEOUT == 20 assert svc_mod.CODE_REVIEW_OLLAMA_NUM_PREDICT == 384 assert svc_mod.CODE_REVIEW_OLLAMA_KEEP_ALIVE == "5m" assert svc_mod.CODE_REVIEW_ALLOW_111_FALLBACK is False assert svc_mod.CODE_REVIEW_HERMES_TIMEOUT == 35 assert svc_mod.CODE_REVIEW_HERMES_PRIMARY_MODEL == "qwen2.5-coder:7b" assert svc_mod.CODE_REVIEW_HERMES_PRIMARY_TIMEOUT == 15 assert svc_mod.CODE_REVIEW_HERMES_SECONDARY_MODEL == "gemma3:4b" assert svc_mod.CODE_REVIEW_HERMES_SECONDARY_TIMEOUT == 45 assert svc_mod.CODE_REVIEW_HERMES_FALLBACK_MODEL == "hermes3:latest" assert svc_mod.CODE_REVIEW_HERMES_FALLBACK_TIMEOUT == 20 assert svc_mod.CODE_REVIEW_HERMES_NUM_PREDICT == 384 assert svc_mod.CODE_REVIEW_HERMES_MAX_FILES == 2 assert svc_mod.CODE_REVIEW_HERMES_MAX_CHARS == 900 assert svc_mod.CODE_REVIEW_HERMES_LLM_SCAN_ENABLED is False def test_openclaw_uses_secondary_local_model_before_gemini(monkeypatch): """GCP-A 失敗時先試 GCP-B 本地模型，成功就不得呼叫 Gemini。""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() import services.ollama_service as ollama_mod calls = [] class FakeResp: def __init__(self, *, success, content, host, model, error=None): self.success = success self.content = content self.host = host self.model = model self.error = error self.input_tokens = 20 if success else 0 self.output_tokens = 8 if success else 0 class FakeOllama: def __init__(self, host=None, model=None): self.host = host self.model = model def generate(self, **kwargs): calls.append({ "host": self.host, "model": kwargs["model"], "timeout": kwargs["timeout"], }) if len(calls) == 1: return FakeResp( success=False, content="", host=self.host, model=kwargs["model"], error="primary timeout", ) return FakeResp( success=True, content="SECONDARY-LOCAL", host=self.host, model=kwargs["model"], ) monkeypatch.setattr(ollama_mod, "OllamaService", FakeOllama) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "SECONDARY-LOCAL" assert [call["model"] for call in calls] == ["qwen2.5-coder:7b", "gemma3:4b"] assert calls[0]["timeout"] == 15 assert calls[1]["timeout"] == 25 fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_openclaw_preflight_skips_dead_primary_before_generate(monkeypatch): """GCP-A preflight 不通時，OpenClaw 應直接進 GCP-B，避免等 primary generate timeout。""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('CODE_REVIEW_OLLAMA_HOST_PREFLIGHT_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() import services.ollama_service as ollama_mod monkeypatch.setattr( svc_mod, "_code_review_ollama_host_reachable", lambda host: host == ollama_mod.OLLAMA_HOST_SECONDARY, ) calls = [] class FakeResp: success = True content = "SECONDARY-PREFLIGHT-OK" error = None input_tokens = 20 output_tokens = 8 def __init__(self, *, host, model): self.host = host self.model = model class FakeOllama: def __init__(self, host=None, model=None): self.host = host self.model = model def generate(self, **kwargs): calls.append({ "host": self.host, "model": kwargs["model"], "timeout": kwargs["timeout"], }) return FakeResp(host=self.host, model=kwargs["model"]) monkeypatch.setattr(ollama_mod, "OllamaService", FakeOllama) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "SECONDARY-PREFLIGHT-OK" assert calls == [{ "host": ollama_mod.OLLAMA_HOST_SECONDARY, "model": "gemma3:4b", "timeout": 25, }] fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_openclaw_preflight_skips_primary_when_model_missing(monkeypatch): """GCP-A 可達但缺 primary model 時，直接進 GCP-B 的可用模型。""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('CODE_REVIEW_OLLAMA_HOST_PREFLIGHT_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() import services.ollama_service as ollama_mod monkeypatch.setattr(svc_mod, "_code_review_ollama_host_reachable", lambda host: True) monkeypatch.setattr( svc_mod, "_code_review_ollama_model_available", lambda host, model: not ( host == ollama_mod.OLLAMA_HOST_PRIMARY and model == "qwen2.5-coder:7b" ), ) calls = [] class FakeResp: success = True content = "SECONDARY-MODEL-OK" error = None input_tokens = 20 output_tokens = 8 def __init__(self, *, host, model): self.host = host self.model = model class FakeOllama: def __init__(self, host=None, model=None): self.host = host self.model = model def generate(self, **kwargs): calls.append({ "host": self.host, "model": kwargs["model"], "timeout": kwargs["timeout"], }) return FakeResp(host=self.host, model=kwargs["model"]) monkeypatch.setattr(ollama_mod, "OllamaService", FakeOllama) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "SECONDARY-MODEL-OK" assert calls == [{ "host": ollama_mod.OLLAMA_HOST_SECONDARY, "model": "gemma3:4b", "timeout": 25, }] fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_openclaw_skips_111_and_cloud_by_default_when_gcp_pair_fails(monkeypatch): """GCP-A/B 都失敗時，預設不把 Code Review 重分析丟給 111 或雲端。""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.delenv('CODE_REVIEW_ALLOW_111_FALLBACK', raising=False) monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'true') monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'false') monkeypatch.delenv('GEMINI_API_KEY', raising=False) _stub_logger(monkeypatch) svc_mod = _reload_pipeline() import services.ollama_service as ollama_mod calls = [] class FakeResp: success = False content = "" error = "timeout" input_tokens = 0 output_tokens = 0 def __init__(self, *, host, model): self.host = host self.model = model class FakeOllama: def __init__(self, host=None, model=None): self.host = host self.model = model def generate(self, **kwargs): calls.append({ "host": self.host, "model": kwargs["model"], "timeout": kwargs["timeout"], }) return FakeResp(host=self.host, model=kwargs["model"]) monkeypatch.setattr(ollama_mod, "OllamaService", FakeOllama) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert "本地降級報告" in result assert [call["model"] for call in calls] == ["qwen2.5-coder:7b", "gemma3:4b"] assert not any("192.168.0.111" in call["host"] for call in calls) fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_hermes_scan_uses_compact_prompt_and_short_timeout(monkeypatch): """Hermes scan 只送 compact snippet，避免大檔讓三主機各卡 120 秒。""" monkeypatch.setenv("CODE_REVIEW_HERMES_LLM_SCAN_ENABLED", "true") monkeypatch.setenv("CODE_REVIEW_HERMES_TIMEOUT", "7") monkeypatch.setenv("CODE_REVIEW_HERMES_MAX_FILES", "2") monkeypatch.setenv("CODE_REVIEW_HERMES_MAX_CHARS", "20") _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _ollama_cls, fake_ollama = _stub_ollama(monkeypatch, content="[]") pipeline = _make_pipeline(svc_mod) result = pipeline._hermes_scan({ "services/a.py": "A" * 80, "services/b.py": "B" * 80, "services/c.py": "C" * 80, }) assert result == [] kwargs = fake_ollama.generate.call_args.kwargs assert kwargs["model"] == "qwen2.5-coder:7b" assert kwargs["timeout"] == 7 assert kwargs["options"] == {"num_predict": 384} prompt = kwargs["prompt"] assert "services/a.py" in prompt assert "services/b.py" in prompt assert "services/c.py" not in prompt assert "截斷" in prompt def test_hermes_scan_defaults_to_static_scan_without_ollama(monkeypatch): """預設不打 LLM，避免部署後 Hermes scan 卡三段 Ollama timeout。""" monkeypatch.delenv("CODE_REVIEW_HERMES_LLM_SCAN_ENABLED", raising=False) _stub_logger(monkeypatch) svc_mod = _reload_pipeline() pipeline = _make_pipeline(svc_mod) findings = pipeline._hermes_scan({ "services/foo.py": ( "import requests\n" "def risky(payload):\n" " requests.get('https://example.com')\n" " return eval(payload)\n" ) }) descriptions = {finding["description"] for finding in findings} assert "HTTP request 未設定 timeout，可能拖住 worker" in descriptions assert "使用 eval() 有安全風險" in descriptions assert pipeline.state["severity_summary"]["high"] == 1 assert pipeline.state["severity_summary"]["medium"] == 1 def test_static_scan_accepts_multiline_requests_timeout(monkeypatch): """多行 requests 呼叫已帶 timeout 時，不應被誤報為未設定 timeout。""" monkeypatch.delenv("CODE_REVIEW_HERMES_LLM_SCAN_ENABLED", raising=False) _stub_logger(monkeypatch) svc_mod = _reload_pipeline() pipeline = _make_pipeline(svc_mod) findings = pipeline._hermes_scan({ "services/foo.py": ( "import requests\n" "def safe(url, timeout_s):\n" " return requests.get(\n" " url,\n" " timeout=timeout_s,\n" " )\n" ) }) assert not any( finding["description"] == "HTTP request 未設定 timeout，可能拖住 worker" for finding in findings ) assert pipeline.state["severity_summary"]["medium"] == 0 def test_flag_true_still_uses_ollama_before_claude(monkeypatch): """flag=true 也不得跳過 Ollama；Ollama 成功時 Claude/Gemini 都不呼叫""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True, success=True, content="CLAUDE-RESULT") fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[{"severity": "HIGH", "file": "services/foo.py", "description": "x", "type": "bug"}], ) assert result == "OLLAMA-RESULT" fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() def test_ollama_failure_flag_true_uses_claude_backup(monkeypatch): """Ollama 失敗 + flag=true + Claude 可用 → Claude 才接手""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch, success=False) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True, success=True, content="CLAUDE-RESULT") fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "CLAUDE-RESULT" fake_claude.generate.assert_called_once() fake_genai.GenerativeModel.assert_not_called() fake_elephant.generate.assert_not_called() call_kwargs = fake_claude.generate.call_args.kwargs assert call_kwargs['cache_system'] is True assert call_kwargs['temperature'] == 0.2 assert call_kwargs['model'] == 'claude-opus-4-7' def test_ollama_failure_flag_false_uses_gemini_backup(monkeypatch): """Ollama 失敗且 Claude flag=false → Gemini 才作備援""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false') monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch, success=False) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "GEMINI-RESULT" fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_called_once() fake_elephant.generate.assert_not_called() def test_gemini_backup_uses_dedicated_caller_in_telemetry(monkeypatch): """Ollama 失敗後的 Gemini 必須記為 code_review_openclaw_gemini。""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'false') monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false') monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') captured = _capture_ai_call_states(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch, success=False) _stub_anthropic(monkeypatch, svc_mod, available=True) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "GEMINI-RESULT" fake_genai.GenerativeModel.assert_called_once() fake_elephant.generate.assert_not_called() assert any( state["caller"] == "code_review_openclaw" and state["provider"] in {"gcp_ollama", "ollama_secondary", "ollama_111"} and state["status"] == "fallback" and state["fallback_to"] == "code_review_openclaw_gemini" for state in captured ) assert any( state["caller"] == "code_review_openclaw_gemini" and state["provider"] == "gemini" and state["status"] == "ok" for state in captured ) assert not any( state["caller"] == "code_review_openclaw" and state["provider"] == "gemini" for state in captured ) def test_ollama_failure_claude_unavailable_uses_gemini_backup(monkeypatch): """Ollama 失敗 + flag=true 但 Claude unavailable → Gemini 才作備援""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'true') monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false') monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch, success=False) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=False) fake_genai, fake_elephant = _stub_gemini_and_elephant(monkeypatch) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "GEMINI-RESULT" fake_claude.generate.assert_not_called() fake_genai.GenerativeModel.assert_called_once() fake_elephant.generate.assert_not_called() def test_full_fallback_chain_after_ollama_failure(monkeypatch): """Ollama 失敗 + Claude 失敗 + Gemini 失敗 → 最終 Elephant 接手""" monkeypatch.setenv('CODE_REVIEW_USE_CLAUDE', 'true') monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false') monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true') monkeypatch.setenv('GEMINI_API_KEY', 'test-key') _stub_logger(monkeypatch) svc_mod = _reload_pipeline() _stub_ollama(monkeypatch, success=False) fake_claude = _stub_anthropic(monkeypatch, svc_mod, available=True, success=False, error="claude down") fake_genai, fake_elephant = _stub_gemini_and_elephant( monkeypatch, gemini_works=False, ) pipeline = _make_pipeline(svc_mod) result = pipeline._openclaw_assess( files={"services/foo.py": "def x(): pass"}, findings=[], ) assert result == "ELEPHANT-RESULT" fake_claude.generate.assert_called_once() fake_elephant.generate.assert_called_once()