334 lines
11 KiB
Python
334 lines
11 KiB
Python
#!/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.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.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.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)
|