AI assist developer for python DT 2.7.1-pytorch-AtomGit

Mtest/contrib/test_activations.py+37-0

 from torch_npu.testing.testcase import TestCase, run_tests
 from torch_npu.testing.common_utils import create_common_tensor
 from torch_npu.contrib.module import Mish, SiLU
++from torch_npu.contrib.function.fused_attention import _check_compatibility_once
++from torch_npu.contrib.function.fused_attention import _is_format_matched
 class TestActivations(TestCase):
             self.assertRtolEqual(cpu_output, npu_output)
             self.assertRtolEqual(cpu_inputgrad, npu_inputgrad)
++    def test_check_compatibility_once_invalid_hidden_states_shape(self):
++        hidden_states = torch_npu.npu_format_cast(torch.randn(30, 1024).npu(), 29)
++        attention_mask = torch_npu.npu_format_cast(torch.randn(2, 1, 8, 8).npu(), 29)
++        query_kernel = torch_npu.npu_format_cast(torch.randn(1024, 1024).npu(), 29)
++        key_kernel = torch_npu.npu_format_cast(torch.randn(1024, 1024).npu(), 29)
++        value_kernel = torch_npu.npu_format_cast(torch.randn(1024, 1024).npu(), 29)
++        query_bias = torch_npu.npu_format_cast(torch.randn(1024).npu(), 2)
++        key_bias = torch_npu.npu_format_cast(torch.randn(1024).npu(), 2)
++        value_bias = torch_npu.npu_format_cast(torch.randn(1024).npu(), 2)
++
++        with self.assertRaises(RuntimeError):
++            _check_compatibility_once(
++                hidden_states,
++                attention_mask,
++                query_kernel,
++                key_kernel,
++                value_kernel,
++                query_bias,
++                key_bias,
++                value_bias
++            )
++
++    def test_is_format_matched_invalid(self):
++        tensor1 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++        tensor2 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++        tensor3 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++        tensor4 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++        tensor5 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++        tensor6 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 2)
++        tensor7 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 2)
++        tensor8 = torch_npu.npu_format_cast(torch.randn(4, 4).npu(), 29)
++
++        result = _is_format_matched([tensor1, tensor2, tensor3, tensor4, tensor5, tensor6, tensor7, tensor8])
++        self.assertFalse(result)
++
 if __name__ == "__main__":
     run_tests()

Mtest/contrib/test_fuse_add_softmax_dropout.py+15-0

         self.assertRtolEqual(npu_output.detach().cpu().numpy(), high_performance_output.detach().cpu().numpy())
++    def test_training_false_with_zero_dropout(self):
++        training = False
++        dropout = torch_npu.contrib.module.DropoutWithByteMask(0.0)
++        npu_input1 = torch.rand(96, 12, 384, 384).npu().half()
++        npu_input2 = torch.rand(96, 12, 384, 384).npu().half()
++        alpha = 64
++
++        output = fuse_add_softmax_dropout(training=training, dropout=dropout,
++                                          attn_mask=npu_input1, attn_scores=npu_input2,
++                                          attn_head_size=alpha, p=0.1)
++
++        self.assertEqual(output.shape, npu_input2.shape)
++        excepted = self.npu_fuse_add_softmax_dropout(dropout, npu_input1, npu_input2, alpha)
++        self.assertRtolEqual(excepted.detach().cpu().numpy(), output.detach().cpu().numpy())
++
 if __name__ == "__main__":
     run_tests()

Mtest/contrib/test_index_op.py+24-0

	@@ -37,6 +37,30 @@ class TestIndexOp(TestCase):
37	npu_fast_output = self.npu_fast_index_op_exec(npu_input)	37	npu_fast_output = self.npu_fast_index_op_exec(npu_input)
38	self.assertRtolEqual(npu_slow_output.cpu(), npu_fast_output.cpu())	38	self.assertRtolEqual(npu_slow_output.cpu(), npu_fast_output.cpu())
39		39
		40	+ def test_nonzero_nonone_value(self):
		41	+ x = torch.randn(2, 3)
		42	+ condition = torch.tensor([[True, False, True], [False, True, False]])
		43	+ value = 5.5
		44	+ result = npu_fast_condition_index_put(x, condition, value)
		45	+ expected = torch.where(condition, torch.zeros_like(x) + value, x)
		46	+ self.assertRtolEqual(result.cpu(), expected.cpu())
		47	+
		48	+ def test_value_one_mask(self):
		49	+ x = torch.randn(2, 3)
		50	+ condition = torch.tensor([[True, False, True],
		51	+ [False, True, False]])
		52	+ value = 1.0
		53	+ result = npu_fast_condition_index_put(x, condition, value)
		54	+ expected = torch.where(condition, torch.ones_like(x), x)
		55	+ self.assertRtolEqual(result.cpu(), expected.cpu())
		56	+
		57	+ def test_invalid_condition_dtype(self):
		58	+ x = torch.randn(2, 3)
		59	+ condition = torch.randint(0, 2, (2, 3), dtype=torch.int32)
		60	+ value = 0.0
		61	+ with self.assertRaises(TypeError):
		62	+ npu_fast_condition_index_put(x, condition, value)
		63	+
