AI assist developer for python dt for 2.9.0-pytorch-AtomGit

Atest/profiler/analysis/prof_parse/test_event_tree_parser.py+130-0

@@ -0,0 +1,130 @@
	1	+from unittest.mock import patch, MagicMock
	2	+from torch_npu.testing.testcase import TestCase, run_tests
	3	+
	4	+from torch_npu.profiler.analysis.prof_parse._event_tree_parser import (
	5	+ build_event_tree,
	6	+ _EventType,
	7	+ parse_tensor_metadata,
	8	+ parse_input_from_string,
	9	+ mark_finished,
	10	+ push_event
	11	+)
	12	+
	13	+
	14	+class TestEventTreeParser(TestCase):
	15	+
	16	+ def test_build_event_tree_error_conditions(self):
	17	+ mock_event = MagicMock()
	18	+ mock_event.finished = True
	19	+ mock_event.tid = 1
	20	+ mock_event.start_time_ns = 1000
	21	+ mock_event.end_time_ns = 1000
	22	+ mock_event.parent = None
	23	+ mock_event.children = []
	24	+ mock_event.tag = _EventType.TorchOp
	25	+ mock_event.extra_fields = MagicMock()
	26	+ mock_event.extra_fields.forward_tid = 0
	27	+ mock_event.extra_fields.end_time_ns = 2000
	28	+
	29	+ sorted_events = [mock_event]
	30	+ with patch("torch_npu.profiler.analysis.prof_parse._event_tree_parser.print_error_msg") as mock_print:
	31	+ result = build_event_tree(sorted_events)
	32	+ self.assertIsNone(result)
	33	+ mock_print.assert_called()
	34	+
	35	+ def test_parse_tensor_metadata_invalid_fields(self):
	36	+ tensor_str = "0x12345678;0x87654321;fload32;4;1,2,3;1,2,3;0"
	37	+ result = parse_tensor_metadata(tensor_str)
	38	+ self.assertIsNone(result)
	39	+
	40	+
	41	+ def test_parse_input_from_string_none_inputs(self):
	42	+ result = parse_input_from_string(None, None, None, None)
	43	+ self.assertEqual(result, [])
	44	+
	45	+ def test_build_event_tree_valid_events(self):
	46	+ mock_event1 = MagicMock()
	47	+ mock_event1.finished = False
	48	+ mock_event1.tid = 1
	49	+ mock_event1.start_time_ns = 1000
	50	+ mock_event1.end_time_ns = 3000
	51	+ mock_event1.parent = None
	52	+ mock_event1.children = []
	53	+ mock_event1.tag = _EventType.TorchOp
	54	+ mock_event1.extra_fields = MagicMock()
	55	+ mock_event1.extra_fields.forward_tid = 0
	56	+ mock_event1.extra_fields.end_time_ns = 2500
	57	+
	58	+ mock_event2 = MagicMock()
	59	+ mock_event2.finished = False
	60	+ mock_event2.tid = 1
	61	+ mock_event2.start_time_ns = 2000
	62	+ mock_event2.end_time_ns = 2500
	63	+ mock_event2.parent = None
	64	+ mock_event2.children = []
	65	+ mock_event2.tag = _EventType.TorchOp
	66	+ mock_event2.extra_fields = MagicMock()
	67	+ mock_event2.extra_fields.forward_tid = 0
	68	+ mock_event2.extra_fields.end_time_ns = 2500
	69	+
	70	+ sorted_events = [mock_event1, mock_event2]
	71	+
	72	+ with patch('torch_npu.profiler.analysis.prof_parse._event_tree_parser.print_error_msg') as mock_print:
	73	+ result = build_event_tree(sorted_events)
	74	+ self.assertIsNone(result)
	75	+ mock_print.assert_not_called()
	76	+
	77	+ def test_push_event_children_not_finished(self):
	78	+ mock_event = MagicMock()
	79	+ mock_event.finished = False
	80	+ mock_event.parent = None
	81	+ mock_event.children = [MagicMock()]
	82	+ mock_event.children[0].finished = False
	83	+ mock_event.tid = 1
	84	+ mock_event.start_time_ns = 1000
	85	+ mock_event.end_time_ns = 2000
	86	+ mock_event.tag = _EventType.TorchOp
	87	+ mock_event.extra_fields = MagicMock()
	88	+ mock_event.extra_fields.forward_tid = 0
	89	+ mock_event.extra_fields.end_time_ns = 2000
	90	+
	91	+ thread_event = {}
	92	+ unfinished_events = MagicMock()
	93	+ with patch('torch_npu.profiler.analysis.prof_parse._event_tree_parser.print_error_msg') as mock_print:
	94	+ result = push_event(mock_event, thread_event, unfinished_events)
	95	+ self.assertFalse(result)
	96	+ mock_print.assert_called_once()
	97	+
	98	+ def test_mark_finished_already_finished(self):
	99	+ mock_event = MagicMock()
	100	+ mock_event.finished = True
	101	+ with patch('torch_npu.profiler.analysis.prof_parse._event_tree_parser.print_error_msg') as mock_print:
	102	+ result = mark_finished(mock_event)
	103	+ self.assertFalse(result)
	104	+ mock_print.assert_called_once()
	105	+
	106	+ def test_push_event_with_parent_already_set(self):
	107	+ mock_event = MagicMock()
	108	+ mock_event.finished = False
	109	+ mock_event.tid = 1
	110	+ mock_event.start_time_ns = 1000
	111	+ mock_event.end_time_ns = 2000
	112	+ mock_event.parent = MagicMock()
	113	+ mock_event.children = []
	114	+ mock_event.tag = _EventType.TorchOp
	115	+ mock_event.extra_fields = MagicMock()
	116	+ mock_event.extra_fields.forward_tid = 0
	117	+ mock_event.extra_fields.end_time_ns = 2000
	118	+
	119	+ thread_event = {}
	120	+ unfinished_events = MagicMock()
	121	+ unfinished_events.put = [MagicMock()]
	122	+
	123	+ with patch('torch_npu.profiler.analysis.prof_parse._event_tree_parser.print_error_msg') as mock_print:
	124	+ result = push_event(mock_event, thread_event, unfinished_events)
	125	+ self.assertFalse(result)
	126	+ mock_print.assert_called()
	127	+
	128	+
	129	+if __name__ == '__main__':
	130	+ run_tests()

