AI assist developer for python dt v2.8.0-pytorch-AtomGit

Atest/profiler/analysis/prof_parse/test_cann_file_parser.py+160-0

@@ -0,0 +1,160 @@
	1	+from unittest.mock import patch, MagicMock
	2	+from torch_npu.profiler.analysis.prof_parse._cann_file_parser import CANNFileParser, CANNDataEnum
	3	+from torch_npu.testing.testcase import TestCase, run_tests
	4	+
	5	+
	6	+class TestFwkFileParser(TestCase):
	7	+
	8	+ def test_combine_acl_to_npu_with_amtching_events(self):
	9	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger:
	10	+ mock_logger.get_instance.return_value = MagicMock()
	11	+ mock_logger.init.return_value = None
	12	+
	13	+ timeline_data = [
	14	+ {"cat": "HostToDevice", "ph": "s", "id": 1, "ts": 1000},
	15	+ {"cat": "HostToDevice", "ph": "f", "id": 1, "ts": 2000, "pid": 1, "tid": 1},
	16	+ {"ph": "X", "pid": 1, "tid": 1, "ts": 2000, "name": "kernel_op"}
	17	+ ]
	18	+ result = CANNFileParser.combine_acl_to_npu(timeline_data)
	19	+ self.assertIsInstance(result, dict)
	20	+ self.assertIn(1000000, result)
	21	+ self.assertEqual(len(result[1000000]), 1)
	22	+
	23	+ def test_json_dict_load_empty_data(self):
	24	+ result = CANNFileParser._json_dict_load("")
	25	+ self.assertEqual(result, {})
	26	+
	27	+ def test_json_load_empty_data(self):
	28	+ result = CANNFileParser._json_load("")
	29	+ self.assertEqual(result, {})
	30	+
	31	+ def test_get_timeline_all_data_empty(self):
	32	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger, \
	33	+ patch(
	34	+ 'torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerPathManager') as mock_path_manager, \
	35	+ patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.FileManager') as mock_file_manager:
	36	+ mock_logger.get_instance.return_value = MagicMock()
	37	+ mock_path_manager.get_cann_path.return_value = "/test/path"
	38	+ mock_file_manager.check_file_readable.return_value = True
	39	+ mock_file_manager.check_file_writable.return_value = True
	40	+ mock_file_manager.file_read_all.return_value = ""
	41	+
	42	+ parser = CANNFileParser("/test/path")
	43	+ parser._file_dict = {CANNDataEnum.MSPROF_TIMELINE: set()}
	44	+
	45	+ result = parser.get_timeline_all_data()
	46	+ self.assertIsInstance(result, list)
	47	+ self.assertEqual(len(result), 0)
	48	+
	49	+ def test_json_dict_load_invalid_json_raises_runtime_error(self):
	50	+ with self.assertRaises(RuntimeError):
	51	+ CANNFileParser._json_dict_load("{invalid json}")
	52	+
	53	+ def test_json_dict_load_valid_dict(self):
	54	+ json_data = '{"key": "value", "key2": "value2"}'
	55	+ result = CANNFileParser._json_dict_load(json_data)
	56	+ self.assertEqual(result, {"key": "value", "key2": "value2"})
	57	+
	58	+ def test_json_load_invalid_json_raises_runtime_error(self):
	59	+ with self.assertRaises(RuntimeError):
	60	+ CANNFileParser._json_load("{invalid json}")
	61	+
	62	+ def test_json_load_valid_list(self):
	63	+ json_data = '[{"key": "value"}, {"key2": "value2"}]'
	64	+ result = CANNFileParser._json_load(json_data)
	65	+ self.assertEqual(result, [{"key": "value"}, {"key2": "value2"}])
	66	+
	67	+ def test_json_dict_load_non_dict_data(self):
	68	+ result = CANNFileParser._json_dict_load('["key", "value"]')
	69	+ self.assertEqual(result, {})
	70	+
	71	+ def test_json_load_non_list_data(self):
	72	+ result = CANNFileParser._json_load('{"key": "value"}')
	73	+ self.assertEqual(result, [])
	74	+
	75	+ def test_get_acl_to_npu_data_with_matching_events(self):
	76	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger, \
	77	+ patch(
	78	+ 'torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerPathManager') as mock_path_manager, \
	79	+ patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.FileManager') as mock_file_manager:
	80	+ mock_logger.get_instance.return_value = MagicMock()
	81	+ mock_logger.error.return_value = None
	82	+ mock_logger.warning.return_value = None
	83	+ mock_path_manager.get_cann_path.return_value = "/test/path"
	84	+ mock_file_manager.check_file_readable.return_value = True
	85	+ mock_file_manager.check_file_writable.return_value = True
	86	+ mock_file_manager.file_read_all.return_value = '[{"cat": "HostToDevice", "ph": "s", "id": 1, "ts": 1000}, {"cat": "HostToDevice", "ph": "f", "id": 1, "ts": 2000, "pid": 1, "tid": 1}, {"ph": "X", "pid": 1, "tid": 1, "ts": 2000, "name": "kernel_op"}]'
	87	+
	88	+ parser = CANNFileParser("/test/path")
	89	+ parser._file_dict = {CANNDataEnum.MSPROF_TIMELINE: {"/test/timeline.json"}}
	90	+
	91	+ result = parser.get_acl_to_npu_data()
	92	+ self.assertIsInstance(result, dict)
	93	+ self.assertIn(1000000, result)
	94	+ self.assertEqual(len(result[1000000]), 1)
	95	+
	96	+ def test_get_analyze_communication_data_with_file(self):
	97	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger, \
	98	+ patch(
	99	+ 'torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerPathManager') as mock_path_manager, \
	100	+ patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.FileManager') as mock_file_manager:
	101	+ mock_logger.get_instance.return_value = MagicMock()
	102	+ mock_path_manager.get_cann_path.return_value = "/test/path"
	103	+ mock_file_manager.check_file_readable.return_value = True
	104	+ mock_file_manager.check_file_writable.return_value = True
	105	+ mock_file_manager.file_read_all.return_value = '{"key": "value"}'
	106	+
	107	+ parser = CANNFileParser("/test/path")
	108	+ parser._file_dict = {CANNDataEnum.COMMUNICATION: {"/test/path/communication.json"}}
	109	+
	110	+ result = parser.get_analyze_communication_data(CANNDataEnum.COMMUNICATION)
	111	+ self.assertIsInstance(result, dict)
	112	+ self.assertEqual(result, {"key": "value"})
	113	+
	114	+ def test_get_timeline_all_data_with_non_empty_timeline(self):
	115	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger, \
	116	+ patch(
	117	+ 'torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerPathManager') as mock_path_manager, \
	118	+ patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.FileManager') as mock_file_manager:
	119	+ mock_logger.get_instance.return_value = MagicMock()
	120	+ mock_path_manager.get_cann_path.return_value = "/test/path"
	121	+ mock_file_manager.check_file_readable.return_value = True
	122	+ mock_file_manager.check_file_writable.return_value = True
	123	+ mock_file_manager.file_read_all.return_value = '[{"name": "test_event"}]'
	124	+
	125	+ parser = CANNFileParser("/test/path")
	126	+ parser._file_dict = {CANNDataEnum.MSPROF_TIMELINE: {"/test/path/msprof_123.json"}}
	127	+
	128	+ result = parser.get_timeline_all_data()
	129	+ self.assertIsInstance(result, list)
	130	+ self.assertEqual(len(result), 1)
	131	+ self.assertEqual(result[0]["name"], "test_event")
	132	+
	133	+ def test_combine_acl_to_npu_no_kernel_events(self):
	134	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger:
	135	+ mock_logger.get_instance.return_value = MagicMock()
	136	+ mock_logger.error.return_value = None
	137	+
	138	+ timeline_data = [
	139	+ {"cat": "HostToDevice", "ph": "s", "id": 1, "ts": 1000},
	140	+ {"cat": "HostToDevice", "ph": "s", "id": 1, "ts": 2000, "pid": 1, "tid": 1},
	141	+ ]
	142	+ result = CANNFileParser.combine_acl_to_npu(timeline_data)
	143	+ self.assertIsInstance(result, dict)
	144	+ self.assertEqual(len(result), 0)
	145	+
	146	+ def test_combine_acl_to_npu_no_flow_events(self):
	147	+ with patch('torch_npu.profiler.analysis.prof_parse._cann_file_parser.ProfilerLogger') as mock_logger:
	148	+ mock_logger.get_instance.return_value = MagicMock()
	149	+ mock_logger.warning.return_value = None
	150	+
	151	+ timeline_data = [
	152	+ {"ph": "X", "pid": 1, "tid": 1, "ts": 2000, "name": "kernel_op"},
	153	+ ]
	154	+ result = CANNFileParser.combine_acl_to_npu(timeline_data)
	155	+ self.assertIsInstance(result, dict)
	156	+ self.assertEqual(len(result), 0)
	157	+
	158	+
	159	+if __name__ == '__main__':
	160	+ run_tests()