40		64
41	if __name__ == "__main__":	65	if __name__ == "__main__":
42	run_tests()	66	run_tests()

Mtest/contrib/test_multiclass_nms.py+12-0

	@@ -75,6 +75,18 @@ class TestMultiClassNms(TestCase):
75	self.assertRtolEqual(expect_det_bboxes, det_bboxes.cpu())	75	self.assertRtolEqual(expect_det_bboxes, det_bboxes.cpu())
76	self.assertRtolEqual(expect_det_labels, det_labels.cpu())	76	self.assertRtolEqual(expect_det_labels, det_labels.cpu())
77		77
		78	+ def test_npu_multiclass_nms_max_num_exceeds_boxes(self):
		79	+ np.random.seed(111)
		80	+ data1 = np.random.randn(5, 4)
		81	+ boxes = torch.tensor(data1, dtype=torch.float32)
		82	+ data2 = np.random.randn(5, 6)
		83	+ scores = torch.tensor(data2, dtype=torch.float32)
		84	+ boxes = boxes.npu().half()
		85	+ scores = scores.npu().half()
		86	+ det_bboxes, det_labels = npu_multiclass_nms(boxes, scores, score_thr=0.9, nms_thr=0.5, max_num=20)
		87	+ self.assertEqual(det_bboxes.shape[0], 20)
		88	+ self.assertEqual(det_labels.shape[0], 20)
		89	+
78		90
79	if __name__ == "__main__":	91	if __name__ == "__main__":
80	run_tests()	92	run_tests()

Mtest/npu/test_recovery.py+10-0

	@@ -3,6 +3,7 @@ import torch_npu
3		3
4	from torch_npu.npu._recovery import check_npu_tensor_is_safe, mark_all_npu_tensor_unsafe, set_npu_tensor_unsafe_check_flag	4	from torch_npu.npu._recovery import check_npu_tensor_is_safe, mark_all_npu_tensor_unsafe, set_npu_tensor_unsafe_check_flag
5	from torch_npu.testing.testcase import TestCase, run_tests	5	from torch_npu.testing.testcase import TestCase, run_tests
		6	+from torch_npu.npu._recovery import restart_device
6		7
7		8
8	class TestNpu(TestCase):	9	class TestNpu(TestCase):
	@@ -36,6 +37,15 @@ class TestNpu(TestCase):
36	tensor_b.copy_(tensor_a_new)	37	tensor_b.copy_(tensor_a_new)
37	self.assertTrue(check_npu_tensor_is_safe(tensor_b))	38	self.assertTrue(check_npu_tensor_is_safe(tensor_b))
38		39
		40	+ def test_restart_device_with_rebuild(self):
		41	+ torch.npu.set_device(0)
		42	+ restart_device(0, rebuild_all_resources=True)
		43	+ self.assertTrue(True)
		44	+
		45	+ def test_check_npu_tensor_is_safe_invalid_type(self):
		46	+ with self.assertRaises(RuntimeError):
		47	+ check_npu_tensor_is_safe("invalid_tensor")
		48	+
39		49
40	if __name__ == '__main__':	50	if __name__ == '__main__':
41	run_tests()	51	run_tests()

Mtest/npu/test_torch_backends.py+20-0

	@@ -270,6 +270,26 @@ class TorchBackendsApiTestCase(TestCase):
