749 lines
28 KiB
Python
749 lines
28 KiB
Python
#!/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 "最後錯誤" not in result
|
||
assert "all hosts failed" not 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()
|