Atest/profiler/analysis/prof_view/prepare_parser/test_basic_db_parser.py+141-0

@@ -0,0 +1,141 @@
	1	+from unittest.mock import patch
	2	+from torch_npu.testing.testcase import TestCase, run_tests
	3	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant
	4	+from torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser import BasicDbParser
	5	+
	6	+
	7	+class TestBasicDbParser(TestCase):
	8	+
	9	+ def setUp(self):
	10	+ self.test_dir = "temp"
	11	+ self.param_dict = {
	12	+ "profiler_path": self.test_dir,
	13	+ "output_path": self.test_dir
	14	+ }
	15	+
	16	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	17	+ return_value=None)
	18	+ def test_basic_db_parser_run_no_cann_db(self, mock_get_cann_path):
	19	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	20	+
	21	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.TorchDb') as mock_db:
	22	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	23	+ mock_db_instance = mock_db.return_value
	24	+ mock_db_instance.create_connect_db.return_value = True
	25	+
	26	+ mock_db_instance.judge_table_exist.return_value = False
	27	+ mock_db_instance.create_table_with_headers.return_value = None
	28	+ mock_db_instance.insert_data_into_table.return_value = None
	29	+
	30	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerConfig') as mock_config:
	31	+ mock_config_instance = mock_config.return_value
	32	+ mock_config_instance.rank_id = 0
	33	+
	34	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerPathManager.get_device_id') as mock_device_id:
	35	+ mock_device_id.return_value = [0]
	36	+
	37	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.get_host_info') as mock_host_info:
	38	+ mock_host_info.return_value = {"host_uid": "uid1", "host_name": "host1"}
	39	+
	40	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.collect_env_vars') as mock_env_vars:
	41	+ mock_env_vars.return_value = {"ENV_VARIABLES": {"key1": "value1"}}
	42	+
	43	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.exists') as mock_exists:
	44	+ mock_exists.return_value = False
	45	+ status, result = parser.run({})
	46	+ self.assertEqual(status, Constant.SUCCESS)
	47	+ self.assertEqual(result, "")
	48	+
	49	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	50	+ return_value="/mock/cann/path")
	51	+ def test_get_cann_db_path_no_valid_files(self, mock_get_cann_path):
	52	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	53	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.listdir') as mock_listdir:
	54	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.exists') as mock_exists:
	55	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.join') as mock_join:
	56	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.re.match') as mock_match:
	57	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.FileManager.check_db_file_vaild') as mock_check:
	58	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	59	+ mock_listdir.return_value = ["invalid_file.txt"]
	60	+ mock_exists.return_value = True
	61	+
	62	+ def mock_join_side_effect(x, y):
	63	+ return f"{x}/{y}"
	64	+
	65	+ mock_join.side_effect = mock_join_side_effect
	66	+ mock_match.return_value = None
	67	+ mock_check.side_effect = RuntimeError("Invalid file")
	68	+ result = parser.get_cann_db_path()
	69	+ self.assertEqual(result, "")
	70	+
	71	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	72	+ return_value="/mock/cann/path")
	73	+ def test_save_profiler_metadata_to_db_json_error(self, mock_get_cann_path):
	74	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	75	+
	76	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.exists') as mock_exists:
	77	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.FileManager.file_read_all') as mock_read:
	78	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	79	+ mock_exists.return_value = True
	80	+ mock_read.return_value = '{"invalid": json}'
	81	+ parser.save_profiler_metadata_to_db()
	82	+
	83	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	84	+ return_value="/mock/cann/path")
	85	+ def test_save_rank_info_to_db_multiple_devices(self, mock_get_cann_path):
	86	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	87	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerConfig') as mock_config:
	88	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	89	+ mock_config_instance = mock_config.return_value
	90	+ mock_config_instance.rank_id = 0
	91	+
	92	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerPathManager.get_device_id') as mock_device_id:
	93	+ mock_device_id.return_value = [0, 1]
	94	+
	95	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.TorchDb') as mock_db:
	96	+ mock_db_instance = mock_db.return_value
	97	+ mock_db_instance.create_table_with_headers.return_value = None
	98	+ mock_db_instance.insert_data_info_table.return_value = None
	99	+
	100	+ parser.save_rank_info_to_db()
	101	+
	102	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	103	+ return_value="/mock/cann/path")
	104	+ def test_get_cann_db_path_from_mindstudio_output(self, mock_get_cann_path):
	105	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	106	+
	107	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.listdir') as mock_listdir:
	108	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.exists') as mock_exists:
	109	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.os.path.join') as mock_join:
	110	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.re.match') as mock_match:
	111	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.FileManager.check_db_file_vaild') as mock_check:
	112	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	113	+ mock_listdir.side_effect = [["invalid_file.txt"], ["msprof_123.db"]]
	114	+ mock_exists.return_value = True
	115	+
	116	+ def mock_join_side_effect(x, y):
	117	+ return f"{x}/{y}"
	118	+
	119	+ mock_join.side_effect = mock_join_side_effect
	120	+ mock_match.side_effect = [None, True]
	121	+ mock_check.return_value = None
	122	+
	123	+ result = parser.get_cann_db_path()
	124	+ self.assertEqual(result, "/mock/cann/path/mindstudio_profiler_output/msprof_123.db")
	125	+
	126	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	127	+ return_value="/mock/cann/path")
	128	+ def test_basic_db_parser_run_db_connection_failure(self, mock_get_cann_path):
	129	+ parser = BasicDbParser("test_basic_db", self.param_dict)
	130	+
	131	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.TorchDb') as mock_db:
	132	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._basic_db_parser.ProfilerLogger'):
	133	+ mock_db_instance = mock_db.return_value
	134	+ mock_db_instance.create_connect_db.return_value = False
	135	+
	136	+ status, result = parser.run({})
	137	+ self.assertEqual(status, Constant.FAIL)
	138	+ self.assertEqual(result, "")
	139	+
	140	+if __name__ == '__main__':
	141	+ run_tests()