270	with self.assertRaises(AssertionError):	270	with self.assertRaises(AssertionError):
271	npu_mode(ins1, ins2)	271	npu_mode(ins1, ins2)
272		272
		273	+ def test_sdp_kernel_all_disabled(self):
		274	+ torch.npu.enable_flash_sdp(True)
		275	+ torch.npu.enable_mem_efficient_sdp(True)
		276	+ torch.npu.enable_math_sdp(True)
		277	+
		278	+ self.assertTrue(torch.npu.flash_sdp_enabled())
		279	+ self.assertTrue(torch.npu.mem_efficient_sdp_enabled())
		280	+ self.assertTrue(torch.npu.math_sdp_enabled())
		281	+
		282	+ with torch.npu.sdp_kernel(enable_flash=False,
		283	+ enable_mem_efficient=False,
		284	+ enable_math=False):
		285	+ self.assertFalse(torch.npu.flash_sdp_enabled())
		286	+ self.assertFalse(torch.npu.mem_efficient_sdp_enabled())
		287	+ self.assertFalse(torch.npu.math_sdp_enabled())
		288	+
		289	+ self.assertTrue(torch.npu.flash_sdp_enabled())
		290	+ self.assertTrue(torch.npu.mem_efficient_sdp_enabled())
		291	+ self.assertTrue(torch.npu.math_sdp_enabled())
		292	+
273		293
274	if __name__ == "__main__":	294	if __name__ == "__main__":
275	run_tests()	295	run_tests()

Mtest/profiler/analysis/prof_bean/test_ai_cpu_bean.py+85-0

	@@ -2,6 +2,11 @@ from collections import OrderedDict
2		2
3	from torch_npu.profiler.analysis.prof_bean._ai_cpu_bean import AiCpuBean	3	from torch_npu.profiler.analysis.prof_bean._ai_cpu_bean import AiCpuBean
4	from torch_npu.testing.testcase import TestCase, run_tests	4	from torch_npu.testing.testcase import TestCase, run_tests
		5	+from torch_npu.profiler.analysis.prof_bean._hccs_bean import HccsBean
		6	+from torch_npu.profiler.analysis.prof_bean._nic_bean import NicBean
		7	+from torch_npu.profiler.analysis.prof_bean._npu_module_mem_bean import NpuModuleMemoryBean
		8	+from torch_npu.profiler.analysis.prof_bean._pcie_bean import PcieBean
		9	+from torch_npu.profiler.analysis.prof_bean._roce_bean import RoCEBean
5		10
6		11
7	class TestAiCPUBean(TestCase):	12	class TestAiCPUBean(TestCase):
	@@ -41,6 +46,86 @@ class TestAiCPUBean(TestCase):
