251 lines
12 KiB
Python
251 lines
12 KiB
Python
"""
|
||
tests/test_mcp_router.py
|
||
─────────────────────────────────────────────────────────────────
|
||
Operation Ollama-First v5.0 / Phase 10.5 — MCP Router unit tests
|
||
|
||
驗證面：
|
||
T1. flag OFF → 直接回 success=False（不打 HTTP）
|
||
T2. flag ON + tool 不在白名單 → 回 success=False
|
||
T3. flag ON + 正常 → HTTP call + cache 命中第二次
|
||
T4. flag ON + HTTP timeout → success=False + error 含 timeout
|
||
T5. flag ON + HTTP 500 → success=False + error 含 HTTP 500
|
||
T6. cache key 排序穩定（相同 args 不同 dict 順序 → 同 key）
|
||
T7. health_check 全失敗時所有 server 回 False
|
||
|
||
紀律：
|
||
- 不打真實 MCP server（全 mock requests.post / requests.get）
|
||
- 不寫 mcp_calls 表（fire-and-forget thread 不影響 test）
|
||
"""
|
||
|
||
import json
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
import pytest
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_state(monkeypatch):
|
||
"""每 test 清 env + memory cache"""
|
||
monkeypatch.delenv('MCP_ROUTER_ENABLED', raising=False)
|
||
import services.mcp_router as mr
|
||
mr._memory_cache.clear()
|
||
yield
|
||
mr._memory_cache.clear()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T1: flag OFF
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_flag_off_returns_failure_without_http(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'false')
|
||
from services.mcp_router import mcp_router
|
||
|
||
with patch('services.mcp_router.requests.post') as mock_post:
|
||
result = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': 'test'}, caller='mcp_collector',
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.status == 'error'
|
||
assert 'MCP_ROUTER_ENABLED=false' in (result.error or '')
|
||
mock_post.assert_not_called()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T2: flag ON + tool 不在白名單
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_unauthorized_caller_tool_combo_rejected(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
with patch('services.mcp_router.requests.post') as mock_post:
|
||
# mcp_collector 不允許 postgres
|
||
result = mcp_router.call(
|
||
server='postgres', tool='query',
|
||
args={'sql': 'SELECT 1'}, caller='mcp_collector',
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.status == 'error'
|
||
assert 'tool not in registry' in (result.error or '')
|
||
mock_post.assert_not_called()
|
||
|
||
|
||
def test_filesystem_registry_is_read_only(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
fake_resp = MagicMock(status_code=200)
|
||
fake_resp.json.return_value = {'content': 'ok'}
|
||
fake_resp.text = '{"content": "ok"}'
|
||
|
||
with patch('services.mcp_router.requests.post', return_value=fake_resp) as mock_post:
|
||
allowed = mcp_router.call(
|
||
server='filesystem', tool='read_file',
|
||
args={'path': '/logs/system.log'}, caller='ops_diagnostics',
|
||
)
|
||
denied = mcp_router.call(
|
||
server='filesystem', tool='write_file',
|
||
args={'path': '/data/x', 'content': 'nope'}, caller='ops_diagnostics',
|
||
)
|
||
|
||
assert allowed.success is True
|
||
assert denied.success is False
|
||
assert 'tool not in registry' in (denied.error or '')
|
||
mock_post.assert_called_once()
|
||
assert mock_post.call_args.args[0].endswith('/tools/read_file')
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T3: flag ON + 正常 + cache
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_successful_call_and_cache_hit(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
import services.mcp_router as mr
|
||
from services.mcp_router import mcp_router
|
||
|
||
written_statuses = []
|
||
monkeypatch.setattr(
|
||
mr,
|
||
'_async_write_mcp_call',
|
||
lambda caller, server, tool, args, result, request_id=None: written_statuses.append(result.status),
|
||
)
|
||
|
||
fake_resp = MagicMock(status_code=200)
|
||
fake_resp.json.return_value = {'results': [{'title': '商品 A', 'content': '熱銷'}]}
|
||
fake_resp.text = '{"ok": true}'
|
||
|
||
with patch('services.mcp_router.requests.post', return_value=fake_resp) as mock_post:
|
||
# 第一次：HTTP call
|
||
r1 = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': '母親節'}, caller='mcp_collector',
|
||
)
|
||
assert r1.success is True
|
||
assert r1.cache_hit is False
|
||
assert r1.status == 'ok'
|
||
assert r1.data['results'][0]['title'] == '商品 A'
|
||
assert mock_post.call_count == 1
|
||
|
||
# 第二次：cache 命中
|
||
r2 = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': '母親節'}, caller='mcp_collector',
|
||
)
|
||
assert r2.success is True
|
||
assert r2.cache_hit is True
|
||
assert r2.status == 'cache_only'
|
||
assert mock_post.call_count == 1 # 沒再打 HTTP
|
||
assert written_statuses == ['ok', 'cache_only']
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T4: HTTP timeout
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_http_timeout_returns_failure(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
import requests as _r
|
||
from services.mcp_router import mcp_router
|
||
|
||
with patch('services.mcp_router.requests.post', side_effect=_r.Timeout('30s')):
|
||
result = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': 'x'}, caller='mcp_collector',
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.status == 'timeout'
|
||
assert 'timeout' in (result.error or '').lower()
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T5: HTTP 500
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_http_500_returns_failure(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
fake_resp = MagicMock(status_code=500)
|
||
fake_resp.text = 'Internal Server Error'
|
||
|
||
with patch('services.mcp_router.requests.post', return_value=fake_resp):
|
||
result = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': 'x'}, caller='mcp_collector',
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.status == 'error'
|
||
assert 'HTTP 500' in (result.error or '')
|
||
|
||
|
||
def test_http_429_records_rate_limited_status(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
fake_resp = MagicMock(status_code=429)
|
||
fake_resp.text = 'Too Many Requests'
|
||
|
||
with patch('services.mcp_router.requests.post', return_value=fake_resp):
|
||
result = mcp_router.call(
|
||
server='omnisearch', tool='tavily_search',
|
||
args={'query': 'x'}, caller='mcp_collector',
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.status == 'rate_limited'
|
||
assert 'HTTP 429' in (result.error or '')
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T6: cache key 穩定性
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_cache_key_sort_stable():
|
||
from services.mcp_router import _cache_key
|
||
k1 = _cache_key('omnisearch', 'tavily_search', {'a': 1, 'b': 2, 'c': 3})
|
||
k2 = _cache_key('omnisearch', 'tavily_search', {'c': 3, 'a': 1, 'b': 2})
|
||
assert k1 == k2 # dict 順序不影響 cache key
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T7: health_check 全失敗
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_health_check_all_fail_returns_all_false(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
with patch('services.mcp_router.requests.get', side_effect=Exception('connection refused')):
|
||
results = mcp_router.health_check()
|
||
|
||
assert results == {'postgres': False, 'firecrawl': False,
|
||
'omnisearch': False, 'filesystem': False}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# T8: 未知 server 直接拒絕
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def test_unknown_server_rejected(monkeypatch):
|
||
monkeypatch.setenv('MCP_ROUTER_ENABLED', 'true')
|
||
from services.mcp_router import mcp_router
|
||
|
||
with patch('services.mcp_router.requests.post') as mock_post:
|
||
# 統帥的 caller 在 registry，但 server 名拼錯
|
||
result = mcp_router.call(
|
||
server='omnisearch_wrong', tool='tavily_search',
|
||
args={'query': 'x'}, caller='mcp_collector',
|
||
)
|
||
|
||
# 注意：白名單檢查在前，會先回 'tool not in registry'（因為 omnisearch_wrong 不在 registry）
|
||
assert result.success is False
|
||
assert result.status == 'error'
|
||
mock_post.assert_not_called()
|