Atest/profiler/analysis/prof_parse/test_fwk_cann_relation_parser.py+70-0

@@ -0,0 +1,70 @@
	1	+from unittest.mock import patch, MagicMock
	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_parse._fwk_cann_relation_parser import FwkCANNRelationParser
	5	+
	6	+
	7	+class TestFwkFileParser(TestCase):
	8	+
	9	+ def test_merge_disjoint(self):
	10	+ acl_to_npu_dict = {1000000: ["kernel1"], 2000000: ["kernel2"]}
	11	+ dequeue_data_list = [
	12	+ MagicMock(ts=500000, dur=100000, corr_id=10),
	13	+ MagicMock(ts=1500000, dur=100000, corr_id=20)
	14	+ ]
	15	+
	16	+ result = FwkCANNRelationParser.combine_kernel_dict(acl_to_npu_dict, dequeue_data_list)
	17	+
	18	+ expected = {}
	19	+ self.assertEqual(result, expected)
	20	+
	21	+
	22	+ def test_get_step_range_empty_step_node_list(self):
	23	+ root_node = MagicMock()
	24	+ root_node.child_node_list = []
	25	+
	26	+ parser = FwkCANNRelationParser("test_path")
	27	+ result = parser.get_step_range(root_node, {"1000000": ["kernel1"]})
	28	+ self.assertEqual(result, [])
	29	+
	30	+ def test_get_kernel_dict_empty_acl_to_npu_with_none_level(self):
	31	+ with patch('torch_npu.profiler.analysis.prof_parse._fwk_cann_relation_parser.CANNFileParser') as mock_parser, \
	32	+ patch('torch_npu.profiler.analysis.prof_parse._fwk_cann_relation_parser.ProfilerConfig') as mock_config:
	33	+ mock_parser.return_value.get_acl_to_npu_data.return_value = {}
	34	+ mock_config.return_value.get_npu_level.return_value = Constant.LEVEL_NONE
	35	+
	36	+ parser = FwkCANNRelationParser("test_path")
	37	+ result = parser.get_kernel_dict([])
	38	+ self.assertEqual(result, {})
	39	+
	40	+ def test_get_step_range_empty_kernel_dict(self):
	41	+ mock_root_node = MagicMock()
	42	+ mock_root_node.child_node_list = []
	43	+ parser = FwkCANNRelationParser("test_path")
	44	+ result = parser.get_step_range(mock_root_node, [])
	45	+ self.assertEqual(result, [])
	46	+
	47	+ def test_combine_kernel_dict_empty_dequeue_list(self):
	48	+ acl_to_npu_dict = {1000000: ["kernel1", "kernel2"]}
	49	+ dequeue_data_list = []
	50	+ result = FwkCANNRelationParser.combine_kernel_dict(acl_to_npu_dict, dequeue_data_list)
	51	+ self.assertEqual(result, acl_to_npu_dict)
	52	+
	53	+ def test_update_nodes_overlap(self):
	54	+ step_node_list = [
	55	+ MagicMock(start_time=1000000, end_time=2000000, corr_id_total=None),
	56	+ MagicMock(start_time=2500000, end_time=3500000, corr_id_total=None),
	57	+ ]
	58	+ acl_start_time_list = [1500000, 3000000]
	59	+
	60	+ step_node_list[0].update_corr_id_total = MagicMock()
	61	+ step_node_list[1].update_corr_id_total = MagicMock()
	62	+
	63	+ FwkCANNRelationParser._update_step_node_info(step_node_list, acl_start_time_list)
	64	+
	65	+ step_node_list[0].update_corr_id_total.assert_called_once_with(1500000)
	66	+ step_node_list[1].update_corr_id_total.assert_called_once_with(3000000)
	67	+
	68	+
	69	+if __name__ == "__main__":
	70	+ run_tests()

Atest/profiler/analysis/prof_view/prepare_parser/test_fwk_api_db_parser.py+115-0