41	continue	46	continue
42	self.assertEqual(keys, _ai_cpu_bean.headers)	47	self.assertEqual(keys, _ai_cpu_bean.headers)
43		48
		49	+ def test_hccs_bean_constructor_initialization(self):
		50	+ valid_data = OrderedDict()
		51	+ valid_data["Timestamps(us)"] = 1768
		52	+ valid_data["Node"] = "IndexPutV2"
		53	+ valid_data["Compute_time(us)"] = 0.2
		54	+ hccs_bean = HccsBean(valid_data)
		55	+ self.assertEqual([1768, "IndexPutV2", 0.2], hccs_bean.row)
		56	+ self.assertEqual(
		57	+ ["Timestamps(us)", "Node", "Compute_time(us)"],
		58	+ hccs_bean.headers
		59	+ )
		60	+
		61	+ def test_nic_bean_constructor_initialization(self):
		62	+ valid_data = OrderedDict()
		63	+ valid_data["Timestamps(us)"] = 1768
		64	+ valid_data["Node"] = "IndexPutV2"
		65	+ valid_data["Compute_time(us)"] = 0.2
		66	+ nic_bean = NicBean(valid_data)
		67	+ self.assertEqual([1768, "IndexPutV2", 0.2], nic_bean.row)
		68	+ self.assertEqual(
		69	+ ["Timestamps(us)", "Node", "Compute_time(us)"],
		70	+ nic_bean.headers
		71	+ )
		72	+
		73	+ def test_npu_module_memory_bean_headers_property(self):
		74	+ valid_data = OrderedDict()
		75	+ valid_data["Device_id"] = "3"
		76	+ valid_data["Component"] = "TestComponent4"
		77	+ valid_data["Timestamp(us)"] = "111111"
		78	+ valid_data["Total Reserved(KB)"] = "512000"
		79	+ valid_data["Device"] = "NPU3"
		80	+
		81	+ npu_bean = NpuModuleMemoryBean(valid_data)
		82	+ result_headers = npu_bean.headers
		83	+ excepted_headers = [
		84	+ "Device_id", "Component", "Timestamp(us)", "Total Reserved(MB)", "Device"
		85	+ ]
		86	+
		87	+ self.assertEqual(result_headers, excepted_headers)
		88	+
		89	+ def test_npu_module_memory_bean_row_property(self):
		90	+ valid_data = OrderedDict()
		91	+ valid_data["Device_id"] = "2"
		92	+ valid_data["Component"] = "TestComponent3"
		93	+ valid_data["Timestamp(us)"] = "987654"
		94	+ valid_data["Total Reserved(KB)"] = "1024000"
		95	+ valid_data["Device"] = "NPU2"
		96	+
		97	+ npu_bean = NpuModuleMemoryBean(valid_data)
		98	+ result_row = npu_bean.row
		99	+ excepted_row = [
		100	+ "2", "TestComponent3", "987654", "1000.0", "NPU2"
		101	+ ]
		102	+
		103	+ self.assertEqual(result_row, excepted_row)
		104	+
		105	+ def test_pcie_bean_constructor_initialization(self):
		106	+ valid_data = OrderedDict()
		107	+ valid_data["Timestamps(us)"] = 1768
		108	+ valid_data["Node"] = "IndexPutV2"
		109	+ valid_data["Compute_time(us)"] = 0.2
		110	+ pcie_bean = PcieBean(valid_data)
		111	+ self.assertEqual([1768, "IndexPutV2", 0.2], pcie_bean.row)
		112	+ self.assertEqual(
		113	+ ["Timestamps(us)", "Node", "Compute_time(us)"],
		114	+ pcie_bean.headers
		115	+ )
		116	+
		117	+ def test_roce_bean_constructor_initialization(self):
		118	+ valid_data = OrderedDict()
		119	+ valid_data["Timestamps(us)"] = 1768
		120	+ valid_data["Node"] = "IndexPutV2"
		121	+ valid_data["Compute_time(us)"] = 0.2
		122	+ roce_bean = RoCEBean(valid_data)
		123	+ self.assertEqual([1768, "IndexPutV2", 0.2], roce_bean.row)
		124	+ self.assertEqual(
		125	+ ["Timestamps(us)", "Node", "Compute_time(us)"],
		126	+ roce_bean.headers
		127	+ )
		128	+
44		129
45	if __name__ == "__main__":	130	if __name__ == "__main__":
46	run_tests()	131	run_tests()

Mtest/profiler/analysis/prof_common_func/test_file_manager.py+68-0

	@@ -96,6 +96,74 @@ class TestFileManager(TestCase):
