346 lines
12 KiB
Python
Executable File
346 lines
12 KiB
Python
Executable File
"""
|
||
主测试运行器
|
||
统一运行所有测试并生成综合报告
|
||
"""
|
||
import sys
|
||
import unittest
|
||
import time
|
||
import argparse
|
||
from pathlib import Path
|
||
from typing import List, Dict, Any, Optional
|
||
import json
|
||
|
||
# 添加测试基础模块
|
||
sys.path.insert(0, str(Path(__file__).parent / "netrans_api" / "unit"))
|
||
sys.path.insert(0, str(Path(__file__).parent / "netrans_api" / "integration"))
|
||
from test_base import TestReporter
|
||
|
||
import signal
|
||
import threading
|
||
from contextlib import contextmanager
|
||
|
||
@contextmanager
|
||
def timeout_context(seconds):
|
||
"""跨平台超时上下文管理器"""
|
||
if seconds is None:
|
||
yield
|
||
return
|
||
|
||
# 使用线程定时器实现超时
|
||
timeout_occurred = threading.Event()
|
||
|
||
def timeout_handler():
|
||
timeout_occurred.set()
|
||
|
||
timer = threading.Timer(seconds, timeout_handler)
|
||
timer.start()
|
||
|
||
try:
|
||
yield timeout_occurred
|
||
finally:
|
||
timer.cancel()
|
||
from test_core_functions import (
|
||
TestNetransCore,
|
||
TestModelMetadata
|
||
)
|
||
from test_quantize_types import TestQuantizerTypes
|
||
from test_integration import (
|
||
TestCompleteConversionPipeline,
|
||
TestBatchConversion,
|
||
TestConversionWithDifferentParameters,
|
||
TestModelCompatibility
|
||
)
|
||
from test_error_handling import (
|
||
TestErrorHandling,
|
||
TestBoundaryConditions,
|
||
TestResourceLimitations,
|
||
TestConcurrencyAndThreadSafety
|
||
)
|
||
from test_performance import (
|
||
TestQuantizationPerformance,
|
||
TestModelSizeAndCompression,
|
||
TestConversionThroughput,
|
||
TestMemoryEfficiency
|
||
)
|
||
from test_all_quantization_types import TestAllQuantizationTypes
|
||
|
||
|
||
class NetransTestSuite:
|
||
"""Netrans测试套件管理器"""
|
||
|
||
def __init__(self):
|
||
# 使用绝对路径，避免工作目录改变后路径失效
|
||
self.test_dir = Path(__file__).parent.absolute()
|
||
# 直接使用test目录，不再使用reports子目录
|
||
self.reporter = TestReporter(self.test_dir)
|
||
self.results = []
|
||
|
||
def create_test_suite(self, test_level: str = "all") -> unittest.TestSuite:
|
||
"""
|
||
创建测试套件
|
||
|
||
Args:
|
||
test_level: 测试级别 ('basic', 'full', 'performance', 'quantization', 'all')
|
||
"""
|
||
suite = unittest.TestSuite()
|
||
|
||
if test_level in ["basic", "all"]:
|
||
# 基础功能测试
|
||
suite.addTest(unittest.makeSuite(TestNetransCore))
|
||
suite.addTest(unittest.makeSuite(TestQuantizerTypes))
|
||
suite.addTest(unittest.makeSuite(TestModelMetadata))
|
||
|
||
if test_level in ["full", "all"]:
|
||
# 集成测试
|
||
suite.addTest(unittest.makeSuite(TestCompleteConversionPipeline))
|
||
suite.addTest(unittest.makeSuite(TestBatchConversion))
|
||
suite.addTest(unittest.makeSuite(TestConversionWithDifferentParameters))
|
||
|
||
# 错误处理测试
|
||
suite.addTest(unittest.makeSuite(TestErrorHandling))
|
||
suite.addTest(unittest.makeSuite(TestBoundaryConditions))
|
||
|
||
if test_level in ["performance", "all"]:
|
||
# 性能测试
|
||
suite.addTest(unittest.makeSuite(TestQuantizationPerformance))
|
||
suite.addTest(unittest.makeSuite(TestModelSizeAndCompression))
|
||
suite.addTest(unittest.makeSuite(TestConversionThroughput))
|
||
|
||
if test_level in ["quantization", "all"]:
|
||
# 全量化类型测试（30种量化类型）
|
||
suite.addTest(unittest.makeSuite(TestAllQuantizationTypes))
|
||
|
||
if test_level == "all":
|
||
# 高级测试（仅在完整测试时运行）
|
||
suite.addTest(unittest.makeSuite(TestModelCompatibility))
|
||
suite.addTest(unittest.makeSuite(TestResourceLimitations))
|
||
suite.addTest(unittest.makeSuite(TestConcurrencyAndThreadSafety))
|
||
suite.addTest(unittest.makeSuite(TestMemoryEfficiency))
|
||
|
||
return suite
|
||
|
||
def run_tests(self, test_level: str = "all", verbosity: int = 2) -> unittest.TestResult:
|
||
"""
|
||
运行测试套件
|
||
|
||
Args:
|
||
test_level: 测试级别
|
||
verbosity: 详细程度
|
||
"""
|
||
print(f"开始运行Netrans测试套件 (级别: {test_level})")
|
||
print("=" * 50)
|
||
|
||
suite = self.create_test_suite(test_level)
|
||
runner = unittest.TextTestRunner(
|
||
verbosity=verbosity,
|
||
stream=sys.stdout,
|
||
buffer=True
|
||
)
|
||
|
||
start_time = time.time()
|
||
result = runner.run(suite)
|
||
end_time = time.time()
|
||
|
||
# 生成测试摘要
|
||
self._generate_test_summary(result, end_time - start_time, test_level)
|
||
|
||
return result
|
||
|
||
def _generate_test_summary(self, result: unittest.TestResult,
|
||
execution_time: float, test_level: str):
|
||
"""生成测试摘要"""
|
||
print("\n" + "=" * 50)
|
||
print("测试执行摘要")
|
||
print("=" * 50)
|
||
|
||
total_tests = result.testsRun
|
||
failures = len(result.failures)
|
||
errors = len(result.errors)
|
||
skipped = len(result.skipped) if hasattr(result, 'skipped') else 0
|
||
passed = total_tests - failures - errors - skipped
|
||
|
||
print(f"测试级别: {test_level}")
|
||
print(f"总测试数: {total_tests}")
|
||
print(f"通过: {passed}")
|
||
print(f"失败: {failures}")
|
||
print(f"错误: {errors}")
|
||
print(f"跳过: {skipped}")
|
||
print(f"成功率: {(passed/total_tests*100):.1f}%" if total_tests > 0 else "0%")
|
||
print(f"执行时间: {execution_time:.2f} 秒")
|
||
|
||
# 详细的失败和错误信息
|
||
if failures:
|
||
print("\n失败的测试:")
|
||
for test, traceback in result.failures:
|
||
print(f" - {test}: {traceback.split('AssertionError:')[-1].strip()}")
|
||
|
||
if errors:
|
||
print("\n错误的测试:")
|
||
for test, traceback in result.errors:
|
||
print(f" - {test}: {traceback.split('Exception:')[-1].strip()}")
|
||
|
||
# 生成JSON报告
|
||
summary_data = {
|
||
'test_level': test_level,
|
||
'execution_time': execution_time,
|
||
'total_tests': total_tests,
|
||
'passed': passed,
|
||
'failures': failures,
|
||
'errors': errors,
|
||
'skipped': skipped,
|
||
'success_rate': (passed/total_tests*100) if total_tests > 0 else 0,
|
||
'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
|
||
'failed_tests': [str(test) for test, _ in result.failures],
|
||
'error_tests': [str(test) for test, _ in result.errors]
|
||
}
|
||
|
||
# 保存摘要到test目录
|
||
try:
|
||
import os
|
||
summary_file = self.test_dir / f"test_summary_{test_level}.json"
|
||
|
||
# 直接使用绝对路径写入，避免使用 Path.cwd() 和 .absolute()
|
||
# 因为 @chdir 装饰器可能导致当前工作目录被删除
|
||
with open(str(summary_file), 'w', encoding='utf-8') as f:
|
||
json.dump(summary_data, f, ensure_ascii=False, indent=2)
|
||
|
||
# 验证文件确实存在
|
||
if summary_file.exists():
|
||
file_size = summary_file.stat().st_size
|
||
print(f"\n✅ 详细报告已保存到: {summary_file}")
|
||
print(f"✅ 文件大小: {file_size} 字节")
|
||
else:
|
||
print(f"\n⚠️ 警告: 文件写入后找不到！")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 错误: 无法保存测试报告")
|
||
print(f"异常类型: {type(e).__name__}")
|
||
print(f"异常信息: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
parser = argparse.ArgumentParser(description="Netrans AI模型转换工具测试套件")
|
||
|
||
parser.add_argument(
|
||
'--level',
|
||
choices=['basic', 'full', 'performance', 'quantization', 'all'],
|
||
default='basic',
|
||
help='测试级别: basic(核心功能), full(集成测试), performance(性能测试), quantization(全量化类型), all(所有测试) (默认: basic)'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--verbosity',
|
||
type=int,
|
||
choices=[0, 1, 2],
|
||
default=2,
|
||
help='输出详细程度 (0=最少, 1=正常, 2=详细)'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--list-tests',
|
||
action='store_true',
|
||
help='列出所有可用的测试'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--specific-test',
|
||
type=str,
|
||
help='运行特定的测试类 (例如: TestNetransCore)'
|
||
)
|
||
|
||
parser.add_argument(
|
||
'--timeout',
|
||
type=int,
|
||
default=None,
|
||
help='测试超时时间（秒）'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 创建测试套件管理器
|
||
test_suite = NetransTestSuite()
|
||
|
||
if args.list_tests:
|
||
# 列出所有测试
|
||
print("可用的测试类:")
|
||
test_classes = [
|
||
"TestNetransCore - 核心功能测试",
|
||
"TestQuantizerTypes - 量化器类型测试",
|
||
"TestModelMetadata - 模型元数据测试",
|
||
"TestCompleteConversionPipeline - 完整转换流程测试",
|
||
"TestBatchConversion - 批量转换测试",
|
||
"TestConversionWithDifferentParameters - 不同参数转换测试",
|
||
"TestModelCompatibility - 模型兼容性测试",
|
||
"TestErrorHandling - 错误处理测试",
|
||
"TestBoundaryConditions - 边界条件测试",
|
||
"TestResourceLimitations - 资源限制测试",
|
||
"TestConcurrencyAndThreadSafety - 并发和线程安全测试",
|
||
"TestQuantizationPerformance - 量化性能测试",
|
||
"TestModelSizeAndCompression - 模型压缩测试",
|
||
"TestConversionThroughput - 转换吞吐量测试",
|
||
"TestMemoryEfficiency - 内存效率测试",
|
||
"TestAllQuantizationTypes - 全量化类型测试 (30种量化类型) ⭐ 新增"
|
||
]
|
||
|
||
for test_class in test_classes:
|
||
print(f" {test_class}")
|
||
|
||
return
|
||
|
||
if args.specific_test:
|
||
# 运行特定测试
|
||
try:
|
||
# 动态导入测试类
|
||
test_module = sys.modules[__name__]
|
||
test_class = getattr(test_module, args.specific_test)
|
||
|
||
suite = unittest.TestSuite()
|
||
suite.addTest(unittest.makeSuite(test_class))
|
||
|
||
runner = unittest.TextTestRunner(verbosity=args.verbosity)
|
||
result = runner.run(suite)
|
||
|
||
if result.wasSuccessful():
|
||
print(f"\n✅ {args.specific_test} 测试通过")
|
||
sys.exit(0)
|
||
else:
|
||
print(f"\n❌ {args.specific_test} 测试失败")
|
||
sys.exit(1)
|
||
|
||
except AttributeError:
|
||
print(f"错误: 未找到测试类 '{args.specific_test}'")
|
||
print("使用 --list-tests 查看可用的测试类")
|
||
sys.exit(1)
|
||
|
||
# 运行测试套件
|
||
try:
|
||
if args.timeout:
|
||
print(f"注意: 超时参数 {args.timeout} 秒已设置，但当前版本不支持强制超时。")
|
||
print("建议使用 Ctrl+C 手动停止过长的测试。\n")
|
||
|
||
result = test_suite.run_tests(
|
||
test_level=args.level,
|
||
verbosity=args.verbosity
|
||
)
|
||
|
||
# 根据测试结果设置退出码
|
||
if result.wasSuccessful():
|
||
print("\n✅ 所有测试通过！")
|
||
sys.exit(0)
|
||
else:
|
||
print("\n❌ 部分测试失败，请查看详细报告")
|
||
sys.exit(1)
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n测试被用户中断")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n测试执行出现错误: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main() |