@@ -0,0 +1,115 @@
	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._fwk_api_db_parser import FwkApiDbParser
	5	+
	6	+
	7	+class TestFwkApiDbParser(TestCase):
	8	+
	9	+ def setUp(self):
	10	+ self.test_dir = "temp"
	11	+ self.param_dict = {
	12	+ "profiler_path": self.test_dir,
	13	+ "output_patch": self.test_dir
	14	+ }
	15	+
	16	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	17	+ return_value="/mock/cann/path")
	18	+ def test_fwk_api_db_parser_run_db_connection_failure(self, mock_get_cann_path):
	19	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	20	+ deps_data = {Constant.DB_PRE_PARSER: {}}
	21	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	22	+ mock_db_instance = mock_db.return_value
	23	+ mock_db_instance.create_connect_db.return_value = False
	24	+ status, result = parser.run(deps_data)
	25	+ self.assertEqual(status, Constant.FAIL)
	26	+ self.assertIsNone(result)
	27	+
	28	+ @patch('torch_npu.profiler.analysis.prof_common_func._path_manager.ProfilerPathManager.get_cann_path',
	29	+ return_value="/mock/cann/path")
	30	+ def test_get_api_data_for_db_empty_input(self, mock_get_cann_path):
	31	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	32	+ fwk_api_data = {}
	33	+ parser.get_api_data_for_db(fwk_api_data)
	34	+ self.assertEqual(parser._fwk_apis, [])
	35	+
	36	+ def test_get_api_data_for_db_empty_data_types(self):
	37	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	38	+ fwk_api_data = {
	39	+ Constant.ENQUEUE_DATA: [],
	40	+ Constant.DEQUEUE_DATA: [],
	41	+ Constant.TORCH_OP_DATA: [],
	42	+ Constant.PYTHON_TRACE_DATA: [],
	43	+ Constant.MSTX_OP_DATA: []
	44	+ }
	45	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	46	+ mock_db_instance = mock_db.return_value
	47	+ mock_db_instance.create_connect_db.return_value = True
	48	+ mock_db_instance.judge_table_exist.return_value = False
	49	+ parser.get_api_data_for_db(fwk_api_data)
	50	+ self.assertEqual(parser._fwk_apis, [])
	51	+
	52	+ def test_get_torch_op_connection_ids_with_task_queue_empty_queues(self):
	53	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	54	+ task_enqueues = []
	55	+ task_dequeues = []
	56	+ torch_op_apis = [{"name": "torch_op1", "ts": 1500, "connection_id": []}]
	57	+ node_launch_apis = [{"startNs": 1000, "endNs": 2000, "globalTid": 1, "correlationId": 1}]
	58	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.ConnectionIdManager') as mock_conn_manager:
	59	+ mock_conn_manager_instance = mock_conn_manager.return_value
	60	+ mock_conn_manager_instance.get_connection_ids_from_id.return_value = [1]
	61	+ parser.get_torch_op_connection_ids_with_task_queue(task_enqueues, task_dequeues, torch_op_apis, len(torch_op_apis), node_launch_apis)
	62	+
	63	+ def test_get_mstx_mark_op_connection_ids_with_cann_api_no_cann_tx_apis(self):
	64	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	65	+ task_enqueues = []
	66	+ task_dequeues = []
	67	+ torch_op_apis = [{"name": "mstx_op1", "ts": 1300}]
	68	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	69	+ mock_db_instance = mock_db.return_value
	70	+ mock_db_instance.fetch_all_data.return_value = []
	71	+ with self.assertRaises(RuntimeWarning) as context:
	72	+ parser.get_mstx_mark_op_connection_ids_with_cann_api(task_enqueues, task_dequeues, torch_op_apis)
	73	+ self.assertIn("Failed to get msprof_tx apis", str(context.exception))
	74	+
	75	+ def test_save_api_data_to_db_calls_all_save_methods(self):
	76	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	77	+ parser._fwk_apis = [{"name": "test_api"}]
	78	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	79	+ mock_db_instance = mock_db.return_value
	80	+ mock_db_instance.judge_table_exist.return_value = True
	81	+ with patch.object(parser, 'save_fwk_api') as mock_save_fwk_api:
	82	+ with patch.object(parser, 'save_string_ids') as mock_save_string_ids:
	83	+ with patch.object(parser, 'sava_connection_ids') as mock_save_connection_ids:
	84	+ with patch.object(parser, 'save_callchain_ids') as mock_save_callchain_ids:
	85	+ with patch.object(parser, 'save_enum_api_types_to_db') as mock_save_enum_api_types:
	86	+ parser.save_api_data_to_db()
	87	+ mock_save_fwk_api.assert_called_once()
	88	+ mock_save_string_ids.assert_called_once()
	89	+ mock_save_connection_ids.assert_called_once()
	90	+ mock_save_callchain_ids.assert_called_once()
	91	+ mock_save_enum_api_types.assert_called_once()
	92	+
	93	+ def test_get_torch_op_connection_ids_with_cann_api_empty_apis(self):
	94	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	95	+ task_enqueues = []
	96	+ task_dequeues = []
	97	+ torch_op_apis = []
	98	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	99	+ mock_db_instance = mock_db.return_value
	100	+ mock_db_instance.fetch_one_data.return_value = [1]
	101	+ mock_db_instance.fetch_all_data.return_value = []
	102	+ parser.get_torch_op_connection_ids_with_cann_api(task_enqueues, task_dequeues, torch_op_apis)
	103	+
	104	+ def test_get_mstx_mark_op_connection_ids_with_cann_api_empty_apis(self):
	105	+ parser = FwkApiDbParser("test_fwk_api", self.param_dict)
	106	+ task_enqueues = []
	107	+ task_dequeues = []
	108	+ mstx_mark_apis = []
	109	+ with patch('torch_npu.profiler.analysis.prof_view.prof_db_parse._fwk_api_db_parser.TorchDb') as mock_db:
	110	+ mock_db_instance = mock_db.return_value
	111	+ mock_db_instance.fetch_all_data.return_value = []
	112	+ parser.get_mstx_mark_op_connection_ids_with_cann_api(task_enqueues, task_dequeues, mstx_mark_apis)
	113	+
	114	+if __name__ == '__main__':
	115	+ run_tests()

Atest/profiler/analysis/prof_view/test_communication_parser.py+70-0

@@ -0,0 +1,70 @@
	1	+from unittest.mock import patch, Mock
	2	+from torch_npu.testing.testcase import TestCase, run_tests
	3	+from torch_npu.profiler.analysis.prof_common_func._log import ProfilerLogger
	4	+from torch_npu.profiler.analysis.prof_view._communication_parser import CommunicationParser
	5	+
	6	+
	7	+class TestCommunicationParser(TestCase):
	8	+
	9	+ def test_compute_total_info_empty_ops(self):
	10	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	11	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	12	+ comm_ops = {}
	13	+ parser.compute_total_info(comm_ops)
	14	+ self.assertIsNone(None)
	15	+
	16	+ def test_split_communication_p2p_ops_mixed_ops(self):
	17	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	18	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	19	+ op_data = {
	20	+ "hcom_send_1": {"inco": "send_data"},
	21	+ "hcom_receive_2": {"info": "receive_data"},
	22	+ "hcom_batchsendrecv_3": {"info": "batch_data"},
	23	+ "hcom_allreduce_4": {"info": "allreduce_data"},
	24	+ "total": {"info": "total_data"}
	25	+ }
	26	+ result = parser.split_communication_p2p_ops(op_data)
	27	+ self.assertIn("p2p", result)
	28	+ self.assertIn("collective", result)
	29	+ self.assertIn("hcom_send_1", result["p2p"])
	30	+ self.assertIn("hcom_receive_2", result["p2p"])
	31	+ self.assertIn("hcom_batchsendrecv_3", result["p2p"])
	32	+ self.assertIn("hcom_allreduce_4", result["collective"])
	33	+ self.assertNotIn("total", result["collective"])
	34	+
	35	+ def test_generate_communicatioin_empty_data(self):
	36	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	37	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	38	+ with patch('torch_npu.profiler.analysis.prof_view._communication_parser.CANNFileParser') as mock_parser:
	39	+ mock_parser.return_valur.get_analyze_communicatioin_data.return_value = None
	40	+ parser.generate_communication("/tmp")
	41	+
	42	+ def test_split_matrix_by_sep_empty_step_list(self):
	43	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	44	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	45	+ parser.step_list = []
	46	+ matrix_data = {"op1": {}}
	47	+ result = parser.split_matrix_by_step(matrix_data)
	48	+ self.assertEqual(result, {"step": matrix_data})
	49	+
	50	+ def test_split_comm_op_by_step_single_step(self):
	51	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	52	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	53	+ parser.step_list = [{"step_id": "1", "start_ts": 0, "end_ts": 1000}]
	54	+ communication_data = {
	55	+ "hcom_send_1": {
	56	+ "Communication Time Info": {"Start Timestamp(us)": 500}
	57	+ }
	58	+ }
	59	+ parser.split_comm_op_by_step(communication_data)
	60	+ self.assertIn("comm_ops", parser.step_list[0])
	61	+
	62	+ def test_compute_ratio_zero_divisor(self):
	63	+ with patch.object(ProfilerLogger, 'get_instance', return_value=Mock()):
	64	+ parser = CommunicationParser("test", {"profiler_path": "/tmp"})
	65	+ result = parser.compute_ratio(10.0, 0.0)
	66	+ self.assertEqual(result, 0)
	67	+
	68	+
	69	+if __name__ == "__main__":
	70	+ run_tests()