Atest/profiler/analysis/prof_view/test_cann_analyze.py+59-0

@@ -0,0 +1,59 @@
	1	+import unittest
	2	+from unittest.mock import patch
	3	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant
	4	+from torch_npu.profiler.analysis.prof_view.cann_parse._cann_analyze import CANNAnalyzeParser
	5	+
	6	+
	7	+def run_test():
	8	+ loader = unittest.TestLoader()
	9	+ suite = unittest.TestSuite()
	10	+ suite.addTests(loader.loadTestsFromTestCase(TestCannAnalyze))
	11	+ runner = unittest.TextTestRunner(verbosity=2)
	12	+ runner.run(suite)
	13	+
	14	+
	15	+class TestCannAnalyze(unittest.TestCase):
	16	+
	17	+ def test_cann_analyze_parser_run_exception(self):
	18	+ param_dict = {"profiler_path": "/tmp", "export_type": ["db"]}
	19	+ with patch('torch_npu.profiler.analysis.prof_view.cann_parse._cann_analyze.ProfilerConfig') as mock_config:
	20	+ mock_config.side_effect = Exception("Test exception")
	21	+ parser = CANNAnalyzeParser("test_parser", param_dict)
	22	+ with patch('torch_npu.profiler.analysis.prof_view.cann_parse._cann_analyze.print_error_msg') as mock_error:
	23	+ result = parser.run({})
	24	+ self.assertEqual(result[0], Constant.FAIL)
	25	+ mock_error.assert_called_with("Failed to analyze CANN Profiling data.")
	26	+
	27	+ def test_cann_analyze_parser_run_db_success(self):
	28	+ param_dict = {"profiler_path": "/tmp", "export_type": ["db"]}
	29	+ with patch('shutil.which') as mock_which:
	30	+ mock_which.return_value = "/usr/bin/msprof"
	31	+ with patch('os.path.isdir') as mock_isdir:
	32	+ mock_isdir.return_value = True
	33	+ with patch('subprocess.run') as mock_run:
	34	+ mock_run.return_value.returncode = 0
	35	+ parser = CANNAnalyzeParser("test_parser", param_dict)
	36	+ result = parser.run({})
	37	+ self.assertEqual(result[0], Constant.SUCCESS)
	38	+
	39	+ def test_cann_analyze_parser_run_no_cann_path(self):
	40	+ param_dict = {"profiler_path": "/tmp", "export_type": ["db", "text"]}
	41	+ with patch('torch_npu.profiler.analysis.prof_view.cann_parse._cann_analyze.ProfilerPathManager.get_cann_path') as mock_get_path:
	42	+ mock_get_path.return_value = "/nonexistent/path"
	43	+ with patch("os.path.isdir") as mock_isdir:
	44	+ mock_isdir.return_value = False
	45	+ parser = CANNAnalyzeParser("test_parser", param_dict)
	46	+ result = parser.run({})
	47	+ self.assertEqual(result[0], Constant.SUCCESS)
	48	+
	49	+ def test_cann_analyze_parser_init(self):
	50	+ param_dict = {"profiler_path": "/tmp", "export_type": ["db", "text"]}
	51	+ parser = CANNAnalyzeParser("test_parser", param_dict)
	52	+ self.assertEqual(parser._name, "test_parser")
	53	+ self.assertEqual(parser._param_dict, param_dict)
	54	+ self.assertIsNotNone(parser._cann_path)
	55	+ self.assertIsNotNone(parser.msprof_path)
	56	+
	57	+
	58	+if __name__ == "__main__":
	59	+ run_test()

