diff --git a/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_compare.py b/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_compare.py new file mode 100644 index 0000000000000000000000000000000000000000..d6ad31094d81d8e950e69ecfb4cd85752a416261 --- /dev/null +++ b/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_compare.py @@ -0,0 +1,96 @@ +import unittest +from unittest.mock import patch, MagicMock +import torch +from ptdbg_ascend.online_dispatch.dump_compare import * + + +class TestDumpCompare(unittest.TestCase): + + def setUp(self): + self.run_param = DispatchRunParam( + debug_flag=True, + device_id=0, + root_npu_path='/npu/path', + root_cpu_path='/cpu/path', + process_num=1 + ) + + @patch('ptdbg_ascend.online_dispatch.dump_compare.logger_debug') + def test_TimeStatistics(self, mock_logger_debug): + with TimeStatistics("test_tag", self.run_param): + pass + mock_logger_debug.assert_called() + + @patch('ptdbg_ascend.online_dispatch.dump_compare.logger_user') + def test_save_summery(self, mock_logger_user): + npu_data = torch.tensor([1.0, 2.0]) + cpu_data = torch.tensor([1.0, 2.0]) + summery_list = [] + save_summery(self.run_param, npu_data, cpu_data, 'prefix', summery_list, True) + self.assertNotEqual(len(summery_list), 0) + + def test_support_basic_type(self): + self.assertTrue(support_basic_type(1)) + self.assertTrue(support_basic_type(1.0)) + self.assertTrue(support_basic_type(True)) + self.assertTrue(support_basic_type(torch.tensor([1.0]))) + + @patch('ptdbg_ascend.online_dispatch.dump_compare.np_save_data') + def test_dump_data(self, mock_np_save_data): + dump_data(torch.tensor([1.0]), 'prefix', '/path') + mock_np_save_data.assert_called_once() + + @patch('ptdbg_ascend.online_dispatch.dump_compare.save_summery') + def test_compare_data(self, mock_save_summery): + npu_data = torch.tensor([1.0]) + cpu_data = torch.tensor([1.0]) + summery_list = [] + result = compare_data(self.run_param, npu_data, cpu_data, 'prefix', summery_list, True) + self.assertTrue(result) + + @patch('ptdbg_ascend.online_dispatch.dump_compare.FileOpen') + @patch('ptdbg_ascend.online_dispatch.dump_compare.json.dump') + def test_save_temp_summery(self, mock_json_dump, mock_file_open): + lock = MagicMock() + save_temp_summery(0, [], '/path', lock) + mock_json_dump.assert_called_once() + + @patch('ptdbg_ascend.online_dispatch.dump_compare.getattr') + def test_get_torch_func(self, mock_getattr): + mock_getattr.return_value = MagicMock() + self.run_param.func_namespace = 'namespace' + self.run_param.aten_api = 'api' + self.run_param.aten_api_overload_name = 'overload' + result = get_torch_func(self.run_param) + self.assertTrue(result) + + @patch('ptdbg_ascend.online_dispatch.dump_compare.dispatch_workflow') + @patch('ptdbg_ascend.online_dispatch.dump_compare.get_torch_func') + def test_dispatch_multiprocess(self, mock_get_torch_func, mock_dispatch_workflow): + mock_get_torch_func.return_value = MagicMock() + cpu_args = (torch.tensor([1.0]),) + cpu_kwargs = {} + all_summery = [] + npu_out_cpu = torch.tensor([1.0]) + cpu_out = torch.tensor([1.0]) + lock = MagicMock() + dispatch_multiprocess(self.run_param, cpu_args, cpu_kwargs, all_summery, npu_out_cpu, cpu_out, lock) + mock_get_torch_func.assert_called_once() + + @patch('ptdbg_ascend.online_dispatch.dump_compare.logger_error') + def test_error_call(self, mock_logger_error): + error_call('test error') + mock_logger_error.assert_called_once_with('multiprocess test error') + + @patch('ptdbg_ascend.online_dispatch.dump_compare.pd.DataFrame') + @patch('ptdbg_ascend.online_dispatch.dump_compare.change_mode') + def test_save_csv(self, mock_change_mode, mock_data_frame): + all_summery = [[{'npu_name': 'test'}]] + call_stack_list = ['stack'] + csv_path = '/path/to/csv' + save_csv(all_summery, call_stack_list, csv_path) + mock_data_frame.assert_called() + mock_change_mode.assert_called_with(csv_path, '644') + + if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_util.py b/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_util.py new file mode 100644 index 0000000000000000000000000000000000000000..e233bdd03e377cd50e3f85e80ad957a63305855d --- /dev/null +++ b/debug/accuracy_tools/ptdbg_ascend/test/ut/online_dispatch/test_dispatch_util.py @@ -0,0 +1,95 @@ +import unittest +from unittest.mock import patch, MagicMock +import numpy as np +import torch +from ptdbg_ascend.online_dispatch.utils import * + +COLOR_YELLOW = '\033[33m' +COLOR_RED = '\033[31m' +COLOR_RESET = '\033[0m' + + +class TestUtils(unittest.TestCase): + + def test_get_callstack(self): + stack = get_callstack() + self.assertIsInstance(stack, list) + self.assertTrue(all(isinstance(item, list) for item in stack)) + + @patch('numpy.save') + def test_np_save_data(self, mock_save): + data = np.array([1, 2, 3]) + np_save_data(data, 'test', '/tmp') + mock_save.assert_called_once() + data = torch.tensor([1, 2, 3]) + np_save_data(data, 'test', '/tmp') + mock_save.assert_called() + + def test_data_to_cpu(self): + tensor = torch.tensor([1, 2, 3], device='npu:0') + result = data_to_cpu(tensor, 0, []) + self.assertEqual(result.device.type, 'cpu') + + @patch('logging.getLogger') + def test_get_mp_logger(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger = get_mp_logger() + self.assertTrue(callable(logger)) + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_debug(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_debug('test') + mock_logger.assert_called_once_with('DEBUG test') + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_info(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_info('test') + mock_logger.assert_called_once_with('INFO test') + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_warn(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_warn('test') + mock_logger.assert_called_once_with(f'{COLOR_YELLOW}WARNING test {COLOR_RESET}') + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_error(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_error('test') + mock_logger.assert_called_once_with(f'{COLOR_RED}ERROR test {COLOR_RESET}') + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_user(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_user('test') + mock_logger.assert_called_once_with('test') + + @patch('ptdbg_ascend.online_dispatch.utils.get_mp_logger') + def test_logger_logo(self, mock_get_logger): + mock_logger = MagicMock() + mock_get_logger.return_value = mock_logger + logger_logo() + mock_logger.assert_called_once() + + @patch('psutil.virtual_memory') + @patch('psutil.cpu_percent') + def test_get_sys_info(self, mock_cpu_percent, mock_virtual_memory): + mock_virtual_memory.return_value = MagicMock(total=8 * 1024 * 1024, available=4 * 1024 * 1024, used=4 * 1024 * 1024) + mock_cpu_percent.return_value = 50 + sys_info = get_sys_info() + self.assertIn('Total:', sys_info) + self.assertIn('Free:', sys_info) + self.assertIn('Used:', sys_info) + self.assertIn('CPU:', sys_info) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file