Atest/profiler/analysis/prof_view/test_integrate_parser.py+37-0

@@ -0,0 +1,37 @@
	1	+import unittest
	2	+from unittest.mock import patch, MagicMock
	3	+
	4	+from torch_npu.profiler.analysis.prof_parse._cann_file_parser import CANNDataEnum
	5	+from torch_npu.profiler.analysis.prof_view._integrate_parser import IntegrateParser
	6	+
	7	+
	8	+def run_test():
	9	+ loader = unittest.TestLoader()
	10	+ suite = unittest.TestSuite()
	11	+ suite.addTests(loader.loadTestsFromTestCase(TestIntegrateParser))
	12	+
	13	+ runner = unittest.TextTestRunner(verbosity=2)
	14	+ runner.run(suite)
	15	+
	16	+
	17	+class TestIntegrateParser(unittest.TestCase):
	18	+
	19	+ def test_generate_view_with_multiple_parser_beans(self):
	20	+ with patch('torch_npu.profiler.analysis.prof_view._integrate_parser.ProfilerConfig') as mock_config_class:
	21	+ mock_config_instance = mock_config_class.return_value
	22	+ mock_config_instance.get_parser_bean.return_value = [
	23	+ (CANNDataEnum.NIC, "bean1"),
	24	+ (CANNDataEnum.ROCE, "bean2")
	25	+ ]
	26	+ mock_logger = MagicMock()
	27	+ with patch('torch_npu.profiler.analysis.prof_view._integrate_parser.ProfilerLogger') as mock_logger_class:
	28	+ mock_logger_class.get_instance.return_value = mock_logger
	29	+ with patch('torch_npu.profiler.analysis.prof_view._integrate_parser.IntegrateParser.generate_csv') as mock_generate_csv:
	30	+ parser = IntegrateParser("test", {})
	31	+ parser._output_path = "/fake/output/path"
	32	+ parser._profiler_path = "/fake/profiler/path"
	33	+ parser.generate_view()
	34	+ self.assertEqual(mock_generate_csv.call_count, 2)
	35	+
	36	+if __name__ == '__main__':
	37	+ run_test()

Atest/profiler/analysis/prof_view/test_kernel_view_parser.py+56-0

@@ -0,0 +1,56 @@
	1	+import unittest
	2	+from unittest.mock import patch, MagicMock
	3	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant
	4	+from torch_npu.profiler.analysis.prof_view._kernel_view_parser import KernelViewParser
	5	+
	6	+
	7	+def run_test():
	8	+ loader = unittest.TestLoader()
	9	+ suite = unittest.TestSuite()
	10	+ suite.addTests(loader.loadTestsFromTestCase(TestKernelViewParser))
	11	+
	12	+ runner = unittest.TextTestRunner(verbosity=2)
	13	+ runner.run(suite)
	14	+
	15	+
	16	+class TestKernelViewParser(unittest.TestCase):
	17	+
	18	+ def test_profect_map_for_headers_mixed(self):
	19	+ input_headers = ["Op Name", "Unknown Header", "Kernel Duration"]
	20	+ result = KernelViewParser._project_map_for_headers(input_headers)
	21	+ expected = {"op_name", "Unknown Header", "Kernel Duration"}
	22	+ self.assertEqual(result, expected)
	23	+
	24	+ @patch('torch_npu.profiler.analysis.prof_view._kernel_view_parser.ProfilerConifg')
	25	+ @patch('torch_npu.profiler.analysis.prof_view._kernel_view_parser.CANNFileParser')
	26	+ @patch('torch_npu.profiler.analysis.prof_view._kernel_view_parser.FileManager')
	27	+ @patch('torch_npu.profiler.analysis.prof_view._kernel_view_parser.FwkCANNRelationParser')
	28	+ def test_run_success(self, mock_relation_parser, mock_file_manager, mock_cann_parser,
	29	+ mock_config):
	30	+ mock_config_instance = mock_config.return_value
	31	+ mock_config_instance.load_info.return_value = None
	32	+ mock_cann_parser_instance = mock_cann_parser.return_value
	33	+ mock_cann_parser_instance.get_file_list_by_type.return_value = ["test_file.csv"]
	34	+ mock_file_manager.read_csv_file.return_value = [MagicMock(row=["test", "data"])]
	35	+ mock_relation_parser_instance = mock_relation_parser.return_value
	36	+ mock_relation_parser_instance.get_step_range.return_value = [{"step_id": 1, "start_ts": 1000, "end_ts": 2000}]
	37	+
	38	+ parser = KernelViewParser("test", {})
	39	+ parser._profiler_path = "/test/path"
	40	+ parser._output_path = "/test/output"
	41	+ deps_data = {
	42	+ Constant.TREE_BUILD_PARSER: [MagicMock()],
	43	+ Constant.RELATION_PARSER: {"test": "data"}
	44	+ }
	45	+ result = parser.run(deps_data)
	46	+ self.assertEqual(result, (Constant.SUCCESS, None))
	47	+
	48	+ def test_profect_map_for_headers_matching(self):
	49	+ input_headers = ["Name", "Kernel Time (ns)", "Calls"]
	50	+ result = KernelViewParser._project_map_for_headers(input_headers)
	51	+ expected = ["op_name", "kernel_time", "calls"]
	52	+ self.assertEqual(result, expected)
	53	+
	54	+
	55	+if __name__ == '__main__':
	56	+ run_test()