Atest/profiler/analysis/prof_view/test_cann_export.py+36-0

@@ -0,0 +1,36 @@
	1	+import unittest
	2	+from unittest.mock import patch
	3	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant
	4	+from torch_npu.profiler.analysis.prof_view.cann_parse._cann_export import CANNExportParser
	5	+
	6	+
	7	+def run_test():
	8	+ loader = unittest.TestLoader()
	9	+ suite = unittest.TestSuite()
	10	+ suite.addTests(loader.loadTestsFromTestCase(TestCannExport))
	11	+
	12	+ runner = unittest.TextTestRunner(verbosity=2)
	13	+ runner.run(suite)
	14	+
	15	+
	16	+class TestCannExport(unittest.TestCase):
	17	+ def test_cann_export_parser_run_db_export_success(self):
	18	+ with patch("os.patch.isdir", return_value=True), \
	19	+ patch("subprocess.run") as mock_run, \
	20	+ patch('from torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path', return_value="/fake/cann/path"):
	21	+ mock_run.return_value.returncode = 0
	22	+ parser = CANNExportParser("test", {"profiler_path": "/fake/path", "export_type": ["db"]})
	23	+ parser.msprof_path = "/usr/bin/msprof"
	24	+ result = parser.run({})
	25	+ self.assertEqual(result, (Constant.SUCCESS, None))
	26	+
	27	+ def test_cann_export_parser_init(self):
	28	+ parser = CANNExportParser("test", {"profiler_path": "/fake/path", "export_type": ["db"]})
	29	+ self.assertEqual(parser._profiler_path, "/fake/path")
	30	+ self.assertEqual(parser._export_type, "db")
	31	+ self.assertIsNotNone(parser._cann_path)
	32	+ self.assertIsNotNone(parser.msprof_path)
	33	+
	34	+
	35	+if __name__ == "__main__":
	36	+ run_test()

Atest/testing/test_common_utils.py+190-0