96	mock_stat.return_value.st_uid = 9999	96	mock_stat.return_value.st_uid = 9999
97	self.assertFalse(FileManager.check_file_owner(test_path))	97	self.assertFalse(FileManager.check_file_owner(test_path))
98		98
		99	+ def test_check_db_file_valid_invalid_size(self):
		100	+ test_file_path = os.path.join(self.tmp_dir, "invalid_db.db")
		101	+ with open(test_file_path, 'w') as fp:
		102	+ fp.write("a" * 20)
		103	+
		104	+ with patch('torch_npu.profiler.analysis.prof_common_func._file_manager.Constant.MAX_FILE_SIZE', 10):
		105	+ with self.assertRaises(RuntimeError):
		106	+ FileManager.check_db_file_vaild(test_file_path)
		107	+
		108	+ def test_file_read_all_nonexistent_file(self):
		109	+ test_file_path = os.path.join(self.tmp_dir, "nonexistent.log")
		110	+ with patch('torch_npu.profiler.analysis.prof_common_func._file_manager.PathManager.check_directory_path_readable'):
		111	+ result = FileManager.file_read_all(test_file_path)
		112	+ self.assertEqual('', result)
		113	+
		114	+ def test_read_csv_file_empty_file(self):
		115	+ test_file_path = os.path.join(self.tmp_dir, "empty.csv")
		116	+ with open(test_file_path, 'w') as fp:
		117	+ pass
		118	+ result = FileManager.read_csv_file(test_file_path, GeMemoryRecordBean)
		119	+ self.assertEqual([], result)
		120	+
		121	+ def test_create_csv_file_empty_data(self):
		122	+ test_file = "empty_file.csv"
		123	+ FileManager.create_csv_file(self.tmp_dir, [], test_file)
		124	+ test_file_path = os.path.join(self.tmp_dir, test_file)
		125	+ self.assertFalse(os.path.exists(test_file_path))
		126	+
		127	+ def test_read_json_file_exceeds_max_size(self):
		128	+ test_file_path = os.path.join(self.tmp_dir, "large_file.json")
		129	+ with open(test_file_path, 'w') as fp:
		130	+ fp.write("a" * 10000)
		131	+
		132	+ with patch('torch_npu.profiler.analysis.prof_common_func._file_manager.Constant.MAX_FILE_SIZE', 10):
		133	+ result = FileManager.read_json_file(test_file_path)
		134	+ self.assertEqual({}, result)
		135	+
		136	+ def test_file_read_all_exceeds_max_size(self):
		137	+ test_file_path = os.path.join(self.tmp_dir, "large_file.log")
		138	+ with open(test_file_path, 'w') as fp:
		139	+ fp.write("a" * 10000)
		140	+
		141	+ with patch('torch_npu.profiler.analysis.prof_common_func._file_manager.Constant.MAX_FILE_SIZE', 10):
		142	+ result = FileManager.file_read_all(test_file_path)
		143	+ self.assertEqual("", result)
		144	+
		145	+ def test_create_json_file_empty_data(self):
		146	+ test_file = "empty_data.json"
		147	+ FileManager.create_json_file(self.tmp_dir, [], test_file)
		148	+ test_file_path = os.path.join(self.tmp_dir, test_file)
		149	+ self.assertFalse(os.path.exists(test_file_path))
		150	+
		151	+ def test_read_csv_file_exceeds_max_size(self):
		152	+ test_file_path = os.path.join(self.tmp_dir, "large_file.csv")
		153	+ with open(test_file_path, 'w') as fp:
		154	+ fp.write("A" * 1024 * 1024 * 2)
		155	+
		156	+ with patch('torch_npu.profiler.analysis.prof_common_func._file_manager.Constant.MAX_CSV_SIZE', 1024):
		157	+ result = FileManager.read_csv_file(test_file_path, GeMemoryRecordBean)
		158	+ self.assertEqual([], result)
		159	+
		160	+ def test_file_read_all_empty_file(self):
		161	+ test_file_path = os.path.join(self.tmp_dir, "empty_file.log")
		162	+ with os.fdopen(os.open(test_file_path,
		163	+ os.O_WRONLY \| os.O_CREAT, stat.S_IWUSR \| stat.S_IRUSR), 'w') as fp:
		164	+ pass
		165	+ self.assertEqual("", FileManager.file_read_all(test_file_path))
		166	+
99		167
100	if __name__ == "__main__":	168	if __name__ == "__main__":
101	run_tests()	169	run_tests()

Mtest/profiler/test_experimental_config.py+162-0

	@@ -1,4 +1,5 @@
1	import unittest	1	import unittest
		2	+import warnings
2		3
3	from torch_npu.profiler.experimental_config import supported_ai_core_metrics	4	from torch_npu.profiler.experimental_config import supported_ai_core_metrics
4	from torch_npu.profiler.experimental_config import supported_profiler_level	5	from torch_npu.profiler.experimental_config import supported_profiler_level
	@@ -7,6 +8,8 @@ from torch_npu.profiler.analysis.prof_common_func._constant import Constant
7	from torch_npu.profiler.experimental_config import _ExperimentalConfig	8	from torch_npu.profiler.experimental_config import _ExperimentalConfig
8	from torch_npu._C._profiler import _ExperimentalConfig as Cpp_ExperimentalConfig	9	from torch_npu._C._profiler import _ExperimentalConfig as Cpp_ExperimentalConfig
9	from torch_npu.testing.testcase import TestCase, run_tests	10	from torch_npu.testing.testcase import TestCase, run_tests
		11	+from torch_npu.profiler.analysis.prof_common_func._constant import Constant, print_warn_msg, print_info_msg
		12	+from torch_npu.profiler.analysis.prof_common_func._cann_package_manager import CannPackageManager
10		13
11		14
12	class TestExperimentalConfig(TestCase):	15	class TestExperimentalConfig(TestCase):
	@@ -179,6 +182,165 @@ class TestExperimentalConfig(TestCase):