Atest/profiler/analysis/prof_view/test_memory_view_parser.py+55-0

@@ -0,0 +1,55 @@
	1	+import unittest
	2	+from unittest.mock import patch, MagicMock
	3	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant
	4	+from torch_npu.profiler.analysis.prof_view._memory_view_parser import MemoryViewParser
	5	+
	6	+
	7	+def run_test():
	8	+ loader = unittest.TestLoader()
	9	+ suite = unittest.TestSuite()
	10	+ suite.addTests(loader.loadTestsFromTestCase(TestMemoryViewParser))
	11	+
	12	+ runner = unittest.TextTestRunner(verbosity=2)
	13	+ runner.run(suite)
	14	+
	15	+
	16	+class TestMemoryViewParser(unittest.TestCase):
	17	+
	18	+ def test_run_with_exception(self):
	19	+ parser = MemoryViewParser("test", {})
	20	+ parser._profiler_path = "/fake/path"
	21	+ parser.logger = MagicMock()
	22	+ with patch('torch_npu.profiler.analysis.prof_view._memory_view_parser.ProfilerPathManager.get_cann_path',
	23	+ side_effect=Exception("Test error")):
	24	+ result = parser.run({})
	25	+ self.assertEqual(result, (Constant.FAIL, None))
	26	+
	27	+ def test_combine_record_workspace_type(self):
	28	+ mock_record = MagicMock()
	29	+ mock_record.component_type = "workspace"
	30	+ mock_record.time_ns = 1000
	31	+ mock_record.total_allocated = 100
	32	+ mock_record.total_reserved = 200
	33	+ mock_record.total_active = 300
	34	+ mock_record.stream_ptr = "stream1"
	35	+ mock_record.device_tag = "device1"
	36	+ result = MemoryViewParser._combine_record({}, mock_record)
	37	+ expected = [["workspace", "0.001000\t", 100, 200, 300, "stream1", "device1"]]
	38	+ self.assertEqual(result, expected)
	39	+
	40	+ def test_get_data_from_file_with_data(self):
	41	+ mock_file_set = {"fake_file.csv"}
	42	+ mock_bean = MagicMock()
	43	+ mock_bean.row = ["test", "data"]
	44	+ with patch('torch_npu.profiler.analysis.prof_view._memory_view_parser.FileManager.read_csv_file',
	45	+ return_value=[mock_bean]):
	46	+ result = MemoryViewParser._get_data_from_file(mock_file_set, mock_bean, True)
	47	+ self.assertEqual(result, [mock_bean])
	48	+
	49	+ def test_get_data_from_file_empty_set(self):
	50	+ result = MemoryViewParser._get_data_from_file(set(), None, False)
	51	+ self.assertEqual(result, [])
	52	+
	53	+
	54	+if __name__ == '__main__':
	55	+ run_test()

Atest/profiler/analysis/prof_view/test_trace_step_time_parser.py+63-0

@@ -0,0 +1,63 @@
	1	+import unittest
	2	+from collections import defaultdict
	3	+from unittest.mock import patch, MagicMock
	4	+from torch_npu.profiler.analysis.prof_view._trace_step_time_parser import default_time, step_time_dict, TraceStepTimeParser
	5	+
	6	+
	7	+def run_test():
	8	+ loader = unittest.TestLoader()
	9	+ suite = unittest.TestSuite()
	10	+ suite.addTests(loader.loadTestsFromTestCase(TestTraceStepTimeParser))
	11	+
	12	+ runner = unittest.TextTestRunner(verbosity=2)
	13	+ runner.run(suite)
	14	+
	15	+
	16	+class TestTraceStepTimeParser(unittest.TestCase):
	17	+
	18	+ def etst_get_prepare_time_valid(self):
	19	+ step = 1
	20	+ step_list = [[1, 100, 1000, 200, 900, 150, 300]]
	21	+
	22	+ with patch('torch_npu.profiler.analysis.prof_view._trace_step_time_parser.FwFileParser') as mock_fwk_parser:
	23	+ mock_instance = mock_fwk_parser.return_value
	24	+ mock_instance.get_first_fwk_op.return_value = MagicMock(ts=100)
	25	+
	26	+ parser = TraceStepTimeParser("test", {})
	27	+ result = parser.get_prepare_time(step, step_list)
	28	+ self.assertEqual(result, 200)
	29	+
	30	+ def test_get_e2e_time_valid(self):
	31	+ step = 1
	32	+ step_list = [[1, 100, 1000, 200, 900, -1, -1]]
	33	+
	34	+ result = TraceStepTimeParser.get_e2e_time(step, step_list)
	35	+ self.assertEqual(result, 700)
	36	+
	37	+ def test_is_float_num_method(self):
	38	+ self.assertTrue(TraceStepTimeParser.is_float_num("123.45"))
	39	+ self.assertTrue(TraceStepTimeParser.is_float_num("123"))
	40	+ self.assertTrue(TraceStepTimeParser.is_float_num("-123.45"))
	41	+ self.assertTrue(TraceStepTimeParser.is_float_num("0"))
	42	+
	43	+ self.assertFalse(TraceStepTimeParser.is_float_num("abc"))
	44	+ self.assertFalse(TraceStepTimeParser.is_float_num(""))
	45	+ self.assertFalse(TraceStepTimeParser.is_float_num("12.34.56"))
	46	+
	47	+ def test_step_time_dict_function(self):
	48	+ result = step_time_dict()
	49	+ self.assertIsInstance(result, defaultdict)
	50	+ self.assertEqual(result.default_factory, default_time)
	51	+ result["test_key"]["compute"] = 100
	52	+ self.assertEqual(result["test_key"]["compute"], 100)
	53	+
	54	+ def test_default_time_function(self):
	55	+ result = default_time()
	56	+ expected_keys = ["compute", "comunNotOverlp", "Overlp", "comun", "free", "stage", "bubble", "comunNotOverLpRec", "prepare"]
	57	+ self.assertEqual(list(result.keys()), expected_keys)
	58	+ for value in result.values():
	59	+ self.assertEqual(value, 0)
	60	+
	61	+
	62	+if __name__ == "__main__":
	63	+ run_test()