@@ -0,0 +1,190 @@
	1	+import os
	2	+from unittest.mock import patch
	3	+
	4	+import numpy as np
	5	+import torch
	6	+
	7	+import torch_npu
	8	+from torch_npu.testing.testcase import TestCase, run_tests
	9	+from torch_npu.testing.common_utils import (
	10	+ freeze_rng_state,
	11	+ iter_indices,
	12	+ is_iterable,
	13	+ get_npu_device,
	14	+ create_common_tensor,
	15	+ test_2args_broadcast,
	16	+ create_dtype_tensor,
	17	+ check_operators_in_prof,
	18	+ _create_scaling_case
	19	+)
	20	+
	21	+
	22	+class TestCommonUtils(TestCase):
	23	+
	24	+ def test_iter_indices_zero_dim(self):
	25	+ zero_dim_tensor = torch.tensor(5)
	26	+ indices = list(iter_indices(zero_dim_tensor))
	27	+ self.assertEqual(indices, [])
	28	+
	29	+ def test_create_scaling_case_dtype(self):
	30	+ mod_control, mod_scaling, opt_control, opt_scaling, data, loss_fn, skip_iter = _create_scaling_case(
	31	+ device="npu", dtype=torch.float
	32	+ )
	33	+ self.assertIsNotNone(mod_control)
	34	+ self.assertIsNotNone(mod_scaling)
	35	+ self.assertIsNotNone(opt_control)
	36	+ self.assertIsNotNone(opt_scaling)
	37	+ self.assertIsNotNone(data)
	38	+ self.assertIsNotNone(loss_fn)
	39	+ self.assertEqual(skip_iter, 2)
	40	+
	41	+ for input_data, target_data in data:
	42	+ self.assertEqual(input_data.dtype, torch.float)
	43	+ self.assertEqual(target_data.dtype, torch.float)
	44	+
	45	+ def test_check_operators_in_prof(self):
	46	+ class MockProf:
	47	+ class MockItem:
	48	+ def __init__(self, key):
	49	+ self.key = key
	50	+
	51	+ def key_averages(self):
	52	+ return[self.MockItem("add"), self.MockItem("mul")]
	53	+
	54	+ expected = ["add", "mul"]
	55	+ result = check_operators_in_prof(expected, MockProf())
	56	+ self.assertTrue(result)
	57	+
	58	+ unexpected = ["add", "mul"]
	59	+ result = check_operators_in_prof(["add"], MockProf(), unexpected)
	60	+ self.assertFalse(result)
	61	+
	62	+ def test_create_dtype_tensor_no_zero(self):
	63	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.int32, no_zero=True)
	64	+ self.assertEqual(cpu_input.shape, (2, 3))
	65	+ self.assertEqual(npu_input.shape, (2, 3))
	66	+
	67	+ self.assertFalse(torch.any(cpu_input == 0))
	68	+ self.assertFalse(torch.any(npu_input == 0))
	69	+
	70	+ def test_iter_indices_2d(self):
	71	+ tensor = torch.tensor([[1, 2], [3, 4]])
	72	+ indices = list(iter_indices(tensor))
	73	+ self.assertEqual(indices, [(0, 0), (0, 1), (1, 0), (1, 1)])
	74	+
	75	+ def test_get_npu_device_with_env_vat(self):
	76	+ with patch.dict(os.environ, {"SET_NPU_DEVICE": "1"}, clear=True):
	77	+ device = get_npu_device()
	78	+ self.assertEqual(device, "npu:1")
	79	+
	80	+ def test_create_common_tensor(self):
	81	+ item = (np.float32, -1, (2, 3))
	82	+ cpu_input, npu_input = create_common_tensor(item, -5, 5)
	83	+
	84	+ self.assertIsInstance(cpu_input, torch.Tensor)
	85	+ self.assertIsInstance(npu_input, torch.Tensor)
	86	+ self.assertEqual(cpu_input.shape, (2, 3))
	87	+ self.assertEqual(npu_input.shape, (2, 3))
	88	+ self.assertEqual(cpu_input.dtype, torch.float32)
	89	+ self.assertEqual(npu_input.dtype, torch.float32)
	90	+
	91	+ def test_is_iterable(self):
	92	+ self.assertTrue(is_iterable([1, 2, 3]))
	93	+ self.assertTrue(is_iterable((1, 2, 3)))
	94	+ self.assertTrue(is_iterable({1, 2, 3}))
	95	+ self.assertTrue(is_iterable("hello"))
	96	+ self.assertFalse(is_iterable(42))
	97	+ self.assertFalse(is_iterable(None))
	98	+
	99	+ def test_iter_indices_1d(self):
	100	+ tensor_1d = torch.tensor([1, 2, 3, 4])
	101	+ indices = list(iter_indices(tensor_1d))
	102	+ self.assertEqual(indices, [0, 1, 2, 3])
	103	+
	104	+ def test_torch_manual_seed_seeds_npu_devices(self):
	105	+ with freeze_rng_state():
	106	+ x = torch.zeros(4, 4).float()
	107	+ torch.manual_seed(2)
	108	+ self.assertEqual(torch_npu.npu.initial_seed(), 2)
	109	+ x.uniform_()
	110	+ torch.manual_seed(2)
	111	+ y = x.clone().uniform_()
	112	+ self.assertEqual(x, y)
	113	+ self.assertEqual((torch_npu.npu.initial_seed()), 2)
	114	+
	115	+ def test_manual_seed(self):
	116	+ with freeze_rng_state():
	117	+ x = torch.zeros(4, 4).float()
	118	+ torch_npu.npu.manual_seed(2)
	119	+ torch.manual_seed(2)
	120	+ self.assertEqual(torch_npu.npu.initial_seed(), 2)
	121	+ x.uniform_()
	122	+ a = torch.bernoulli(torch.full_like(x, 0.5))
	123	+ torch.manual_seed(2)
	124	+ y = x.clone().uniform_()
	125	+ b = torch.bernoulli(torch.full_like(x, 0.5))
	126	+ self.assertEqual(x, y)
	127	+ self.assertEqual(a, b)
	128	+ self.assertEqual(torch_npu.npu.initial_seed(), 2)
	129	+
	130	+ def test_get_set_rng_state(self):
	131	+ with freeze_rng_state():
	132	+ torch.manual_seed(3)
	133	+ cpu_state = torch.get_rng_state()
	134	+ npu_state = torch_npu.npu.get_rng_state()
	135	+ self.assertEqual(int(cpu_state[0]), 3)
	136	+ self.assertEqual(cpu_state[0], npu_state[0])
	137	+ torch_npu.npu.manual_seed(2)
	138	+ cpu_state_new = torch.get_rng_state()
	139	+ npu_state = torch_npu.npu.get_rng_state()
	140	+ self.assertEqual(cpu_state, cpu_state_new)
	141	+ self.assertEqual(int(npu_state[0]), 2)
	142	+
	143	+ def test_create_dtype_tensor_with_format(self):
	144	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.float, npu_format=2)
	145	+ self.assertEqual(cpu_input.shape, (2, 3))
	146	+ self.assertEqual(npu_input.shape, (2, 3))
	147	+ self.assertEqual(cpu_input.dtype, torch.float)
	148	+ self.assertEqual(npu_input.dtype, torch.float)
	149	+
	150	+ def test_create_dtype_tensor_bool(self):
	151	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.bool)
	152	+ self.assertEqual(cpu_input.shape, (2, 3))
	153	+ self.assertEqual(npu_input.shape, (2, 3))
	154	+ self.assertEqual(cpu_input.dtype, torch.bool)
	155	+ self.assertEqual(npu_input.dtype, torch.bool)
	156	+ self.assertTrue(torch.isfinite(cpu_input).all())
	157	+
	158	+ def test_check_operators_in_prof_empty_prof(self):
	159	+ class MockEmptyProf:
	160	+ def key_averages(self):
	161	+ return []
	162	+ expected = ["add"]
	163	+ result = check_operators_in_prof(expected, MockEmptyProf())
	164	+ self.assertFalse(result)
	165	+
	166	+ def test_create_dtype_tensor_different_dtypes(self):
	167	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.half)
	168	+ self.assertEqual(cpu_input.dtype, torch.float16)
	169	+ self.assertEqual(npu_input.dtype, torch.float16)
	170	+
	171	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.int32)
	172	+ self.assertEqual(cpu_input.dtype, torch.int32)
	173	+ self.assertEqual(npu_input.dtype, torch.int32)
	174	+
	175	+ cpu_input, npu_input = create_dtype_tensor((2, 3), torch.float32)
	176	+ self.assertEqual(cpu_input.dtype, torch.float32)
	177	+ self.assertEqual(npu_input.dtype, torch.float32)
	178	+
	179	+ def test_2args_broadcast(self):
	180	+ def add_fn(x, y):
	181	+ result = test_2args_broadcast()
	182	+ self.assertEqual(len(result), 2)
	183	+
	184	+ for cpu_out, npu_out in result:
	185	+ self.assertIsNotNone(cpu_out)
	186	+ self.assertIsNotNone(npu_out)
	187	+
	188	+
	189	+if __name__ == "__main__":
	190	+ run_tests()

