588 lines
26 KiB
Python
588 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
tests/test_openclaw_qa_routing.py
|
||
OpenClaw Q&A 路由 + 品質守門 unit tests
|
||
(Operation Ollama-First v5.0 — Phase 3, A7 fullstack-engineer)
|
||
|
||
涵蓋:
|
||
- feature flag OPENCLAW_QA_OLLAMA_FIRST=false → 相容舊設定但仍強制 Ollama-first
|
||
- flag=true + 高品質 Ollama 回應 → 直接回 Ollama 結果，不走 Gemini
|
||
- flag=true + 低品質 Ollama 回應 → 升級至 Gemini，並標 fallback_to=openclaw_qa_gemini_fallback
|
||
- flag=true + Ollama 呼叫失敗 → 升級至 Gemini
|
||
- _is_low_quality_response 各規則：空字串 / 長度過短 / 簡體污染 / 拒答 / 流水帳
|
||
|
||
執行:
|
||
pytest tests/test_openclaw_qa_routing.py -v
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
from types import ModuleType, SimpleNamespace
|
||
from typing import Any, Dict, Optional
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
import services.openclaw_strategist_service as svc
|
||
import services.ai_call_logger as logger_mod
|
||
from services.ai_call_logger import _reset_kill_switch
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Fixtures
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def reset_state(monkeypatch):
|
||
"""每個測試重置 logger kill-switch + stub DB 寫入收集 ai_calls 紀錄。"""
|
||
_reset_kill_switch()
|
||
captured = []
|
||
|
||
def fake_write(state):
|
||
captured.append({
|
||
'caller': state.caller,
|
||
'provider': state.provider,
|
||
'model': state.model,
|
||
'status': state.status,
|
||
'fallback_to': state.fallback_to,
|
||
'error': state.error,
|
||
'meta': dict(state.meta),
|
||
'request_id': state.request_id,
|
||
})
|
||
|
||
monkeypatch.setattr(logger_mod, '_write_to_db', fake_write)
|
||
monkeypatch.setenv('AI_CALL_LOGGING_ENABLED', 'true')
|
||
# 預設 flag=true（Ollama-first）
|
||
monkeypatch.delenv('OPENCLAW_QA_OLLAMA_FIRST', raising=False)
|
||
yield captured
|
||
|
||
|
||
def _wait_async(captured, n=1, timeout=2.0):
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
if len(captured) >= n:
|
||
return True
|
||
time.sleep(0.01)
|
||
return False
|
||
|
||
|
||
def _stub_ollama_generate(
|
||
monkeypatch,
|
||
*,
|
||
success: bool = True,
|
||
content: str = '本週 momo 業績成長 12%，建議加碼家電促銷。',
|
||
error: str = 'ConnectionError: connection refused',
|
||
host: str = 'http://34.143.170.20:11434',
|
||
input_tokens: int = 150,
|
||
output_tokens: int = 60,
|
||
):
|
||
"""讓 OpenClaw QA 測試走 OllamaService 介面，而非直打單一 host。"""
|
||
import services.ollama_service as ollama_mod
|
||
|
||
fake_resp = SimpleNamespace(
|
||
success=success,
|
||
content=content if success else '',
|
||
model=svc.OPENCLAW_QA_OLLAMA_MODEL,
|
||
error=None if success else error,
|
||
total_duration=0.12,
|
||
host=host,
|
||
input_tokens=input_tokens if success else 0,
|
||
output_tokens=output_tokens if success else 0,
|
||
)
|
||
|
||
class FakeOllamaService:
|
||
instances = []
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
self.init_args = args
|
||
self.init_kwargs = kwargs
|
||
self.generate_calls = []
|
||
FakeOllamaService.instances.append(self)
|
||
|
||
def generate(self, **kwargs):
|
||
self.generate_calls.append(kwargs)
|
||
return fake_resp
|
||
|
||
monkeypatch.setattr(ollama_mod, 'OllamaService', FakeOllamaService)
|
||
return FakeOllamaService, fake_resp
|
||
|
||
|
||
def _stub_gemini(monkeypatch, *, text: str = "Gemini 備援回覆", raise_error: bool = False):
|
||
"""替 google.generativeai 裝測試替身，避免單測打外部 API。"""
|
||
google_module = ModuleType('google')
|
||
genai_module = ModuleType('google.generativeai')
|
||
|
||
class FakeGenerativeModel:
|
||
def __init__(self, **kwargs):
|
||
self.kwargs = kwargs
|
||
|
||
def generate_content(self, user_prompt, request_options=None):
|
||
if raise_error:
|
||
raise RuntimeError('gemini down')
|
||
return SimpleNamespace(
|
||
text=text,
|
||
usage_metadata=SimpleNamespace(
|
||
prompt_token_count=42,
|
||
candidates_token_count=16,
|
||
),
|
||
)
|
||
|
||
genai_module.configure = lambda api_key=None: None
|
||
genai_module.GenerativeModel = FakeGenerativeModel
|
||
genai_module.types = SimpleNamespace(GenerationConfig=lambda **kwargs: kwargs)
|
||
google_module.generativeai = genai_module
|
||
monkeypatch.setitem(sys.modules, 'google', google_module)
|
||
monkeypatch.setitem(sys.modules, 'google.generativeai', genai_module)
|
||
|
||
|
||
def _stub_nim(monkeypatch, *, text: str = "NIM 備援回覆"):
|
||
"""替 NVIDIA NIM HTTP 呼叫裝測試替身。"""
|
||
class FakeNimResponse:
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
def json(self):
|
||
return {
|
||
"usage": {"prompt_tokens": 21, "completion_tokens": 9},
|
||
"choices": [{"message": {"content": text}}],
|
||
}
|
||
|
||
monkeypatch.setattr(svc.requests, 'post', lambda *args, **kwargs: FakeNimResponse())
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 1. _is_low_quality_response 純函式規則
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestLowQualityRules:
|
||
|
||
def test_empty_string_is_low_quality(self):
|
||
assert svc._is_low_quality_response("") is True
|
||
assert svc._is_low_quality_response(None) is True
|
||
assert svc._is_low_quality_response(" \n ") is True
|
||
|
||
def test_too_short_is_low_quality(self):
|
||
# 長度 < 50 字元 → 低品質
|
||
assert svc._is_low_quality_response("你好，我是 OpenClaw") is True
|
||
|
||
def test_acceptable_response_passes(self):
|
||
good = (
|
||
"本週 momo 業績較上週成長 12%，主要受惠於家電與生活雜貨。\n"
|
||
"建議：持續關注 PChome 競價動態，必要時調整定價策略。\n"
|
||
"預估下週 momo 仍有 5-8% 成長空間。"
|
||
)
|
||
assert svc._is_low_quality_response(good) is False
|
||
|
||
def test_simplified_pollution_detected(self):
|
||
# 句中含 >= 3 個簡體字 hint → 低品質（Qwen 繁中短板核心檢查）
|
||
polluted = (
|
||
"本周业绩比上周增长，您可以关注这个产品的价格变动趋势，"
|
||
"我们建议处理掉滞销库存以提高资产效率"
|
||
)
|
||
assert svc._is_low_quality_response(polluted) is True
|
||
|
||
def test_two_simplified_chars_still_acceptable(self):
|
||
# 只有 2 個簡體 hint 字（边界以下）+ 結構良好 → 仍可接受
|
||
# （避免過度敏感誤殺正常繁體回覆中混入零星簡體字的情境）
|
||
text = (
|
||
"本週 momo 业绩成長明顯，建議持續關注競品動向。\n"
|
||
"重點品類：家電、3C、生活雜貨。\n"
|
||
"下週可加碼促銷檔期。"
|
||
)
|
||
assert svc._is_low_quality_response(text) is False
|
||
|
||
def test_refusal_pattern_detected(self):
|
||
for refusal in ['無法回答', '我不知道', '抱歉，我無法協助']:
|
||
text = f"關於這個問題，{refusal}，請改問其他內容以便我協助您。"
|
||
assert svc._is_low_quality_response(text) is True, f"應被判定為拒答：{refusal}"
|
||
|
||
def test_flowing_text_no_breaks_is_low_quality(self):
|
||
# 200+ 字無換行 → 流水帳
|
||
text = "本週業績整體呈現上升趨勢。" * 20 # ~200+ 字
|
||
assert "\n" not in text
|
||
assert len(text) > 200
|
||
assert svc._is_low_quality_response(text) is True
|
||
|
||
def test_long_text_with_breaks_is_acceptable(self):
|
||
# 200+ 字但有適度斷行 → 結構良好
|
||
text = (
|
||
"本週業績整體呈現上升趨勢，主要驅動類別為家電與生活雜貨大類別。\n"
|
||
"競品動向：PChome 在 3C 類發動大規模補貼戰，預估壓縮我方 3 至 5 個百分點毛利率。\n"
|
||
"蝦皮也在母嬰用品加碼免運券促銷，需密切觀察跟降節奏，避免市占下滑。\n"
|
||
"建議行動：(1) 加碼家電促銷檔期，重點操作大尺寸電視與廚房家電，"
|
||
"(2) 觀察 PChome 補貼是否延續至下週，準備二段反擊方案，"
|
||
"(3) 對價差大於 5% 的 SKU 主動啟動 EA 流程，避免毛利持續流失。"
|
||
)
|
||
assert len(text) > 200
|
||
assert svc._is_low_quality_response(text) is False
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 2. Routing：feature flag = false 時也不可 Gemini-first
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestFlagOff:
|
||
|
||
def test_flag_false_still_routes_to_ollama_first(self, monkeypatch, reset_state):
|
||
"""flag=false → 舊緊急退路已關閉，仍必須先走 Ollama。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'false')
|
||
legacy_called = {'count': 0}
|
||
ollama_called = {'count': 0}
|
||
|
||
def fake_legacy(q, ctx, request_id=None):
|
||
legacy_called['count'] += 1
|
||
return "[legacy gemini reply]"
|
||
|
||
def fake_ollama(q, ctx, rid):
|
||
ollama_called['count'] += 1
|
||
return "[ollama reply with enough length to pass quality gate。這是一段繁體中文策略回覆，避免被判定過短。]"
|
||
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa', fake_legacy)
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa', fake_ollama)
|
||
|
||
result = svc.generate_strategy_response("本週業績如何？")
|
||
assert result.startswith("[ollama reply")
|
||
assert legacy_called['count'] == 0
|
||
assert ollama_called['count'] == 1
|
||
|
||
|
||
class TestOpenClawReportRouting:
|
||
|
||
def test_report_llm_uses_ollama_before_gemini(self, monkeypatch):
|
||
calls = []
|
||
|
||
def fake_ollama(system_prompt, user_prompt, **kwargs):
|
||
calls.append(("ollama", kwargs["caller"]))
|
||
return "Ollama 報告內容足夠完整"
|
||
|
||
def fake_gemini(*args, **kwargs):
|
||
raise AssertionError("Gemini must not run when Ollama succeeds")
|
||
|
||
monkeypatch.setattr(svc, "_call_ollama_strategy", fake_ollama)
|
||
monkeypatch.setattr(svc, "_call_gemini", fake_gemini)
|
||
|
||
result = svc._call_openclaw_llm_ollama_first(
|
||
"system",
|
||
"user",
|
||
temperature=0.3,
|
||
caller="openclaw_weekly",
|
||
)
|
||
|
||
assert result == "Ollama 報告內容足夠完整"
|
||
assert calls == [("ollama", "openclaw_weekly")]
|
||
|
||
def test_report_llm_disables_111_for_long_strategy_reports(self, monkeypatch, reset_state):
|
||
FakeOllamaService, _fake_resp = _stub_ollama_generate(
|
||
monkeypatch,
|
||
content="OpenClaw 報告內容足夠完整，並且只允許 GCP-A/GCP-B 承接長報告。",
|
||
)
|
||
|
||
result = svc._call_ollama_strategy(
|
||
"system",
|
||
"user",
|
||
temperature=0.3,
|
||
caller="openclaw_meta",
|
||
num_predict=3072,
|
||
)
|
||
|
||
assert result.startswith("OpenClaw 報告內容")
|
||
assert FakeOllamaService.instances
|
||
call_kwargs = FakeOllamaService.instances[-1].generate_calls[-1]
|
||
assert call_kwargs["allow_111_fallback"] is False
|
||
|
||
def test_report_llm_gemini_is_suffix_fallback_only(self, monkeypatch):
|
||
monkeypatch.setenv("GEMINI_API_HARD_DISABLED", "false")
|
||
monkeypatch.setenv("GEMINI_FALLBACK_ENABLED", "true")
|
||
monkeypatch.setenv("GEMINI_API_KEY", "test-key")
|
||
calls = []
|
||
|
||
monkeypatch.setattr(svc, "_call_ollama_strategy", lambda *a, **kw: None)
|
||
|
||
def fake_gemini(system_prompt, user_prompt, **kwargs):
|
||
calls.append(("gemini", kwargs["caller"]))
|
||
return "Gemini fallback content"
|
||
|
||
monkeypatch.setattr(svc, "_call_gemini", fake_gemini)
|
||
|
||
result = svc._call_openclaw_llm_ollama_first(
|
||
"system",
|
||
"user",
|
||
temperature=0.3,
|
||
caller="openclaw_monthly",
|
||
)
|
||
|
||
assert result == "Gemini fallback content"
|
||
assert calls == [("gemini", "openclaw_monthly_gemini_fallback")]
|
||
|
||
def test_flag_unset_defaults_to_ollama_first(self, monkeypatch, reset_state):
|
||
"""環境變數完全未設 → 預設 true → 先走 Ollama。"""
|
||
monkeypatch.delenv('OPENCLAW_QA_OLLAMA_FIRST', raising=False)
|
||
legacy_called = {'count': 0}
|
||
ollama_called = {'count': 0}
|
||
|
||
def fake_legacy(q, ctx, request_id=None):
|
||
legacy_called['count'] += 1
|
||
return "[legacy reply]"
|
||
|
||
def fake_ollama(q, ctx, rid):
|
||
ollama_called['count'] += 1
|
||
return (
|
||
"Ollama 主路徑已接手競品分析。建議先檢查近七日價差、銷售跌幅、"
|
||
"PChome 優勢品項與高毛利 SKU，再依 HIGH/MED/LOW 分層處理。"
|
||
"若價差超過 15% 且銷售下滑超過 20%，應優先送人工覆核與 Telegram 告警。"
|
||
)
|
||
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa', fake_legacy)
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa', fake_ollama)
|
||
result = svc.generate_strategy_response("競品分析")
|
||
assert ollama_called['count'] == 1
|
||
assert legacy_called['count'] == 0
|
||
assert result.startswith("Ollama 主路徑已接手")
|
||
|
||
def test_empty_query_short_circuits(self, monkeypatch, reset_state):
|
||
"""空 query 不應觸發任何 LLM 呼叫。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'true')
|
||
legacy_called = {'count': 0}
|
||
ollama_called = {'count': 0}
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa',
|
||
lambda q, ctx, request_id=None: legacy_called.update({'count': legacy_called['count']+1}) or "")
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa',
|
||
lambda q, ctx, rid: ollama_called.update({'count': ollama_called['count']+1}) or "")
|
||
|
||
out = svc.generate_strategy_response("")
|
||
assert "請輸入您的問題" in out
|
||
assert legacy_called['count'] == 0
|
||
assert ollama_called['count'] == 0
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 3. Routing：feature flag = true + Ollama 高/低品質
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestFlagOn:
|
||
|
||
def test_flag_true_high_quality_returns_ollama(self, monkeypatch, reset_state):
|
||
"""flag=true + Ollama 回高品質 → 直接回 Ollama，不走 Gemini。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'true')
|
||
legacy_called = {'count': 0}
|
||
good_reply = (
|
||
"本週 momo 業績成長 12%，主要驅動類別為家電。\n"
|
||
"建議：持續關注 PChome 競價並加碼家電促銷檔期。\n"
|
||
"下週預估仍有 5-8% 成長空間。"
|
||
)
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa', lambda q, ctx, rid: good_reply)
|
||
|
||
def fake_legacy(q, ctx, request_id=None):
|
||
legacy_called['count'] += 1
|
||
return "[gemini fallback]"
|
||
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa', fake_legacy)
|
||
|
||
out = svc.generate_strategy_response("本週業績如何？")
|
||
assert out == good_reply
|
||
assert legacy_called['count'] == 0 # Gemini 沒被呼叫
|
||
|
||
def test_flag_true_low_quality_falls_back_to_gemini(self, monkeypatch, reset_state):
|
||
"""flag=true + Ollama 回低品質（簡體污染）→ fallback Gemini。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'true')
|
||
bad_reply = "本周业绩增长，您可以关注这个产品的价格变动，我们建议处理库存"
|
||
legacy_called = {'count': 0}
|
||
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa', lambda q, ctx, rid: bad_reply)
|
||
|
||
def fake_legacy(q, ctx, request_id=None):
|
||
legacy_called['count'] += 1
|
||
return "[gemini high quality reply]"
|
||
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa', fake_legacy)
|
||
|
||
out = svc.generate_strategy_response("本週業績如何？")
|
||
assert out == "[gemini high quality reply]"
|
||
assert legacy_called['count'] == 1
|
||
|
||
def test_flag_true_ollama_returns_none_falls_back(self, monkeypatch, reset_state):
|
||
"""flag=true + Ollama 呼叫失敗（回 None）→ fallback Gemini。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'true')
|
||
legacy_called = {'count': 0}
|
||
monkeypatch.setattr(svc, '_call_qwen3_qa', lambda q, ctx, rid: None)
|
||
|
||
def fake_legacy(q, ctx, request_id=None):
|
||
legacy_called['count'] += 1
|
||
return "[gemini reply after ollama down]"
|
||
|
||
monkeypatch.setattr(svc, '_legacy_gemini_first_qa', fake_legacy)
|
||
|
||
out = svc.generate_strategy_response("test")
|
||
assert out == "[gemini reply after ollama down]"
|
||
assert legacy_called['count'] == 1
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 4. _call_qwen3_qa: ai_call_logger 整合 + fallback_to 標記
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestCallQwen3Telemetry:
|
||
|
||
def test_qwen3_logs_ok_status_on_success(self, monkeypatch, reset_state):
|
||
"""高品質回應 → ai_calls 應記 status=ok, caller=openclaw_qa, provider=gcp_ollama"""
|
||
captured = reset_state
|
||
|
||
fake_service, _fake_resp = _stub_ollama_generate(monkeypatch)
|
||
|
||
result = svc._call_qwen3_qa("本週業績？", None, "qa-test123")
|
||
assert result is not None
|
||
assert "業績成長" in result
|
||
assert fake_service.instances
|
||
generate_kwargs = fake_service.instances[0].generate_calls[0]
|
||
assert generate_kwargs['model'] == svc.OPENCLAW_QA_OLLAMA_MODEL
|
||
assert generate_kwargs['system_prompt'] == svc.QWEN3_TC_SYSTEM_PROMPT
|
||
|
||
assert _wait_async(captured, 1)
|
||
assert len(captured) == 1
|
||
rec = captured[0]
|
||
assert rec['caller'] == 'openclaw_qa'
|
||
assert rec['provider'] == 'gcp_ollama'
|
||
assert rec['model'] == svc.OPENCLAW_QA_OLLAMA_MODEL
|
||
assert rec['status'] == 'ok'
|
||
assert rec['fallback_to'] is None
|
||
assert rec['meta'].get('flag') == 'OPENCLAW_QA_OLLAMA_FIRST'
|
||
assert rec['meta'].get('route') == 'ollama_first'
|
||
assert rec['meta'].get('host') == 'http://34.143.170.20:11434'
|
||
assert rec['meta'].get('host_label') == 'GCP-SSD'
|
||
assert rec['request_id'] == "qa-test123"
|
||
|
||
def test_qwen3_logs_actual_secondary_provider_after_retry(self, monkeypatch, reset_state):
|
||
"""OllamaService 若落到 GCP-B，ai_calls.provider 必須寫 ollama_secondary。"""
|
||
captured = reset_state
|
||
_stub_ollama_generate(
|
||
monkeypatch,
|
||
host='http://34.21.145.224:11434',
|
||
)
|
||
|
||
result = svc._call_qwen3_qa("本週業績？", None, "qa-secondary")
|
||
assert result is not None
|
||
|
||
assert _wait_async(captured, 1)
|
||
rec = captured[0]
|
||
assert rec['provider'] == 'ollama_secondary'
|
||
assert rec['meta'].get('host_label') == 'GCP-SSD-2'
|
||
|
||
def test_qwen3_logs_fallback_on_exception(self, monkeypatch, reset_state):
|
||
"""Ollama 連線失敗 → ai_calls 應記 fallback_to=openclaw_qa_gemini_fallback + status=fallback"""
|
||
captured = reset_state
|
||
|
||
_stub_ollama_generate(monkeypatch, success=False)
|
||
|
||
result = svc._call_qwen3_qa("test", None, "qa-fail123")
|
||
assert result is None
|
||
|
||
assert _wait_async(captured, 1)
|
||
rec = captured[0]
|
||
assert rec['status'] == 'fallback'
|
||
assert rec['fallback_to'] == 'openclaw_qa_gemini_fallback'
|
||
assert rec['error'] is not None
|
||
assert 'ConnectionError' in rec['error']
|
||
|
||
def test_qwen3_logs_fallback_on_empty_response(self, monkeypatch, reset_state):
|
||
"""Ollama 回空 response → 視為 empty_response，標 fallback。"""
|
||
captured = reset_state
|
||
|
||
_stub_ollama_generate(
|
||
monkeypatch,
|
||
content='',
|
||
input_tokens=100,
|
||
output_tokens=0,
|
||
)
|
||
|
||
result = svc._call_qwen3_qa("test", None, "qa-empty")
|
||
assert result is None
|
||
|
||
assert _wait_async(captured, 1)
|
||
rec = captured[0]
|
||
assert rec['status'] == 'fallback'
|
||
assert rec['fallback_to'] == 'openclaw_qa_gemini_fallback'
|
||
assert rec['error'] == 'empty_response'
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 5. Gemini / NIM 備援遙測：Gemini 不可被誤記為 openclaw_qa 主路徑
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestLegacyFallbackTelemetry:
|
||
|
||
def test_gemini_backup_uses_dedicated_caller(self, monkeypatch, reset_state):
|
||
"""Ollama 後的 Gemini 備援應記 openclaw_qa_gemini_fallback，不污染 openclaw_qa。"""
|
||
captured = reset_state
|
||
monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false')
|
||
monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true')
|
||
monkeypatch.setenv('GEMINI_API_KEY', 'test-gemini-key')
|
||
monkeypatch.setattr(svc, 'NVIDIA_API_KEY', '')
|
||
_stub_gemini(monkeypatch, text="Gemini 備援：請先檢查近七日業績與競品價差。")
|
||
|
||
out = svc._legacy_gemini_first_qa("本週業績如何？", None, request_id="qa-gemini")
|
||
|
||
assert out.startswith("Gemini 備援")
|
||
assert _wait_async(captured, 1)
|
||
assert len(captured) == 1
|
||
rec = captured[0]
|
||
assert rec['caller'] == 'openclaw_qa_gemini_fallback'
|
||
assert rec['provider'] == 'gemini'
|
||
assert rec['status'] == 'ok'
|
||
assert rec['fallback_to'] is None
|
||
assert not any(
|
||
r['caller'] == 'openclaw_qa' and r['provider'] == 'gemini'
|
||
for r in captured
|
||
)
|
||
|
||
def test_gemini_backup_failure_falls_to_standard_nim_caller(self, monkeypatch, reset_state):
|
||
"""Gemini 備援失敗後，NIM 應記 openclaw_qa_nim，而非 fallback_fallback_nim。"""
|
||
captured = reset_state
|
||
monkeypatch.setenv('GEMINI_API_HARD_DISABLED', 'false')
|
||
monkeypatch.setenv('GEMINI_FALLBACK_ENABLED', 'true')
|
||
monkeypatch.setenv('GEMINI_API_KEY', 'test-gemini-key')
|
||
monkeypatch.setattr(svc, 'NVIDIA_API_KEY', 'test-nim-key')
|
||
_stub_gemini(monkeypatch, raise_error=True)
|
||
_stub_nim(monkeypatch, text="NIM 備援：請改看 /daily 與 /threats。")
|
||
|
||
out = svc._legacy_gemini_first_qa("本週業績如何？", None, request_id="qa-nim")
|
||
|
||
assert out.startswith("NIM 備援")
|
||
assert _wait_async(captured, 2)
|
||
assert [r['caller'] for r in captured] == [
|
||
'openclaw_qa_gemini_fallback',
|
||
'openclaw_qa_nim',
|
||
]
|
||
gemini_rec, nim_rec = captured
|
||
assert gemini_rec['provider'] == 'gemini'
|
||
assert gemini_rec['status'] == 'fallback'
|
||
assert gemini_rec['fallback_to'] == 'openclaw_qa_nim'
|
||
assert nim_rec['provider'] == 'nim'
|
||
assert nim_rec['status'] == 'ok'
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 6. 環境變數讀取即時性（runtime toggle）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestRuntimeToggle:
|
||
|
||
def test_flag_helper_reads_env_each_call(self, monkeypatch):
|
||
"""_qa_ollama_first_enabled() 應每次重讀 env，允許 runtime 灰度切換。"""
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'false')
|
||
assert svc._qa_ollama_first_enabled() is False
|
||
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', 'true')
|
||
assert svc._qa_ollama_first_enabled() is True
|
||
|
||
# 各種真值字串
|
||
for v in ('TRUE', 'True', '1', 'yes', 'on'):
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', v)
|
||
assert svc._qa_ollama_first_enabled() is True, f"應視為 true: {v!r}"
|
||
|
||
for v in ('false', '0', 'no', 'off', '', 'foo'):
|
||
monkeypatch.setenv('OPENCLAW_QA_OLLAMA_FIRST', v)
|
||
assert svc._qa_ollama_first_enabled() is False, f"應視為 false: {v!r}"
|