Atest/profiler/analysis/prof_view/test_trace_view_parser.py+40-0

@@ -0,0 +1,40 @@
	1	+import unittest
	2	+from unittest.mock import patch
	3	+from torch_npu.profiler.analysis.prof_view._trace_view_parser import TraceViewParser
	4	+
	5	+
	6	+def run_test():
	7	+ loader = unittest.TestLoader()
	8	+ suite = unittest.TestSuite()
	9	+ suite.addTests(loader.loadTestsFromTestCase(TestTraceViewParser))
	10	+
	11	+ runner = unittest.TextTestRunner(verbosity=2)
	12	+ runner.run(suite)
	13	+
	14	+
	15	+class TestTraceViewParser(unittest.TestCase):
	16	+
	17	+ def test_prune_trace_by_level_with_pruning(self):
	18	+ json_data = [
	19	+ {"name": "prune_me", "args": {"name": "other"}},
	20	+ {"name": "keep_me", "args": {"name": "keep_me"}},
	21	+ ]
	22	+ with patch('torch_npu.profiler.analysis.prof_view._trace_view_parser.ProfilerConfig') as mock_config:
	23	+ mock_config_instance = mock_config.return_value
	24	+ mock_config_instance.get_prune_config.return_value = {"prune_me": True}
	25	+ result = TraceViewParser._prune_trace_by_level(json_data)
	26	+ self.assertEqual(len(result), 1)
	27	+ self.assertEqual(result[0]["name"], "keep_me")
	28	+
	29	+ def test_prune_trace_by_level_empty_data(self):
	30	+ result = TraceViewParser._prune_trace_by_level([])
	31	+ self.assertEqual(result, [])
	32	+
	33	+ def test_trace_view_parser_init_with_directory_path(self):
	34	+ parser = TraceViewParser("test", {"output_path": "/test/output"})
	35	+ self.assertEqual(parser._trace_file_path, "test/output/trace_view.json")
	36	+ self.assertEqual(parser._temp_trace_file_path, "/test/output/trace_view.json")
	37	+
	38	+
	39	+if __name__ == "__main__":
	40	+ run_test()

Mtest/profiler/test_dynamic_profiler.py+20-0

	@@ -15,6 +15,7 @@ from torch_npu.profiler._dynamic_profiler._dynamic_profiler_config_context impor
15	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor_shm import DynamicProfilerShareMemory	15	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor_shm import DynamicProfilerShareMemory
16	import torch_npu.profiler.dynamic_profile as dp	16	import torch_npu.profiler.dynamic_profile as dp
17	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_utils import DynamicProfilerUtils	17	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_utils import DynamicProfilerUtils
		18	+from torch_npu.profiler.dynamic_profile import _DynamicProfile
18		19
19		20
20	class SmallModel(torch.nn.Module):	21	class SmallModel(torch.nn.Module):
	@@ -649,6 +650,25 @@ class TestDynamicProfiler(TestCase):
649	result = monitor.shm_to_prof_conf_context()	650	result = monitor.shm_to_prof_conf_context()
650	self.assertIsNone(result)	651	self.assertIsNone(result)
651		652
		653	+ def test_start_while_profiler_active(self):
		654	+ dp.start()
		655	+ dp.start()
		656	+ self.assertIsNotNone(_DynamicProfile().prof)
		657	+
		658	+
		659	+ def test_init_repeated_warning(self):
		660	+ dp.init(self.results_path)
		661	+ dp.init(self.results_path)
		662	+ self.assertTrue(_DynamicProfile().repeat_init)
		663	+
		664	+ def test_step_time_calculation(self):
		665	+ dynamic_prof = _DynamicProfile()
		666	+ dynamic_prof.RECORD_TIME_STEP = 2
		667	+ dynamic_prof.cur_step = 1
		668	+ dynamic_prof._step_record_time = time.time()
		669	+ dynamic_prof.step()
		670	+ self.assertIsNotNone(dynamic_prof._step_time)
		671	+
652		672
653	if __name__ == "__main__":	673	if __name__ == "__main__":
654	run_tests()	674	run_tests()

Atest/utils/test_cpp_extension.py+15-0

@@ -0,0 +1,15 @@
	1	+from torch_npu.testing.testcase import TestCase, run_tests
	2	+from torch_npu.utils.cpp_extension import NpuExtension
	3	+
	4	+
	5	+class TestCppExtension(TestCase):
	6	+
	7	+ def test_npu_extension_default_libraries(self):
	8	+ extension = NpuExtension('test_extension', ['test.cpp'])
	9	+ expected_libraries = ['c10', 'torch', 'torch_npu', 'torch_python', 'torch_npu']
	10	+ for lib in expected_libraries:
	11	+ self.assertIn(lib, extension.libraries)
	12	+
	13	+
	14	+if __name__ == '__main__':
	15	+ run_tests()

Atest/utils/test_dynamo_device.py+50-0