Atest/utils/test_affinity.py+71-0

@@ -0,0 +1,71 @@
	1	+import torch_npu
	2	+from torch_npu.testing.testcase import TestCase, run_tests
	3	+from torch_npu.utils.affinity import _set_thread_affinity, _reset_thread_affinity
	4	+
	5	+
	6	+class TestAffinity(TestCase):
	7	+
	8	+ def test_reset_thread_affinity(self):
	9	+ original_func = torch_npu._C._npu_reset_thread_affinity
	10	+ call_count = 0
	11	+
	12	+ def mock_npu_reset_thread_affinity():
	13	+ nonlocal call_count
	14	+ call_count += 1
	15	+
	16	+ torch_npu._C._npu_reset_thread_affinity = mock_npu_reset_thread_affinity
	17	+ try:
	18	+ _reset_thread_affinity()
	19	+ self.assertEqual(call_count, 1)
	20	+ finally:
	21	+ torch_npu._C._npu_reset_thread_affinity = original_func
	22	+
	23	+
	24	+ def test_set_thread_affinity_invalid_length(self):
	25	+ with self.assertRaises(ValueError) as context:
	26	+ _set_thread_affinity([1, 2, 3])
	27	+ self.assertIn("The length of input list of set_thread_affinity should be 2", str(context.exception))
	28	+
	29	+ with self.assertRaises(ValueError) as context:
	30	+ _set_thread_affinity([])
	31	+ self.assertIn("The length of input list of set_thread_affinity should be 2", str(context.exception))
	32	+
	33	+ def test_set_thread_affinity_negative_values(self):
	34	+ with self.assertRaises(ValueError) as context:
	35	+ _set_thread_affinity([-1, 5])
	36	+ self.assertIn("Core range should be nonnegative", str(context.exception))
	37	+
	38	+ with self.assertRaises(ValueError) as context:
	39	+ _set_thread_affinity([2, -3])
	40	+ self.assertIn("Core range should be nonnegative", str(context.exception))
	41	+
	42	+ def test_set_thread_affinity_valid_range(self):
	43	+ original_func = torch_npu._C._npu_set_thread_affinity
	44	+ call_args = []
	45	+
	46	+ def mock_npu_set_thread_affinity(start, end):
	47	+ call_args.append((start, end))
	48	+
	49	+ torch_npu._C._npu_set_thread_affinity = mock_npu_set_thread_affinity
	50	+ try:
	51	+ _set_thread_affinity([2, 5])
	52	+ self.assertEqual(call_args, [(2, 5)])
	53	+ finally:
	54	+ torch_npu._C._npu_set_thread_affinity = original_func
	55	+
	56	+ def test_set_thread_affinity_none(self):
	57	+ original_func = torch_npu._C._npu_set_thread_affinity
	58	+ call_args = []
	59	+
	60	+ def mock_npu_set_thread_affinity(start, end):
	61	+ call_args.append((start, end))
	62	+
	63	+ torch_npu._C._npu_set_thread_affinity = mock_npu_set_thread_affinity
	64	+ try:
	65	+ _set_thread_affinity(None)
	66	+ self.assertEqual(call_args, [(-1, -1)])
	67	+ finally:
	68	+ torch_npu._C._npu_set_thread_affinity = original_func
	69	+
	70	+if __name__ == '__main__':
	71	+ run_tests()

