test_mcp.py•28.9 kB
#!/usr/bin/env python3
"""
翻译 MCP 服务综合测试
整合所有测试功能的综合测试文件，包括：
- 基础 MCP 功能测试
- 文本翻译测试
- 语言检测测试
- 长文本分块翻译测试
- 文件翻译测试
- 环境变量配置测试
"""
import asyncio
import json
import os
import subprocess
import sys
from typing import Any, Dict, List
from dotenv import load_dotenv
load_dotenv()
class ComprehensiveMCPTester:
"""综合 MCP 测试器"""
def __init__(self):
self.mcp_process = None
self.test_results = []
async def start_mcp_server(self) -> subprocess.Popen:
"""启动 MCP 服务器"""
try:
if not os.getenv("OPENAI_API_KEY"):
print("错误: 请设置 OPENAI_API_KEY 环境变量")
sys.exit(1)
process = subprocess.Popen(
[sys.executable, "translation_mcp.py"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
cwd=os.path.dirname(os.path.abspath(__file__))
)
print("✓ MCP 服务器已启动")
return process
except Exception as e:
print(f"✗ 启动 MCP 服务器失败: {e}")
sys.exit(1)
def send_mcp_request(self, process: subprocess.Popen, request: Dict[str, Any]) -> Dict[str, Any]:
"""发送 MCP 请求"""
try:
request_json = json.dumps(request) + "\n"
process.stdin.write(request_json)
process.stdin.flush()
response_line = process.stdout.readline()
if not response_line:
raise Exception("未收到响应")
response = json.loads(response_line.strip())
return response
except Exception as e:
raise Exception(f"MCP 请求失败: {e}")
def initialize_mcp(self, process: subprocess.Popen) -> bool:
"""初始化 MCP 连接"""
try:
# 初始化请求
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "comprehensive-tester",
"version": "1.0.0"
}
}
}
response = self.send_mcp_request(process, init_request)
if "result" in response:
# 发送 initialized 通知
initialized_notification = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
notification_json = json.dumps(initialized_notification) + "\n"
process.stdin.write(notification_json)
process.stdin.flush()
print("✓ MCP 初始化成功")
return True
else:
print(f"✗ MCP 初始化失败: {response}")
return False
except Exception as e:
print(f"✗ MCP 初始化失败: {e}")
return False
async def test_initialization(self, process: subprocess.Popen) -> bool:
"""测试初始化"""
try:
print("\n=== 测试 1: 初始化 ===")
# 初始化已在 initialize_mcp 中完成
print("✓ 初始化测试通过")
return True
except Exception as e:
print(f"✗ 初始化测试失败: {e}")
return False
async def test_tool_list(self, process: subprocess.Popen) -> bool:
"""测试工具列表"""
try:
print("\n=== 测试 2: 获取工具列表 ===")
request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
response = self.send_mcp_request(process, request)
if "result" in response and "tools" in response["result"]:
tools = response["result"]["tools"]
tool_names = [tool["name"] for tool in tools]
expected_tools = ["translate_text", "detect_language", "translate_long_text"]
if all(tool in tool_names for tool in expected_tools):
print(f"✓ 工具列表正确: {tool_names}")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
return True
else:
print(f"✗ 工具列表不完整，期望: {expected_tools}，实际: {tool_names}")
return False
else:
print(f"✗ 获取工具列表失败: {response}")
return False
except Exception as e:
print(f"✗ 测试工具列表失败: {e}")
return False
async def test_translate_text(self, process: subprocess.Popen) -> bool:
"""测试文本翻译"""
try:
print("\n=== 测试 3: 文本翻译 ===")
test_cases = [
{"text": "Hello, world!", "target": "中文", "expected_contains": ["你好", "世界"]},
{"text": "Good morning", "target": "日文", "expected_contains": ["おはよう"]},
{"text": "Thank you", "target": "韩文", "expected_contains": ["감사", "고마워"]}
]
for i, case in enumerate(test_cases, 1):
print(f"\n 测试用例 {i}: {case['text']} -> {case['target']}")
request = {
"jsonrpc": "2.0",
"id": 10 + i,
"method": "tools/call",
"params": {
"name": "translate_text",
"arguments": {
"text": case["text"],
"target_language": case["target"]
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
translated = result.get("translated_text", "")
print(f" ✓ 翻译成功: {translated}")
else:
print(f" ✗ 翻译失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 响应格式错误: {result_text}")
return False
else:
print(f" ✗ 响应内容为空")
return False
else:
print(f" ✗ 请求失败: {response}")
return False
# 测试间隔
await asyncio.sleep(0.5)
print("✓ 所有文本翻译测试通过")
return True
except Exception as e:
print(f"✗ 测试文本翻译失败: {e}")
return False
async def test_detect_language(self, process: subprocess.Popen) -> bool:
"""测试语言检测"""
try:
print("\n=== 测试 4: 语言检测 ===")
test_cases = [
{"text": "Hello, how are you?", "expected": "en"},
{"text": "你好，你好吗？", "expected": "zh"},
{"text": "こんにちは、元気ですか？", "expected": "ja"},
{"text": "안녕하세요, 어떻게 지내세요?", "expected": "ko"}
]
for i, case in enumerate(test_cases, 1):
print(f"\n 测试用例 {i}: {case['text']}")
request = {
"jsonrpc": "2.0",
"id": 20 + i,
"method": "tools/call",
"params": {
"name": "detect_language",
"arguments": {
"text": case["text"]
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
detected = result.get("detected_language", "")
print(f" ✓ 检测结果: {detected}")
else:
print(f" ✗ 检测失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 响应格式错误: {result_text}")
return False
else:
print(f" ✗ 响应内容为空")
return False
else:
print(f" ✗ 请求失败: {response}")
return False
# 测试间隔
await asyncio.sleep(0.5)
print("✓ 所有语言检测测试通过")
return True
except Exception as e:
print(f"✗ 测试语言检测失败: {e}")
return False
async def test_long_text_translation(self, process: subprocess.Popen) -> bool:
"""测试长文本翻译"""
try:
print("\n=== 测试 5: 长文本翻译 ===")
# 短文本测试（应该不分块）
print("\n 测试短文本（不分块）:")
short_text = "This is a short text for testing."
request = {
"jsonrpc": "2.0",
"id": 30,
"method": "tools/call",
"params": {
"name": "translate_long_text",
"arguments": {
"text": short_text,
"target_language": "中文"
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
chunks = result.get("chunks_count", "N/A")
print(f" ✓ 短文本翻译成功，分块数: {chunks}")
else:
print(f" ✗ 短文本翻译失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 短文本翻译响应格式错误: {result_text}")
return False
else:
print(f" ✗ 短文本翻译响应内容为空")
return False
else:
print(f" ✗ 短文本翻译请求失败: {response}")
return False
await asyncio.sleep(1)
# 长文本测试（应该分块）
print("\n 测试长文本（自动分块）:")
long_text = """This is a comprehensive test for the long text translation feature.
The translate_long_text tool should automatically split this text into smaller chunks
and translate each chunk separately, then combine the results.
This approach is particularly useful for translating large documents,
books, articles, or any content that exceeds the typical token limits
of language models. The chunking mechanism ensures that the translation
process remains efficient and reliable, even for very long texts.
Each chunk is processed independently, which helps maintain consistency
and allows for better error handling. The final result should be
a seamless translation that preserves the original structure and meaning
of the entire document. This test verifies that the chunking functionality
works correctly and produces high-quality translations for long content."""
request = {
"jsonrpc": "2.0",
"id": 31,
"method": "tools/call",
"params": {
"name": "translate_long_text",
"arguments": {
"text": long_text,
"target_language": "中文",
"max_chunk_size": 500 # 设置较小的块大小来测试分块功能
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
chunks = result.get("chunks_count", "N/A")
translated_length = len(result.get("translated_text", ""))
print(f" ✓ 长文本翻译成功")
print(f" 原文长度: {len(long_text)} 字符")
print(f" 译文长度: {translated_length} 字符")
print(f" 分块数: {chunks}")
if chunks and chunks > 1:
print(f" ✓ 分块功能正常工作")
else:
print(f" ⚠️ 分块数可能不正确: {chunks}")
else:
print(f" ✗ 长文本翻译失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 长文本翻译响应格式错误: {result_text}")
return False
else:
print(f" ✗ 长文本翻译响应内容为空")
return False
else:
print(f" ✗ 长文本翻译请求失败: {response}")
return False
print("✓ 长文本翻译测试通过")
return True
except Exception as e:
print(f"✗ 测试长文本翻译失败: {e}")
return False
async def test_file_translation(self, process: subprocess.Popen) -> bool:
"""测试文件翻译"""
try:
print("\n=== 测试 6: 文件翻译 ===")
# 检查测试文件是否存在
test_file = "test_text.md"
if not os.path.exists(test_file):
print(f" ⚠️ 测试文件 {test_file} 不存在，跳过文件翻译测试")
return True
# 读取测试文件
try:
with open(test_file, 'r', encoding='utf-8') as f:
file_content = f.read()
print(f" ✓ 已读取测试文件: {test_file} ({len(file_content)} 字符)")
except Exception as e:
print(f" ✗ 读取测试文件失败: {e}")
return False
# 使用长文本翻译功能翻译文件内容
print(f" 正在翻译文件内容...")
request = {
"jsonrpc": "2.0",
"id": 40,
"method": "tools/call",
"params": {
"name": "translate_long_text",
"arguments": {
"text": file_content,
"target_language": "中文",
"max_chunk_size": 2000
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
translated_content = result.get("translated_text", "")
chunks = result.get("chunks_count", "N/A")
# 保存翻译结果
output_file = "test_comprehensive_translation.md"
try:
with open(output_file, 'w', encoding='utf-8') as f:
f.write(translated_content)
print(f" ✓ 文件翻译成功")
print(f" 原文长度: {len(file_content)} 字符")
print(f" 译文长度: {len(translated_content)} 字符")
print(f" 分块数: {chunks}")
print(f" 翻译结果已保存到: {output_file}")
return True
except Exception as e:
print(f" ✗ 保存翻译结果失败: {e}")
return False
else:
print(f" ✗ 文件翻译失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 文件翻译响应格式错误: {result_text}")
return False
else:
print(f" ✗ 文件翻译响应内容为空")
return False
else:
print(f" ✗ 文件翻译请求失败: {response}")
return False
except Exception as e:
print(f"✗ 测试文件翻译失败: {e}")
return False
async def test_environment_config(self, process: subprocess.Popen) -> bool:
"""测试环境变量配置"""
try:
print("\n=== 测试 7: 环境变量配置 ===")
# 测试环境变量是否正确配置
api_key = os.getenv("OPENAI_API_KEY")
model = os.getenv("OPENAI_MODEL", "gpt-3.5-turbo")
base_url = os.getenv("OPENAI_BASE_URL")
print(f" API Key: {'已设置' if api_key else '未设置'}")
print(f" Model: {model}")
print(f" Base URL: {base_url if base_url else '默认'}")
# 测试翻译功能是否使用环境变量
request = {
"jsonrpc": "2.0",
"id": 50,
"method": "tools/call",
"params": {
"name": "translate_text",
"arguments": {
"text": "Environment configuration test",
"target_language": "中文"
}
}
}
response = self.send_mcp_request(process, request)
if "result" in response and "content" in response["result"]:
content = response["result"]["content"]
if content and len(content) > 0:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if result.get("success"):
print(f" ✓ 环境变量配置测试成功")
print(f" 译文: {result.get('translated_text', '')}")
return True
else:
print(f" ✗ 环境变量配置测试失败: {result.get('error', '未知错误')}")
return False
except json.JSONDecodeError:
print(f" ✗ 环境变量配置测试响应格式错误: {result_text}")
return False
else:
print(f" ✗ 环境变量配置测试响应内容为空")
return False
else:
print(f" ✗ 环境变量配置测试请求失败: {response}")
return False
except Exception as e:
print(f"✗ 测试环境变量配置失败: {e}")
return False
async def test_error_handling(self, process: subprocess.Popen) -> bool:
"""测试错误处理"""
try:
print("\n=== 测试 8: 错误处理 ===")
# 测试无效工具名
print("\n 测试无效工具名:")
request = {
"jsonrpc": "2.0",
"id": 60,
"method": "tools/call",
"params": {
"name": "invalid_tool",
"arguments": {}
}
}
response = self.send_mcp_request(process, request)
if "error" in response:
print(f" ✓ 正确处理无效工具名错误")
else:
print(f" ⚠️ 无效工具名未返回错误: {response}")
await asyncio.sleep(0.5)
# 测试缺少必需参数
print("\n 测试缺少必需参数:")
request = {
"jsonrpc": "2.0",
"id": 61,
"method": "tools/call",
"params": {
"name": "translate_text",
"arguments": {
"text": "Test text"
# 缺少 target_language
}
}
}
response = self.send_mcp_request(process, request)
if "error" in response or ("result" in response and "content" in response["result"]):
content = response.get("result", {}).get("content", [])
if content:
result_text = content[0].get("text", "")
try:
result = json.loads(result_text)
if not result.get("success"):
print(f" ✓ 正确处理缺少参数错误")
else:
print(f" ⚠️ 缺少参数未返回错误")
except json.JSONDecodeError:
print(f" ⚠️ 缺少参数响应格式异常")
else:
print(f" ✓ 正确处理缺少参数错误")
else:
print(f" ⚠️ 缺少参数未返回错误: {response}")
print("✓ 错误处理测试完成")
return True
except Exception as e:
print(f"✗ 测试错误处理失败: {e}")
return False
async def run_all_tests(self):
"""运行所有测试"""
print("开始运行翻译 MCP 服务综合测试")
print("=" * 80)
# 启动 MCP 服务器
process = await self.start_mcp_server()
try:
# 等待服务器启动
await asyncio.sleep(1)
# 初始化 MCP 连接
if not self.initialize_mcp(process):
return False
# 定义所有测试
tests = [
("初始化", self.test_initialization),
("工具列表", self.test_tool_list),
("文本翻译", self.test_translate_text),
("语言检测", self.test_detect_language),
("长文本翻译", self.test_long_text_translation),
("文件翻译", self.test_file_translation),
("环境变量配置", self.test_environment_config),
("错误处理", self.test_error_handling)
]
passed = 0
total = len(tests)
# 运行所有测试
for test_name, test_func in tests:
try:
print(f"\n{'='*20} 运行测试: {test_name} {'='*20}")
if await test_func(process):
passed += 1
self.test_results.append((test_name, True, None))
print(f"✓ 测试 '{test_name}' 通过")
else:
self.test_results.append((test_name, False, "测试失败"))
print(f"✗ 测试 '{test_name}' 失败")
# 测试间隔
await asyncio.sleep(1)
except Exception as e:
self.test_results.append((test_name, False, str(e)))
print(f"✗ 测试 '{test_name}' 异常: {e}")
# 输出测试总结
self.print_test_summary(passed, total)
return passed == total
finally:
# 清理
if process:
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
print("\n✓ MCP 服务器已停止")
def print_test_summary(self, passed: int, total: int):
"""打印测试总结"""
print("\n" + "=" * 80)
print("测试总结")
print("=" * 80)
for test_name, success, error in self.test_results:
status = "✓ 通过" if success else f"✗ 失败 ({error})"
print(f"{test_name:<20} {status}")
print("\n" + "-" * 80)
print(f"总计: {passed}/{total} 个测试通过")
if passed == total:
print("🎉 所有测试都通过了！MCP 翻译服务功能完整且正常工作。")
else:
failed = total - passed
print(f"⚠️ 有 {failed} 个测试失败，请检查相关功能。")
print("=" * 80)
def main():
"""主函数"""
# 检查环境变量
if not os.getenv("OPENAI_API_KEY"):
print("错误: 请设置 OPENAI_API_KEY 环境变量")
print("示例: export OPENAI_API_KEY=your_api_key")
sys.exit(1)
# 运行综合测试
tester = ComprehensiveMCPTester()
try:
result = asyncio.run(tester.run_all_tests())
sys.exit(0 if result else 1)
except KeyboardInterrupt:
print("\n测试被用户中断")
sys.exit(1)
except Exception as e:
print(f"\n测试执行失败: {e}")
sys.exit(1)
if __name__ == "__main__":
main()