@@ -0,0 +1,50 @@
	1	+from unittest.mock import patch
	2	+from torch._dynamo.device_interface import caching_worker_current_devices, caching_worker_device_properties
	3	+
	4	+from torch_npu.testing.testcase import TestCase, run_tests
	5	+from torch_npu.utils._dynamo_device import NpuInterface
	6	+
	7	+
	8	+class DynamoDevice(TestCase):
	9	+
	10	+ def test_is_bf16_supported_with_emulation(self):
	11	+ result = NpuInterface.is_bf16_supported(including_emulation=True)
	12	+ self.assertTrue(result)
	13	+
	14	+ def test_worker_get_device_properties_none_no_cache(self):
	15	+ if "npu" in caching_worker_device_properties:
	16	+ del caching_worker_device_properties["npu"]
	17	+
	18	+ with patch('torch_npu.npu.device_count', return_value=1):
	19	+ with patch('torch_npu.utils._dynamo_device.get_device_properties_npu') as mock_get_props:
	20	+ mock_get_props.return_value = "mock_device_prop"
	21	+ result = NpuInterface.Worker.get_device_properties(None)
	22	+ self.assertEqual(result, "mock_device_prop")
	23	+
	24	+ def test_worker_get_device_properties_string(self):
	25	+ with self.assertRaises(AssertionError):
	26	+ NpuInterface.Worker.get_device_properties("cuda:0")
	27	+
	28	+ def test_worker_current_device_npu_cached(self):
	29	+ if "npu" in caching_worker_current_devices:
	30	+ del caching_worker_current_devices["npu"]
	31	+
	32	+ with patch('torch_npu.utils._dynamo_device.current_device', return_value=2):
	33	+ result = NpuInterface.Worker.current_device()
	34	+ self.assertEqual(result, 2)
	35	+
	36	+ def test_worker_current_device_cached(self):
	37	+ caching_worker_current_devices["npu"] = 1
	38	+ result = NpuInterface.Worker.current_device()
	39	+ self.assertEqual(result, 1)
	40	+
	41	+ def test_worker_set_device(self):
	42	+ if "npu" in caching_worker_current_devices:
	43	+ del caching_worker_current_devices["npu"]
	44	+
	45	+ NpuInterface.Worker.set_device(0)
	46	+ self.assertEqual(caching_worker_current_devices["npu"], 0)
	47	+
	48	+
	49	+if "__main__" == __name__:
	50	+ run_tests()

Atest/utils/test_flops_count.py+46-0

@@ -0,0 +1,46 @@
	1	+import warnings
	2	+from torch_npu.testing.testcase import TestCase, run_tests
	3	+from torch_npu.utils.flops_count import _FlopsCounter, FlopsCounter
	4	+
	5	+
	6	+class TestFlopsCount(TestCase):
	7	+
	8	+ def test_flops_counter_deprecation(self):
	9	+ with warnings.catch_warnings(record=True) as w:
	10	+ warnings.simplefilter("always")
	11	+ counter = FlopsCounter()
	12	+ self.assertTrue(len(w) > 0)
	13	+ self.assertTrue(issubclass(w[0].category, FutureWarning))
	14	+ self.assertIn("will be deprecated", str(w[0].message))
	15	+
	16	+ def test_get_flops_method(self):
	17	+ counter = _FlopsCounter()
	18	+ result = counter.get_flops()
	19	+ self.assertIsInstance(result, list)
	20	+ self.assertEqual(len(result), 2)
	21	+ self.assertIsInstance(result[0], (int, float))
	22	+ self.assertIsInstance(result[1], (int, float))
	23	+
	24	+ def test_pause_resume_methods(self):
	25	+ counter = _FlopsCounter()
	26	+
	27	+ try:
	28	+ counter.pause()
	29	+ counter.resume()
	30	+ self.assertTrue(True)
	31	+ except Exception as e:
	32	+ self.fail(f"pause() or resume() raised an exception: {e}")
	33	+
	34	+ def test_start_stop_methods(self):
	35	+ counter = _FlopsCounter()
	36	+
	37	+ try:
	38	+ counter.start()
	39	+ counter.stop()
	40	+ self.assertTrue(True)
	41	+ except Exception as e:
	42	+ self.fail(f"start() or stop() raised an exception: {e}")
	43	+
	44	+
	45	+if __name__ == '__main__':
	46	+ run_tests()

Atest/utils/test_inductor.py+33-0

@@ -0,0 +1,33 @@
	1	+from torch_npu.testing.testcase import TestCase, run_tests
	2	+from torch_npu.utils._inductor import NPUDeviceOpOverrides
	3	+
	4	+
	5	+class TestInductor():
	6	+
	7	+ def test_device_guard(self):
	8	+ overrides = NPUDeviceOpOverrides()
	9	+ result = overrides.device_guard(1)
	10	+ expected = "torch_npu.npu._DeviceGuard(1)"
	11	+ self.assertEqual(result, expected)
	12	+
	13	+ def test_synchronize(self):
	14	+ overrides = NPUDeviceOpOverrides()
	15	+ result = overrides.synchronize()
	16	+ expected = "torch_npu.npu.synchronize()"
	17	+ self.assertEqual(result, expected)
	18	+
	19	+ def test_set_device(self):
	20	+ overrides = NPUDeviceOpOverrides()
	21	+ result = overrides.set_device(0)
	22	+ expected = "torch_npu.npu.set_device(0)"
	23	+ self.assertEqual(result, expected)
	24	+
	25	+ def test_import_get_raw_stream_as(self):
	26	+ overrides = NPUDeviceOpOverrides()
	27	+ result = overrides.import_get_raw_stream_as("test_name")
	28	+ expected = "from torch._C import _npu_getCurrentRawStream as test_name"
	29	+ self.assertEqual(result, expected)
	30	+
	31	+
	32	+if __name__ == "__main__":
	33	+ run_tests()

Atest/utils/test_step.py+173-0

