#!/usr/bin/env python3 """ DCU Performance Analyzer - 测试套件 用于验证工具的功能正确性和稳定性 """ import unittest import tempfile import shutil import json import os import sys from unittest.mock import patch, MagicMock # 添加父目录到路径 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dcu_performance_analyzer import DCUPerformanceAnalyzer, CheckResult class TestDCUPerformanceAnalyzer(unittest.TestCase): """DCU性能分析器测试类""" def setUp(self): """测试前设置""" self.test_dir = tempfile.mkdtemp() self.analyzer = DCUPerformanceAnalyzer(output_dir=self.test_dir, debug=True) def tearDown(self): """测试后清理""" shutil.rmtree(self.test_dir, ignore_errors=True) @patch('dcu_performance_analyzer.subprocess.run') def test_run_command_success(self, mock_run): """测试命令执行成功""" # 模拟成功的命令执行 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "test output" mock_result.stderr = "" mock_run.return_value = mock_result returncode, stdout, stderr = self.analyzer.run_command("echo test") self.assertEqual(returncode, 0) self.assertEqual(stdout, "test output") self.assertEqual(stderr, "") @patch('dcu_performance_analyzer.subprocess.run') def test_run_command_failure(self, mock_run): """测试命令执行失败""" # 模拟失败的命令执行 mock_result = MagicMock() mock_result.returncode = 1 mock_result.stdout = "" mock_result.stderr = "error message" mock_run.return_value = mock_result returncode, stdout, stderr = self.analyzer.run_command("false") self.assertEqual(returncode, 1) self.assertEqual(stdout, "") self.assertEqual(stderr, "error message") @patch('dcu_performance_analyzer.subprocess.run') def test_run_command_timeout(self, mock_run): """测试命令超时""" # 模拟超时异常 mock_run.side_effect = subprocess.TimeoutExpired("cmd", 1) returncode, stdout, stderr = self.analyzer.run_command("sleep 10", timeout=1) self.assertEqual(returncode, -1) self.assertEqual(stdout, "") self.assertEqual(stderr, "Command timeout") @patch('dcu_performance_analyzer.subprocess.run') def test_check_system_info(self, mock_run): """测试系统信息检查""" # 模拟系统命令输出 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "CPU info test" mock_result.stderr = "" mock_run.return_value = mock_result result = self.analyzer.check_system_info() self.assertIsInstance(result, CheckResult) self.assertEqual(result.module, "System Information") self.assertIn(result.status, ["PASS", "FAIL"]) self.assertIsInstance(result.execution_time, float) self.assertGreater(result.execution_time, 0) @patch('dcu_performance_analyzer.subprocess.run') def test_check_pcie_devices(self, mock_run): """测试PCIe设备检查""" # 模拟lspci命令输出 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "00:00.0 Host bridge: Intel test\n01:00.0 VGA compatible controller: 1d94:1234 DCU" mock_result.stderr = "" mock_run.return_value = mock_result result = self.analyzer.check_pcie_devices() self.assertIsInstance(result, CheckResult) self.assertEqual(result.module, "PCIe Devices") self.assertIn("dcu_devices", result.details) self.assertGreaterEqual(result.details["dcu_devices"], 0) @patch('dcu_performance_analyzer.subprocess.run') def test_check_driver_status(self, mock_run): """测试驱动状态检查""" # 模拟lsmod命令输出 mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "hydcu 12345 1 - Live 0xffffffffa0000000" mock_result.stderr = "" mock_run.return_value = mock_result result = self.analyzer.check_driver_status() self.assertIsInstance(result, CheckResult) self.assertEqual(result.module, "Driver Status") self.assertIn("loaded_modules", result.details) self.assertIsInstance(result.details["loaded_modules"], list) @patch('dcu_performance_analyzer.subprocess.run') def test_check_performance_metrics(self, mock_run): """测试性能指标检查""" # 模拟性能命令输出 def side_effect(cmd, **kwargs): mock_result = MagicMock() if "top" in cmd: mock_result.returncode = 0 mock_result.stdout = "%Cpu(s): 5.7 us" elif "free" in cmd: mock_result.returncode = 0 mock_result.stdout = "Mem: 1542486 54998 1487488 0 1234 23456" elif "df" in cmd: mock_result.returncode = 0 mock_result.stdout = "/dev/sda1 50G 45G 5G 90% /" elif "uptime" in cmd: mock_result.returncode = 0 mock_result.stdout = "14:30:00 up 1 day, 0:30, 1 user, load average: 2.17, 3.73, 4.72" else: mock_result.returncode = 0 mock_result.stdout = "" mock_result.stderr = "" return mock_result mock_run.side_effect = side_effect result = self.analyzer.check_performance_metrics() self.assertIsInstance(result, CheckResult) self.assertEqual(result.module, "Performance Metrics") self.assertIn("cpu_usage", result.details) self.assertIn("memory_usage_percent", result.details) self.assertIn("disk_usage_percent", result.details) self.assertIn("load_average_1m", result.details) def test_generate_report(self): """测试报告生成""" # 添加一些测试结果 test_result = CheckResult( module="Test Module", status="PASS", message="Test passed", details={"test": "data"}, timestamp="2026-02-25T12:00:00", execution_time=1.5 ) self.analyzer.results.append(test_result) report_file = self.analyzer.generate_report() self.assertTrue(os.path.exists(report_file)) # 验证JSON报告格式 with open(report_file, 'r') as f: report = json.load(f) self.assertIn("metadata", report) self.assertIn("summary", report) self.assertIn("results", report) self.assertEqual(report["summary"]["total_checks"], 1) self.assertEqual(report["summary"]["status_distribution"]["PASS"], 1) def test_create_package(self): """测试数据打包""" # 创建一些测试文件 test_file = os.path.join(self.test_dir, "test.txt") with open(test_file, 'w') as f: f.write("test content") package_file = self.analyzer.create_package() self.assertTrue(os.path.exists(package_file)) self.assertTrue(package_file.endswith(".tar.gz")) # 验证包内容 import tarfile with tarfile.open(package_file, 'r:gz') as tar: members = tar.getnames() self.assertGreater(len(members), 0) def test_result_to_dict(self): """测试结果转换""" result = CheckResult( module="Test", status="PASS", message="OK", details={"key": "value"}, timestamp="2026-02-25T12:00:00", execution_time=1.0 ) result_dict = self.analyzer._result_to_dict(result) self.assertIsInstance(result_dict, dict) self.assertEqual(result_dict["module"], "Test") self.assertEqual(result_dict["status"], "PASS") self.assertEqual(result_dict["message"], "OK") self.assertEqual(result_dict["details"]["key"], "value") self.assertEqual(result_dict["timestamp"], "2026-02-25T12:00:00") self.assertEqual(result_dict["execution_time"], 1.0) class TestCLI(unittest.TestCase): """命令行界面测试""" @patch('dcu_performance_analyzer.DCUPerformanceAnalyzer') def test_main_success(self, mock_analyzer_class): """测试主函数成功执行""" mock_analyzer = MagicMock() mock_analyzer.run_analysis.return_value = "test_package.tar.gz" mock_analyzer.results = [] mock_analyzer_class.return_value = mock_analyzer with patch('sys.argv', ['dcu_performance_analyzer.py']): with patch('sys.exit') as mock_exit: from dcu_performance_analyzer import main main() mock_exit.assert_called_with(0) @patch('dcu_performance_analyzer.DCUPerformanceAnalyzer') def test_main_with_checks(self, mock_analyzer_class): """测试指定检查模块""" mock_analyzer = MagicMock() mock_analyzer.run_analysis.return_value = "test_package.tar.gz" mock_analyzer.results = [] mock_analyzer_class.return_value = mock_analyzer with patch('sys.argv', ['dcu_performance_analyzer.py', '-c', 'system', 'pcie']): with patch('sys.exit') as mock_exit: from dcu_performance_analyzer import main main() mock_analyzer.run_analysis.assert_called_with(['system', 'pcie']) mock_exit.assert_called_with(0) def test_create_cli(self): """测试CLI创建""" from dcu_performance_analyzer import create_cli parser = create_cli() # 测试基本参数解析 args = parser.parse_args(['-c', 'system', '-o', '/tmp/test', '-d']) self.assertEqual(args.checks, ['system']) self.assertEqual(args.output, '/tmp/test') self.assertTrue(args.debug) if __name__ == '__main__': # 运行测试 unittest.main(verbosity=2)