json rpc commander 封装
This commit is contained in:
root
333
scripts/test_jsonrpc_commander.py
Normal file
333
scripts/test_jsonrpc_commander.py
Normal file
@@ -0,0 +1,333 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
测试JSON-RPC Commander基类
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
# 添加项目根目录到Python路径
|
||||
project_root = Path(__file__).parent.parent
|
||||
sys.path.insert(0, str(project_root))
|
||||
|
||||
def test_commander_import():
|
||||
"""测试Commander导入"""
|
||||
print("🔍 测试Commander导入")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import (
|
||||
JSONRPCCommander, SimpleJSONRPCCommander, create_simple_commander
|
||||
)
|
||||
print("✅ JSON-RPC Commander导入成功")
|
||||
|
||||
# 测试创建简单Commander
|
||||
commander = create_simple_commander("test_service")
|
||||
print("✅ 简单Commander创建成功")
|
||||
|
||||
return True
|
||||
|
||||
except ImportError as e:
|
||||
print(f"❌ 导入失败: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"❌ 测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_simple_commander():
|
||||
"""测试简单Commander功能"""
|
||||
print("\n🎯 测试简单Commander功能")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import create_simple_commander
|
||||
|
||||
# 创建Commander
|
||||
commander = create_simple_commander("test_service")
|
||||
|
||||
# 定义测试命令处理器
|
||||
def hello_handler(name: str = "World", count: int = 1):
|
||||
"""测试命令处理器"""
|
||||
return {
|
||||
"message": f"Hello, {name}!",
|
||||
"count": count,
|
||||
"repeated": [f"Hello, {name}!" for _ in range(count)]
|
||||
}
|
||||
|
||||
def add_handler(a: str, b: str):
|
||||
"""加法命令处理器"""
|
||||
# 转换为浮点数
|
||||
num_a = float(a)
|
||||
num_b = float(b)
|
||||
return {
|
||||
"operation": "add",
|
||||
"operands": [num_a, num_b],
|
||||
"result": num_a + num_b
|
||||
}
|
||||
|
||||
# 添加命令
|
||||
commander.add_command(
|
||||
name="hello",
|
||||
handler=hello_handler,
|
||||
description="打招呼命令",
|
||||
required_args=[],
|
||||
optional_args={
|
||||
"name": {"type": str, "default": "World", "description": "名称"},
|
||||
"count": {"type": int, "default": 1, "description": "重复次数"}
|
||||
}
|
||||
)
|
||||
|
||||
commander.add_command(
|
||||
name="add",
|
||||
handler=add_handler,
|
||||
description="加法运算",
|
||||
required_args=["a", "b"],
|
||||
optional_args={}
|
||||
)
|
||||
|
||||
print("✅ 命令注册成功")
|
||||
|
||||
# 测试命令解析和执行
|
||||
test_cases = [
|
||||
# (args, expected_success)
|
||||
(["hello"], True),
|
||||
(["hello", "--name", "Alice"], True),
|
||||
(["hello", "--name", "Bob", "--count", "3"], True),
|
||||
(["add", "5.5", "3.2"], True),
|
||||
(["unknown"], False), # 未知命令
|
||||
(["add", "5.5"], False), # 缺少参数
|
||||
]
|
||||
|
||||
for args, expected_success in test_cases:
|
||||
try:
|
||||
command, parsed_args = commander.parse_arguments(args)
|
||||
result = commander.execute_command(command, parsed_args)
|
||||
|
||||
if expected_success:
|
||||
print(f"✅ 测试成功: {args} -> {result}")
|
||||
else:
|
||||
print(f"⚠️ 预期失败但成功了: {args}")
|
||||
|
||||
except SystemExit:
|
||||
if not expected_success:
|
||||
print(f"✅ 预期失败: {args}")
|
||||
else:
|
||||
print(f"❌ 意外失败: {args}")
|
||||
except Exception as e:
|
||||
if not expected_success:
|
||||
print(f"✅ 预期失败: {args} -> {e}")
|
||||
else:
|
||||
print(f"❌ 意外错误: {args} -> {e}")
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 简单Commander测试失败: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
def test_video_splitter_commander():
|
||||
"""测试视频拆分Commander"""
|
||||
print("\n🎬 测试视频拆分Commander")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
# 检查依赖
|
||||
try:
|
||||
import scenedetect
|
||||
print(f"✅ PySceneDetect {scenedetect.__version__} 可用")
|
||||
except ImportError:
|
||||
print("⚠️ PySceneDetect不可用，跳过视频拆分测试")
|
||||
return True
|
||||
|
||||
from python_core.services.video_splitter.cli import VideoSplitterCommander
|
||||
|
||||
# 创建Commander
|
||||
commander = VideoSplitterCommander()
|
||||
print("✅ 视频拆分Commander创建成功")
|
||||
|
||||
# 检查注册的命令
|
||||
commands = list(commander.commands.keys())
|
||||
expected_commands = ["analyze", "detect_scenes"]
|
||||
|
||||
for cmd in expected_commands:
|
||||
if cmd in commands:
|
||||
print(f"✅ 命令 '{cmd}' 已注册")
|
||||
else:
|
||||
print(f"❌ 命令 '{cmd}' 未注册")
|
||||
return False
|
||||
|
||||
# 查找测试视频
|
||||
assets_dir = project_root / "assets"
|
||||
video_files = list(assets_dir.rglob("*.mp4"))
|
||||
|
||||
if not video_files:
|
||||
print("⚠️ 没有找到测试视频，跳过功能测试")
|
||||
return True
|
||||
|
||||
test_video = str(video_files[0])
|
||||
print(f"📹 测试视频: {test_video}")
|
||||
|
||||
# 测试命令解析
|
||||
test_args = ["analyze", test_video, "--threshold", "30.0"]
|
||||
|
||||
try:
|
||||
command, parsed_args = commander.parse_arguments(test_args)
|
||||
print(f"✅ 参数解析成功: {command}, {parsed_args}")
|
||||
|
||||
# 测试命令执行
|
||||
result = commander.execute_command(command, parsed_args)
|
||||
|
||||
if isinstance(result, dict) and result.get("success"):
|
||||
print(f"✅ 命令执行成功:")
|
||||
print(f" 总场景数: {result.get('total_scenes', 0)}")
|
||||
print(f" 总时长: {result.get('total_duration', 0):.2f}秒")
|
||||
else:
|
||||
print(f"❌ 命令执行失败: {result}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 命令测试失败: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 视频拆分Commander测试失败: {e}")
|
||||
return False
|
||||
|
||||
def test_jsonrpc_output():
|
||||
"""测试JSON-RPC输出格式"""
|
||||
print("\n📡 测试JSON-RPC输出格式")
|
||||
print("=" * 50)
|
||||
|
||||
try:
|
||||
from python_core.utils.jsonrpc_commander import create_simple_commander
|
||||
import io
|
||||
import contextlib
|
||||
|
||||
# 创建Commander
|
||||
commander = create_simple_commander("test_service")
|
||||
|
||||
def test_handler(message: str = "test"):
|
||||
return {"message": message, "timestamp": "2025-01-01T00:00:00"}
|
||||
|
||||
commander.add_command(
|
||||
name="test",
|
||||
handler=test_handler,
|
||||
description="测试命令",
|
||||
optional_args={
|
||||
"message": {"type": str, "default": "test"}
|
||||
}
|
||||
)
|
||||
|
||||
# 捕获输出
|
||||
output = io.StringIO()
|
||||
|
||||
with contextlib.redirect_stdout(output):
|
||||
try:
|
||||
commander.run(["test", "--message", "hello"])
|
||||
except SystemExit:
|
||||
pass # 正常退出
|
||||
|
||||
output_text = output.getvalue()
|
||||
print(f"📤 输出内容: {output_text[:100]}...")
|
||||
|
||||
# 验证输出是JSON格式
|
||||
try:
|
||||
if output_text.startswith("JSONRPC:"):
|
||||
json_str = output_text[8:]
|
||||
json_data = json.loads(json_str)
|
||||
print("✅ JSON-RPC格式输出")
|
||||
|
||||
if "result" in json_data:
|
||||
print(f"✅ 包含result字段: {json_data['result']}")
|
||||
else:
|
||||
print("⚠️ 缺少result字段")
|
||||
else:
|
||||
json_data = json.loads(output_text)
|
||||
print("✅ 直接JSON格式输出")
|
||||
print(f" 内容: {json_data}")
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"❌ 输出不是有效JSON: {e}")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ JSON-RPC输出测试失败: {e}")
|
||||
return False
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("🚀 测试JSON-RPC Commander基类")
|
||||
|
||||
try:
|
||||
# 运行所有测试
|
||||
tests = [
|
||||
test_commander_import,
|
||||
test_simple_commander,
|
||||
test_video_splitter_commander,
|
||||
test_jsonrpc_output
|
||||
]
|
||||
|
||||
results = []
|
||||
for test in tests:
|
||||
try:
|
||||
result = test()
|
||||
results.append(result)
|
||||
except Exception as e:
|
||||
print(f"❌ 测试 {test.__name__} 异常: {e}")
|
||||
results.append(False)
|
||||
|
||||
# 总结
|
||||
print("\n" + "=" * 60)
|
||||
print("📊 JSON-RPC Commander测试总结")
|
||||
print("=" * 60)
|
||||
|
||||
passed = sum(results)
|
||||
total = len(results)
|
||||
|
||||
print(f"通过测试: {passed}/{total}")
|
||||
|
||||
if passed == total:
|
||||
print("🎉 所有JSON-RPC Commander测试通过！")
|
||||
print("\n✅ 基类功能验证:")
|
||||
print(" 1. 命令注册和解析 - ✅")
|
||||
print(" 2. 参数类型转换 - ✅")
|
||||
print(" 3. 错误处理 - ✅")
|
||||
print(" 4. JSON-RPC输出 - ✅")
|
||||
print(" 5. 视频拆分集成 - ✅")
|
||||
|
||||
print("\n🔧 使用优势:")
|
||||
print(" 1. 统一接口 - 所有命令行工具使用相同基类")
|
||||
print(" 2. 自动解析 - 参数解析和类型转换自动化")
|
||||
print(" 3. 错误处理 - 统一的错误响应格式")
|
||||
print(" 4. JSON-RPC - 标准化的通信协议")
|
||||
print(" 5. 易于扩展 - 简单添加新命令")
|
||||
|
||||
print("\n📝 使用示例:")
|
||||
print(" # 继承基类")
|
||||
print(" class MyCommander(JSONRPCCommander):")
|
||||
print(" def _register_commands(self): ...")
|
||||
print(" def execute_command(self, cmd, args): ...")
|
||||
print(" # 或使用简化版本")
|
||||
print(" commander = create_simple_commander('my_service')")
|
||||
print(" commander.add_command('cmd', handler, 'description')")
|
||||
|
||||
return 0
|
||||
else:
|
||||
print("⚠️ 部分JSON-RPC Commander测试失败")
|
||||
return 1
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 测试过程中出错: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return 1
|
||||
|
||||
if __name__ == "__main__":
|
||||
exit_code = main()
|
||||
sys.exit(exit_code)
|
||||
Reference in New Issue
Block a user