179	self.assertEqual(True, experimental_config._sys_io)	182	self.assertEqual(True, experimental_config._sys_io)
180	self.assertEqual(True, experimental_config._sys_interconnection)	183	self.assertEqual(True, experimental_config._sys_interconnection)
181		184
		185	+ def test_check_params_reset_data_simplification(self):
		186	+ experimental_config = _ExperimentalConfig(data_simplification="invalid")
		187	+ self.assertEqual(True, experimental_config._data_simplification)
		188	+
		189	+ def test_check_params_reset_l2_cache(self):
		190	+ experimental_config = _ExperimentalConfig(l2_cache="invalid")
		191	+ self.assertEqual(False, experimental_config._l2_cache)
		192	+
		193	+ def test_check_params_reset_invalid_profiler_level(self):
		194	+ experimental_config = _ExperimentalConfig(profiler_level=999)
		195	+ self.assertEqual(Constant.LEVEL0, experimental_config._profiler_level)
		196	+
		197	+ def test_check_params_reset_aic_metrics_level0(self):
		198	+ experimental_config = _ExperimentalConfig(profiler_level=Constant.LEVEL0, aic_metrics=Constant.AicMemory)
		199	+ self.assertEqual(Constant.AicMetricsNone, experimental_config._aic_metrics)
		200	+
		201	+ def test_convert_export_type_string(self):
		202	+ experimental_config = _ExperimentalConfig(export_type="text")
		203	+ self.assertEqual(["text"], experimental_config._export_type)
		204	+
		205	+ def test_check_params_invalid_gc_detect_threshold(self):
		206	+ experimental_config = _ExperimentalConfig(gc_detect_threshold=-1.0)
		207	+ self.assertIsNone(experimental_config._gc_detect_threshold)
		208	+
		209	+ def test_check_host_sys_params_invalid_elements(self):
		210	+ experimental_config = _ExperimentalConfig(host_sys=[Constant.CPU, "invalid"])
		211	+ self.assertEqual([], experimental_config._host_sys)
		212	+
		213	+ def test_check_params_invalid_record_op_args(self):
		214	+ experimental_config = _ExperimentalConfig(record_op_args="invalid")
		215	+ self.assertEqual(False, experimental_config.record_op_args)
		216	+
		217	+ def test_get_proxy_returns_none_no_failure(self):
		218	+ import sys
		219	+ from unittest.mock import patch
		220	+
		221	+ with patch.dict(sys.modules, {'IPCMonitor': None}):
		222	+ from torch_npu.profiler._dynamic_profiler._dynamic_monitor_proxy import PyDynamicMonitorProxySingleton
		223	+ singleton = PyDynamicMonitorProxySingleton()
		224	+ singleton._proxy = None
		225	+ singleton._load_success = True
		226	+ result = singleton.get_proxy()
		227	+ self.assertIsNone(result)
		228	+
		229	+ def test_load_proxy_import_failure(self):
		230	+ import sys
		231	+ from unittest.mock import patch
		232	+
		233	+ with patch.dict(sys.modules, {'IPCMonitor': None}):
		234	+ from torch_npu.profiler._dynamic_profiler._dynamic_monitor_proxy import PyDynamicMonitorProxySingleton
		235	+ singleton = PyDynamicMonitorProxySingleton()
		236	+ singleton._proxy = None
		237	+ singleton._load_success = True
		238	+ singleton._load_proxy()
		239	+ self.assertFalse(singleton._load_success)
		240	+ self.assertIsNone(singleton._proxy)
		241	+
		242	+ def test_load_proxy_initialization_success(self):
		243	+ import sys
		244	+ from unittest.mock import MagicMock, patch
		245	+
		246	+ mock_proxy_class = MagicMock()
		247	+ mock_proxy_instance = MagicMock()
		248	+ mock_proxy_class.return_value = mock_proxy_instance
		249	+
		250	+ with patch.dict(sys.modules, {'IPCMonitor': MagicMock()}):
		251	+ with patch('IPCMonitor.PyDynamicMonitorProxy', mock_proxy_class):
		252	+ from torch_npu.profiler._dynamic_profiler._dynamic_monitor_proxy import PyDynamicMonitorProxySingleton
		253	+ singleton = PyDynamicMonitorProxySingleton()
		254	+ singleton._proxy = None
		255	+ singleton._load_success = True
		256	+ singleton._load_proxy()
		257	+ self.assertTrue(singleton._load_success)
		258	+ self.assertEqual(singleton._proxy, mock_proxy_instance)
		259	+
		260	+ def test_gc_detector_stop(self):
		261	+ from torch_npu.profiler._profiler_gc_detect import ProfGCDetector
		262	+ from unittest.mock import patch, MagicMock
		263	+ import gc
		264	+ detector = ProfGCDetector(1.0)
		265	+ detector.start()
		266	+ detector.save_info = [(1, 2, 3)]
		267	+ original_callbacks = gc.callbacks[:]
		268	+
		269	+ try:
		270	+ with patch('torch_npu.profiler._profiler_path_creator.ProfPathCreator') as mock_creator, \
		271	+ patch(
		272	+ 'torch_npu.profiler._profiler_gc_detect.ProfilerPathManager.get_fwk_path') as mock_get_fwk_path:
		273	+ mock_creator_instance = MagicMock()
		274	+ mock_creator_instance.get_prof_dir.return_value = '/mock/path'
		275	+ mock_creator.return_value = mock_creator_instance
		276	+ mock_get_fwk_path.return_value = '/mock/path'
		277	+ detector.stop()
		278	+ self.assertEqual(detector.time_info, {})
		279	+ self.assertEqual(detector.save_info, [])
		280	+ self.assertNotIn(detector.gc_callback, gc.callbacks)
		281	+ finally:
		282	+ if detector.gc_callback in gc.callbacks:
		283	+ gc.callbacks.remove(detector.gc_callback)
		284	+
		285	+ def test_gc_detector_save_file_creation_failure(self):
		286	+ from torch_npu.profiler._profiler_gc_detect import ProfGCDetector
		287	+ from unittest.mock import patch, MagicMock
		288	+ import os
		289	+ detector = ProfGCDetector(1.0)
		290	+ detector.save_info = [(1, 2, 3)]
		291	+
		292	+ with patch('torch_npu.profiler._profiler_path_creator.ProfPathCreator') as mock_creator, \
		293	+ patch('torch_npu.profiler._profiler_gc_detect.ProfilerPathManager.get_fwk_path') as mock_get_fwk_path, \
		294	+ patch('torch_npu.profiler._profiler_gc_detect.FileManager.create_bin_file_by_path') as mock_create_file:
		295	+ mock_creator_instance = MagicMock()
		296	+ mock_creator_instance.get_prof_dir.return_value = '/mock/path'
		297	+ mock_creator.return_value = mock_creator_instance
		298	+ mock_get_fwk_path.return_value = '/mock/path'
		299	+ mock_create_file.side_effect = Exception("File creation failed")
		300	+ detector.save()
		301	+
		302	+ def test_gc_callback_valid_phases(self):
		303	+ from torch_npu.profiler._profiler_gc_detect import ProfGCDetector
		304	+ import gc
		305	+ import os
		306	+
		307	+ detector = ProfGCDetector(0.001)
		308	+ detector.start()
		309	+
		310	+ try:
		311	+ detector.gc_callback(detector.START_PHASE, {})
		312	+ pid = os.getpid()
		313	+ self.assertIn(pid, detector.time_info)
		314	+ detector.gc_callback(detector.STOP_PHASE, {})
		315	+ self.assertEqual(len(detector.save_info), 1)
		316	+ detector.gc_callback("invalid", {})
		317	+ finally:
		318	+ if detector.gc_callback in gc.callbacks:
		319	+ gc.callbacks.remove(detector.gc_callback)
		320	+
		321	+ def test_gc_detector_start(self):
		322	+ from torch_npu.profiler._profiler_gc_detect import ProfGCDetector
		323	+ import gc
		324	+
		325	+ detector = ProfGCDetector(1.0)
		326	+ original_callbacks = gc.callbacks[:]
		327	+
		328	+ try:
		329	+ detector.start()
		330	+ self.assertIn(detector.gc_callback, gc.callbacks)
		331	+ finally:
		332	+ if detector.gc_callback in gc.callbacks:
		333	+ gc.callbacks.remove(detector.gc_callback)
		334	+
		335	+ def test_gc_detector_init(self):
		336	+ from torch_npu.profiler._profiler_gc_detect import ProfGCDetector
		337	+
		338	+ detector = ProfGCDetector(1.0)
		339	+ self.assertEqual(detector.threshold, 1.0 * Constant.NS_TO_MS)
		340	+ self.assertEqual(detector.time_info, {})
		341	+ self.assertEqual(detector.save_info, [])
		342	+ self.assertIsNotNone(detector.get_cur_ts)
		343	+
182		344
183	if __name__ == "__main__":	345	if __name__ == "__main__":
184	run_tests()	346	run_tests()