Atest/utils/test_asd_detector.py+60-0

@@ -0,0 +1,60 @@
	1	+import torch
	2	+import torch_npu
	3	+from torch_npu.testing.testcase import TestCase, run_tests
	4	+from torch_npu.utils._asd_detector import set_asd_loss_scale, register_asd_hook
	5	+
	6	+
	7	+class AsdDetector(TestCase):
	8	+
	9	+ def test_register_asd_hook_with_conditions(self):
	10	+ original_func = torch_npu._C._get_silent_check_version
	11	+
	12	+ def mock_get_silent_check_version():
	13	+ return 1
	14	+
	15	+ torch_npu._C._get_silent_check_version = mock_get_silent_check_version
	16	+
	17	+ try:
	18	+ x = torch.tensor([1.0, 2.0], requires_grad=True)
	19	+ weight = torch.tensor([1.0])
	20	+ self.assertIsNone(x._backward_hooks)
	21	+ result = register_asd_hook(x, weight)
	22	+
	23	+ self.assertIsNone(result)
	24	+ self.assertIsNotNone(x._backward_hooks)
	25	+ finally:
	26	+ torch_npu._C._get_silent_check_version = original_func
	27	+
	28	+ def test_register_asd_hook_early_return(self):
	29	+ original_func = torch_npu._C._get_silent_check_version
	30	+
	31	+ def mock_get_silent_check_version():
	32	+ return 2
	33	+
	34	+ torch_npu._C._get_silent_check_version = mock_get_silent_check_version
	35	+
	36	+ try:
	37	+ x = torch.tensor([1.0, 2.0])
	38	+ weight = torch.tensor([1.0])
	39	+ result = register_asd_hook(x, weight)
	40	+ self.assertIsNone(result)
	41	+ finally:
	42	+ torch_npu._C._get_silent_check_version = original_func
	43	+
	44	+ def test_set_asd_loss_scale_early_return(self):
	45	+ original_func = torch_npu._C._get_silent_check_version
	46	+
	47	+ def mock_get_silent_check_version():
	48	+ return 2
	49	+
	50	+ torch_npu._C._get_silent_check_version = mock_get_silent_check_version
	51	+
	52	+ try:
	53	+ result = set_asd_loss_scale(2.0)
	54	+ self.assertIsNone(result)
	55	+ finally:
	56	+ torch_npu._C._get_silent_check_version = original_func
	57	+
	58	+
	59	+if __name__ == '__main__':
	60	+ run_tests()

Atest/utils/test_collet_env.py+157-0

