388 lines
13 KiB
Python
388 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试模块化的Commander
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 添加项目根目录到Python路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
def test_modular_imports():
|
||
"""测试模块化导入"""
|
||
print("🔍 测试模块化导入")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
# 测试基础模块导入
|
||
from python_core.utils.commander.types import CommandConfig
|
||
from python_core.utils.commander.parser import ArgumentParser
|
||
from python_core.utils.commander.base import JSONRPCCommander
|
||
from python_core.utils.commander.simple import SimpleJSONRPCCommander, create_simple_commander
|
||
print("✅ 基础模块导入成功")
|
||
|
||
# 测试统一导入
|
||
from python_core.utils.commander import (
|
||
CommandConfig, ArgumentParser, JSONRPCCommander,
|
||
SimpleJSONRPCCommander, create_simple_commander
|
||
)
|
||
print("✅ 统一导入成功")
|
||
|
||
# 测试进度模块导入
|
||
from python_core.utils.progress import (
|
||
ProgressInfo, TaskResult, ProgressiveTask,
|
||
ProgressJSONRPCCommander, create_progress_commander
|
||
)
|
||
print("✅ 进度模块导入成功")
|
||
|
||
return True
|
||
|
||
except ImportError as e:
|
||
print(f"❌ 导入失败: {e}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
return False
|
||
|
||
def test_command_config():
|
||
"""测试命令配置"""
|
||
print("\n📋 测试命令配置")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
from python_core.utils.commander.types import CommandConfig
|
||
|
||
# 创建命令配置
|
||
config = CommandConfig(
|
||
name="test_command",
|
||
description="测试命令",
|
||
required_args=["input"],
|
||
optional_args={
|
||
"output": {"type": str, "default": "output.txt"},
|
||
"verbose": {"type": bool, "default": False}
|
||
}
|
||
)
|
||
|
||
print(f"✅ 命令配置创建成功: {config.name}")
|
||
print(f" 描述: {config.description}")
|
||
print(f" 必需参数: {config.required_args}")
|
||
print(f" 可选参数: {list(config.optional_args.keys())}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 命令配置测试失败: {e}")
|
||
return False
|
||
|
||
def test_argument_parser():
|
||
"""测试参数解析器"""
|
||
print("\n🔧 测试参数解析器")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
from python_core.utils.commander.types import CommandConfig
|
||
from python_core.utils.commander.parser import ArgumentParser
|
||
|
||
# 创建命令配置
|
||
commands = {
|
||
"process": CommandConfig(
|
||
name="process",
|
||
description="处理文件",
|
||
required_args=["input_file"],
|
||
optional_args={
|
||
"output": {"type": str, "default": "output.txt"},
|
||
"format": {"type": str, "default": "json", "choices": ["json", "xml"]},
|
||
"verbose": {"type": bool, "default": False}
|
||
}
|
||
)
|
||
}
|
||
|
||
# 创建解析器
|
||
parser = ArgumentParser(commands)
|
||
|
||
# 测试解析
|
||
test_cases = [
|
||
(["process", "input.txt"], True),
|
||
(["process", "input.txt", "--output", "result.txt"], True),
|
||
(["process", "input.txt", "--format", "xml", "--verbose"], True),
|
||
(["unknown"], False), # 未知命令
|
||
(["process"], False), # 缺少必需参数
|
||
]
|
||
|
||
for args, should_succeed in test_cases:
|
||
try:
|
||
command, parsed_args = parser.parse_arguments(args)
|
||
if should_succeed:
|
||
print(f"✅ 解析成功: {args} -> {command}, {parsed_args}")
|
||
else:
|
||
print(f"⚠️ 预期失败但成功了: {args}")
|
||
except ValueError as e:
|
||
if not should_succeed:
|
||
print(f"✅ 预期失败: {args} -> {e}")
|
||
else:
|
||
print(f"❌ 意外失败: {args} -> {e}")
|
||
|
||
return True
|
||
|
||
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"):
|
||
return {"message": f"Hello, {name}!"}
|
||
|
||
def add_handler(a: str, b: str):
|
||
return {"result": float(a) + float(b)}
|
||
|
||
# 添加命令
|
||
commander.add_command(
|
||
name="hello",
|
||
handler=hello_handler,
|
||
description="打招呼",
|
||
optional_args={
|
||
"name": {"type": str, "default": "World"}
|
||
}
|
||
)
|
||
|
||
commander.add_command(
|
||
name="add",
|
||
handler=add_handler,
|
||
description="加法运算",
|
||
required_args=["a", "b"]
|
||
)
|
||
|
||
print("✅ 命令注册成功")
|
||
|
||
# 测试命令执行
|
||
test_cases = [
|
||
(["hello"], {"message": "Hello, World!"}),
|
||
(["hello", "--name", "Alice"], {"message": "Hello, Alice!"}),
|
||
(["add", "5", "3"], {"result": 8.0}),
|
||
]
|
||
|
||
for args, expected in test_cases:
|
||
try:
|
||
command, parsed_args = commander.parse_arguments(args)
|
||
result = commander.execute_command(command, parsed_args)
|
||
|
||
if result == expected:
|
||
print(f"✅ 命令测试成功: {args} -> {result}")
|
||
else:
|
||
print(f"❌ 结果不匹配: {args} -> 期望 {expected}, 得到 {result}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"❌ 命令执行失败: {args} -> {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 简单Commander测试失败: {e}")
|
||
return False
|
||
|
||
def test_video_splitter_integration():
|
||
"""测试视频拆分服务集成"""
|
||
print("\n🎬 测试视频拆分服务集成")
|
||
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
|
||
|
||
# 测试参数解析
|
||
test_args = ["analyze", "test.mp4", "--threshold", "30.0"]
|
||
|
||
try:
|
||
command, parsed_args = commander.parse_arguments(test_args)
|
||
print(f"✅ 参数解析成功: {command}, {parsed_args}")
|
||
except Exception as e:
|
||
print(f"❌ 参数解析失败: {e}")
|
||
return False
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 视频拆分集成测试失败: {e}")
|
||
return False
|
||
|
||
def test_file_structure():
|
||
"""测试文件结构"""
|
||
print("\n📁 测试文件结构")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
# 检查Commander模块文件
|
||
commander_files = [
|
||
"python_core/utils/commander/__init__.py",
|
||
"python_core/utils/commander/types.py",
|
||
"python_core/utils/commander/parser.py",
|
||
"python_core/utils/commander/base.py",
|
||
"python_core/utils/commander/simple.py"
|
||
]
|
||
|
||
for file_path in commander_files:
|
||
full_path = project_root / file_path
|
||
if full_path.exists():
|
||
print(f"✅ {file_path} 存在")
|
||
else:
|
||
print(f"❌ {file_path} 不存在")
|
||
return False
|
||
|
||
# 检查Progress模块文件
|
||
progress_files = [
|
||
"python_core/utils/progress/__init__.py",
|
||
"python_core/utils/progress/types.py",
|
||
"python_core/utils/progress/task.py",
|
||
"python_core/utils/progress/reporter.py",
|
||
"python_core/utils/progress/generator.py",
|
||
"python_core/utils/progress/decorators.py",
|
||
"python_core/utils/progress/commander.py"
|
||
]
|
||
|
||
for file_path in progress_files:
|
||
full_path = project_root / file_path
|
||
if full_path.exists():
|
||
print(f"✅ {file_path} 存在")
|
||
else:
|
||
print(f"❌ {file_path} 不存在")
|
||
return False
|
||
|
||
# 检查原文件是否已删除
|
||
old_files = [
|
||
"python_core/utils/jsonrpc_commander.py",
|
||
"python_core/utils/progress_commander.py"
|
||
]
|
||
|
||
for file_path in old_files:
|
||
full_path = project_root / file_path
|
||
if not full_path.exists():
|
||
print(f"✅ {file_path} 已删除")
|
||
else:
|
||
print(f"⚠️ {file_path} 仍然存在")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 文件结构测试失败: {e}")
|
||
return False
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("🚀 测试模块化Commander")
|
||
|
||
try:
|
||
# 运行所有测试
|
||
tests = [
|
||
test_modular_imports,
|
||
test_command_config,
|
||
test_argument_parser,
|
||
test_simple_commander,
|
||
test_video_splitter_integration,
|
||
test_file_structure
|
||
]
|
||
|
||
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("📊 模块化Commander测试总结")
|
||
print("=" * 60)
|
||
|
||
passed = sum(results)
|
||
total = len(results)
|
||
|
||
print(f"通过测试: {passed}/{total}")
|
||
|
||
if passed == total:
|
||
print("🎉 所有模块化测试通过！")
|
||
print("\n✅ 模块化优势:")
|
||
print(" 1. 代码组织 - 每个模块职责单一明确")
|
||
print(" 2. 易于维护 - 修改某个功能只需改对应模块")
|
||
print(" 3. 可重用性 - 模块可以独立使用")
|
||
print(" 4. 测试友好 - 可以单独测试每个模块")
|
||
print(" 5. 扩展性强 - 新功能可以独立添加")
|
||
|
||
print("\n📁 模块结构:")
|
||
print(" commander/")
|
||
print(" ├── types.py - 数据类型定义")
|
||
print(" ├── parser.py - 参数解析器")
|
||
print(" ├── base.py - 基础Commander类")
|
||
print(" ├── simple.py - 简化Commander类")
|
||
print(" └── __init__.py - 统一导入接口")
|
||
print(" ")
|
||
print(" progress/")
|
||
print(" ├── types.py - 进度相关类型")
|
||
print(" ├── task.py - 任务管理")
|
||
print(" ├── reporter.py - 进度报告")
|
||
print(" ├── generator.py - 进度生成器")
|
||
print(" ├── decorators.py - 装饰器")
|
||
print(" ├── commander.py - 进度Commander")
|
||
print(" └── __init__.py - 统一导入接口")
|
||
|
||
print("\n🎯 使用方式:")
|
||
print(" # 基础Commander")
|
||
print(" from python_core.utils.commander import JSONRPCCommander")
|
||
print(" # 简单Commander")
|
||
print(" from python_core.utils.commander import create_simple_commander")
|
||
print(" # 进度Commander")
|
||
print(" from python_core.utils.progress import ProgressJSONRPCCommander")
|
||
|
||
return 0
|
||
else:
|
||
print("⚠️ 部分模块化测试失败")
|
||
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)
|