70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
from types import SimpleNamespace
|
|
|
|
|
|
def test_mcp_collector_uses_ollama_before_gemini(monkeypatch):
|
|
import services.mcp_collector_service as mcp_mod
|
|
import services.mcp_router as router_mod
|
|
|
|
service = mcp_mod.MCPCollectorService()
|
|
monkeypatch.setattr(service, "_read_cache", lambda _topic: None)
|
|
monkeypatch.setattr(router_mod, "is_mcp_router_enabled", lambda: False)
|
|
monkeypatch.setattr(mcp_mod, "_OLLAMA_AVAILABLE", True)
|
|
|
|
def fail_gemini_init():
|
|
raise AssertionError("Gemini must not initialize before Ollama fallback is tried")
|
|
|
|
calls = []
|
|
|
|
class FakeOllamaService:
|
|
def __init__(self, model=None):
|
|
self.model = model
|
|
|
|
def generate(self, **kwargs):
|
|
calls.append(kwargs)
|
|
return SimpleNamespace(
|
|
success=True,
|
|
content="(此為基於歷史趨勢的預測性洞察) 台灣電商促銷可先檢查母親節、618 與競品價差。",
|
|
error=None,
|
|
)
|
|
|
|
monkeypatch.setattr(service, "_ensure_init", fail_gemini_init)
|
|
monkeypatch.setattr(mcp_mod, "OllamaService", FakeOllamaService)
|
|
|
|
result = service._search_topic("market_trends", "台灣電商促銷趨勢")
|
|
|
|
assert "基於歷史趨勢" in result
|
|
assert calls
|
|
assert calls[0]["model"]
|
|
assert calls[0]["allow_111_fallback"] is False
|
|
assert calls[0]["timeout"] == mcp_mod.MCP_OLLAMA_TIMEOUT
|
|
assert calls[0]["options"]["num_predict"] == mcp_mod.MCP_OLLAMA_NUM_PREDICT
|
|
|
|
|
|
def test_mcp_collector_market_insight_does_not_use_111_fallback(monkeypatch):
|
|
import services.mcp_collector_service as mcp_mod
|
|
|
|
service = mcp_mod.MCPCollectorService()
|
|
monkeypatch.setattr(mcp_mod, "_OLLAMA_AVAILABLE", True)
|
|
calls = []
|
|
|
|
class FakeOllamaService:
|
|
def __init__(self, model=None):
|
|
self.model = model
|
|
|
|
def generate(self, **kwargs):
|
|
calls.append(kwargs)
|
|
return SimpleNamespace(
|
|
success=False,
|
|
content="",
|
|
error="111 fallback disabled; no approved GCP Ollama host available",
|
|
)
|
|
|
|
monkeypatch.setattr(mcp_mod, "OllamaService", FakeOllamaService)
|
|
|
|
result = service._ollama_topic_fallback("market_trends", "台灣電商促銷趨勢")
|
|
|
|
assert result is None
|
|
assert calls
|
|
assert calls[0]["allow_111_fallback"] is False
|
|
assert calls[0]["timeout"] <= 25
|