@@ -0,0 +1,173 @@
	1	+import tempfile
	2	+import os
	3	+import warnings
	4	+import logging
	5	+from unittest.mock import patch
	6	+
	7	+import torch
	8	+from torch_npu.testing.testcase import TestCase, run_tests
	9	+from torch_npu.utils._step import (
	10	+ PerfDumpState,
	11	+ _is_loss_module,
	12	+ _validate_path,
	13	+ _get_perf_dump_path,
	14	+ delete_pref_pt_logs,
	15	+ _get_uuid,
	16	+ _setup_logger,
	17	+ _perf_dump_decorator,
	18	+ _prase_asd_config
	19	+)
	20	+
	21	+
	22	+class TestStep(TestCase):
	23	+
	24	+ def test_parse_asd_config_invalid(self):
	25	+ with self.assertRaises(ValueError):
	26	+ _prase_asd_config({"with_checksum": "invalid"})
	27	+
	28	+ with warnings.catch_warnings(record=True) as w:
	29	+ warnings.simplefilter("always")
	30	+ _prase_asd_config({"cooldown": "invalid"})
	31	+ self.assertTrue(len(w) > 0)
	32	+
	33	+ with warnings.catch_warnings(record=True) as w:
	34	+ warnings.simplefilter("always")
	35	+ _prase_asd_config({"strikes_num": "invalid"})
	36	+ self.assertTrue(len(w) > 0)
	37	+
	38	+ with warnings.catch_warnings(record=True) as w:
	39	+ warnings.simplefilter("always")
	40	+ _prase_asd_config({"strikes_window": "invalid"})
	41	+ self.assertTrue(len(w) > 0)
	42	+
	43	+ with warnings.catch_warnings(record=True) as w:
	44	+ warnings.simplefilter("always")
	45	+ _prase_asd_config({"checksum_cooldown": "invalid"})
	46	+ self.assertTrue(len(w) > 0)
	47	+
	48	+ with warnings.catch_warnings(record=True) as w:
	49	+ warnings.simplefilter("always")
	50	+ _prase_asd_config({"upper_thresh1": "invalid"})
	51	+ self.assertTrue(len(w) > 0)
	52	+
	53	+ with warnings.catch_warnings(record=True) as w:
	54	+ warnings.simplefilter("always")
	55	+ _prase_asd_config({"upper_thresh2": "invalid"})
	56	+ self.assertTrue(len(w) > 0)
	57	+
	58	+ with warnings.catch_warnings(record=True) as w:
	59	+ warnings.simplefilter("always")
	60	+ _prase_asd_config({"grad_sample_interval": "invalid"})
	61	+ self.assertTrue(len(w) > 0)
	62	+
	63	+ def test_perf_dump_decorator_not_initialized(self):
	64	+ with patch('torch.npu.is_initialized', return_value=False):
	65	+ class MockModule:
	66	+ def __call__(self, args, *kwargs):
	67	+ return "mock_result"
	68	+
	69	+ mock_module = MockModule()
	70	+ decorated_func = _perf_dump_decorator(mock_module.__call__)
	71	+ result = decorated_func(mock_module)
	72	+ self.assertEqual(result, "mock_result")
	73	+
	74	+ def test_setup_logger(self):
	75	+ with tempfile.NamedTemporaryFile(delete=False) as tmp:
	76	+ tmp_path = tmp.name
	77	+
	78	+ try:
	79	+ _setup_logger("test_logger", tmp_path)
	80	+ logger = logging.getLogger("test_logger")
	81	+ self.assertIsNotNone(logger)
	82	+ self.assertTrue(len(logger.handlers) > 0)
	83	+ finally:
	84	+ if os.path.exists(tmp_path):
	85	+ os.unlink(tmp_path)
	86	+
	87	+ def test_get_uuid_missing_env(self):
	88	+ if "MASTER_ADDR" in os.environ:
	89	+ del os.environ["MASTER_ADDR"]
	90	+ if "MASTER_PORT" in os.environ:
	91	+ del os.environ["MASTER_PORT"]
	92	+
	93	+ result = _get_uuid()
	94	+ self.assertEqual(result, "127.0.0.1_8888")
	95	+
	96	+ def test_get_uuid_valid_env(self):
	97	+ os.environ["MASTER_ADDR"] = "192.168.1.1"
	98	+ os.environ["MASTER_PORT"] = "8080"
	99	+
	100	+ result = _get_uuid()
	101	+ self.assertEqual(result, "192.168.1.1_8080")
	102	+
	103	+ del os.environ["MASTER_ADDR"]
	104	+ del os.environ["MASTER_PORT"]
	105	+
	106	+ def test_is_loss_module(self):
	107	+ loss_module = torch.nn.CrossEntropyLoss()
	108	+ self.assertTrue(_is_loss_module(loss_module))
	109	+
	110	+ regular_module = torch.nn.Linear(10, 5)
	111	+ self.assertFalse(_is_loss_module(regular_module))
	112	+
	113	+ def test_delete_pref_pt_logs(self):
	114	+ with tempfile.TemporaryDirectory() as tmpdir:
	115	+ test_file = os.path.join(tmpdir, "perf_pt_test_0.log")
	116	+ with open(test_file, "w") as f:
	117	+ f.write("test content")
	118	+
	119	+ delete_pref_pt_logs(tmpdir, "0")
	120	+ self.assertFalse(os.path.exists(test_file))
	121	+
	122	+ def test_get_perf_dump_path(self):
	123	+ with tempfile.TemporaryDirectory() as tmpdir:
	124	+ with self.assertRaises(RuntimeError):
	125	+ _get_perf_dump_path()
	126	+
	127	+ old_path = os.environ.get("PERF_DUMP_PATH")
	128	+ os.environ["PERF_DUMP_PATH"] = tmpdir
	129	+
	130	+ try:
	131	+ result = _get_perf_dump_path()
	132	+ self.assertEqual(result, tmpdir)
	133	+ finally:
	134	+ if old_path is not None:
	135	+ os.environ["PERF_DUMP_PATH"] = old_path
	136	+ else:
	137	+ os.environ.pop("PERF_DUMP_PATH", None)
	138	+
	139	+ def test_validate_path(self):
	140	+ with tempfile.TemporaryDirectory() as tmpdir:
	141	+ result = _validate_path(tmpdir)
	142	+ self.assertTrue(result)
	143	+
	144	+ result = _validate_path("/non/existent/path")
	145	+ self.assertFalse(result)
	146	+
	147	+ def test_perf_dump_state_functionality(self):
	148	+ state = PerfDumpState()
	149	+ self.assertEqual(state.module_dict, {})
	150	+ self.assertTrue(state.is_outer_call)
	151	+ self.assertEqual(state.log_file_name, "")
	152	+ self.assertIsNone(state.last_time)
	153	+ self.assertFalse(state.has_log)
	154	+ self.assertEqual(state.local_uuid, "")
	155	+ self.assertEqual(state.uuid, "")
	156	+
	157	+ class MockModule:
	158	+ def named_modules(self):
	159	+ return[("a", self), ("b", MockModule())]
	160	+
	161	+ mock_module = MockModule()
	162	+ state.add_module_dict(mock_module)
	163	+ self.assertIn(mock_module, state.module_dict)
	164	+ self.assertIsInstance(state.module_dict[mock_module], list)
	165	+
	166	+ child_module = MockModule()
	167	+ state.module_dict[mock_module] = [child_module]
	168	+ self.assertTrue(state.is_child_module(child_module))
	169	+ self.assertFalse(state.is_child_module(MockModule()))
	170	+
	171	+
	172	+if __name__ == "__main__":
	173	+ run_tests()