@@ -0,0 +1,157 @@
	1	+import os
	2	+import site
	3	+import tempfile
	4	+import torch
	5	+
	6	+from torch.utils import collect_env as torch_collect_env
	7	+
	8	+from torch_npu.testing.testcase import TestCase, run_tests
	9	+
	10	+try:
	11	+ import torch_npu
	12	+
	13	+ TORCH_AVAILABLE = True
	14	+except (ImportError, NameError, AttributeError, OSError):
	15	+ TORCH_AVAILABLE = False
	16	+
	17	+try:
	18	+ import torch_npu
	19	+ TORCH_NPU_AVAILABLE = True
	20	+except (ImportError, NameError, AttributeError, OSError):
	21	+ TORCH_NPU_AVAILABLE = False
	22	+
	23	+from torch_npu.utils.collect_env import (
	24	+ SystemEnv,
	25	+ get_torch_npu_install_path,
	26	+ check_path_owner_consistent,
	27	+ check_directory_path_readable,
	28	+ get_torch_npu_version,
	29	+ get_env_info,
	30	+ pretty_str
	31	+)
	32	+
	33	+
	34	+class TestCollectEnv(TestCase):
	35	+
	36	+ def test_pretty_str_with_none_and_empty(self):
	37	+ env_info = SystemEnv(
	38	+ torch_version='1.0.0',
	39	+ torch_npu_version='1.0.0',
	40	+ is_debug_build='False',
	41	+ gcc_version='9.3.0',
	42	+ clang_version='10.0.0',
	43	+ cmake_version='3.16.0',
	44	+ os="Linux",
	45	+ libc_version='2.27',
	46	+ python_version='3.8.0 (64-big runtime)',
	47	+ python_platform='Linux',
	48	+ pip_version='pip',
	49	+ pip_packages='',
	50	+ conda_packages='',
	51	+ caching_allocator_config='default',
	52	+ is_xnnpack_available=True,
	53	+ cpu_info='Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz',
	54	+ cann_version='not known'
	55	+ )
	56	+
	57	+ result = pretty_str(env_info)
	58	+ self.assertIn('PyTorch version: 1.0.0', result)
	59	+ self.assertIn('CANN:', result)
	60	+ self.assertIn('not known', result)
	61	+
	62	+ def test_check_directory_path_readable_symlink(self):
	63	+ with tempfile.TemporaryDirectory() as temp_dir:
	64	+ symlink_path = os.path.join(temp_dir, "symlink_to_dir")
	65	+ os.symlink(temp_dir, symlink_path)
	66	+
	67	+ with self.assertRaises(RuntimeError) as context:
	68	+ check_directory_path_readable(symlink_path)
	69	+ self.assertIn("Invalid path is a soft chain", str(context.exception))
	70	+
	71	+ def test_get_env_info_torch_not_available(self):
	72	+ import torch_npu.utils.collect_env as collect_env_module
	73	+ original_torch_available = collect_env_module.TORCH_AVAILABLE
	74	+ original_torch = collect_env_module.torch
	75	+
	76	+ collect_env_module.TORCH_AVAILABLE = False
	77	+ collect_env_module.torch = None
	78	+
	79	+ try:
	80	+ result = get_env_info()
	81	+ self.assertEqual(result.torch_version, 'N/A')
	82	+ self.assertEqual(result.is_debug_build, 'N/A')
	83	+ finally:
	84	+ collect_env_module.TORCH_AVAILABLE = original_torch_available
	85	+ collect_env_module.torch = original_torch
	86	+
	87	+ def test_get_torch_npu_varsion_npu_availabel(self):
	88	+ import torch_npu.utils.collect_env as collect_env_module
	89	+ original_torch_npu_available = collect_env_module.TORCH_NPU_AVAILABLE
	90	+ original_torch_npu = collect_env_module.torch_npu
	91	+
	92	+ collect_env_module.TORCH_NPU_AVAILABLE = False
	93	+ collect_env_module.torch_npu = None
	94	+
	95	+ try:
	96	+ result = get_torch_npu_version()
	97	+ self.assertEqual(result, 'N/A')
	98	+ finally:
	99	+ collect_env_module.TORCH_NPU_AVAILABLE = original_torch_npu_available
	100	+ collect_env_module.torch_npu = original_torch_npu
	101	+
	102	+ def test_check_path_owner_consistent_nonexistent_path(self):
	103	+ with self.assertRaises(RuntimeError) as context:
	104	+ check_path_owner_consistent("/non/existent/path")
	105	+ self.assertIn("The path does not exist", str(context.exception))
	106	+
	107	+ def test_get_torch_npu_install_path_empty_site_packages(self):
	108	+ original_getsitepackages = site.getsitepackages
	109	+
	110	+ def mock_getsitepackages():
	111	+ return []
	112	+
	113	+ site.getsitepackages = mock_getsitepackages
	114	+
	115	+ try:
	116	+ result = get_torch_npu_install_path()
	117	+ self.assertEqual(result, "")
	118	+ finally:
	119	+ site.getsitepackages = original_getsitepackages
	120	+
	121	+ def test_pretty_str_multiline_cpu_info(self):
	122	+ env_info = SystemEnv(
	123	+ torch_version='1.0.0',
	124	+ torch_npu_version='1.0.0',
	125	+ is_debug_build='False',
	126	+ gcc_version='9.3.0',
	127	+ clang_version='10.0.0',
	128	+ cmake_version='3.16.0',
	129	+ os="Linux",
	130	+ libc_version='2.27',
	131	+ python_version='3.8.0 (64-big runtime)',
	132	+ python_platform='Linux',
	133	+ pip_version='pip',
	134	+ pip_packages='numpy==1.19.0',
	135	+ conda_packages='',
	136	+ caching_allocator_config='default',
	137	+ is_xnnpack_available=True,
	138	+ cpu_info='Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz\\nCore(s): 6\\nThread(s) per core: 2',
	139	+ cann_version='not known'
	140	+ )
	141	+
	142	+ result = pretty_str(env_info)
	143	+ self.assertIn('PyTorch version: 1.0.0', result)
	144	+ self.assertIn('CANN:', result)
	145	+ self.assertIn('not known', result)
	146	+
	147	+ def test_get_env_info_torch_available(self):
	148	+ if not TORCH_AVAILABLE:
	149	+ self.skipTest("torch is not available")
	150	+
	151	+ result = get_env_info()
	152	+ self.assertEqual(result.torch_version, torch.__version__)
	153	+ self.assertEqual(result.is_debug_build, str(torch.version.debug))
	154	+
	155	+
	156	+if __name__ == "__main__":
	157	+ run_tests()