#!/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)