AI assist developer for python DT second batch for 2.9.0-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/distributed/rpc/test_single_machine_multi_devices.py+64-0

	@@ -10,11 +10,14 @@ from torch import multiprocessing as mp
10	from torch import nn, Tensor	10	from torch import nn, Tensor
11	from torch.distributed.nn.api.remote_module import RemoteModule	11	from torch.distributed.nn.api.remote_module import RemoteModule
12	from torch.distributed.rpc import WorkerInfo, PyRRef	12	from torch.distributed.rpc import WorkerInfo, PyRRef
		13	+from torch._C import _get_privateuse1_backend_name
13		14
14	import torch_npu	15	import torch_npu
15	from torch_npu.distributed.rpc.options import NPUTensorPipeRpcBackendOptions	16	from torch_npu.distributed.rpc.options import NPUTensorPipeRpcBackendOptions
16	from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU	17	from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU
17	from torch_npu.testing.testcase import TestCase, run_tests	18	from torch_npu.testing.testcase import TestCase, run_tests
		19	+from torch_npu.distributed.rpc.options import _to_device_map
		20	+from torch_npu.distributed.rpc.options import _to_device
18		21
19		22
20	class TestRpc(TestCase):	23	class TestRpc(TestCase):
	@@ -389,6 +392,67 @@ class TestRpc(TestCase):
389	def test_local_value_npu(self):	392	def test_local_value_npu(self):
390	self._test_multiprocess(TestRpc._test_local_value, [], self.world_size_2p)	393	self._test_multiprocess(TestRpc._test_local_value, [], self.world_size_2p)
391		394
		395	+ @skipIfUnsupportMultiNPU(2)
		396	+ def test_set_devices(self):
		397	+ options = NPUTensorPipeRpcBackendOptions()
		398	+ devices = ["npu:0", "npu:1"]
		399	+ options.set_devices(devices)
		400	+ self.assertEqual(len(options.devices), 2)
		401	+ self.assertEqual(options.devices[0].type, _get_privateuse1_backend_name())
		402	+ self.assertEqual(options.devices[1].type, _get_privateuse1_backend_name())
		403	+
		404	+ @skipIfUnsupportMultiNPU(2)
		405	+ def test_set_device_map_conflicting(self):
		406	+ options = NPUTensorPipeRpcBackendOptions()
		407	+ device_map1 = {'npu:0': 'npu:1'}
		408	+ options.set_device_map('worker1', device_map1)
		409	+
		410	+ device_map2 = {'npu:0': 'npu:2'}
		411	+ with self.assertRaises(ValueError):
		412	+ options.set_device_map('worker1', device_map2)
		413	+
		414	+ @skipIfUnsupportMultiNPU(2)
		415	+ def test_set_device_map_valid(self):
		416	+ options = NPUTensorPipeRpcBackendOptions()
		417	+ device_map = {'npu:0': 'npu:1'}
		418	+ options.set_device_map('worker1', device_map)
		419	+ self.assertEqual(len(options.device_maps), 1)
		420	+ self.assertIn('worker1', options.device_maps)
		421	+ self.assertEqual(len(options.device_maps["worker1"]), 1)
		422	+
		423	+ @skipIfUnsupportMultiNPU(2)
		424	+ def test_options_with_device_maps(self):
		425	+ device_maps = {'worker1': {'npu:0': 'npu:1'}}
		426	+ options = NPUTensorPipeRpcBackendOptions(device_maps=device_maps)
		427	+ self.assertEqual(len(options.device_maps), 1)
		428	+ self.assertIn('worker1', options.device_maps)
		429	+ self.assertEqual(len(options.device_maps["worker1"]), 1)
		430	+
		431	+ @skipIfUnsupportMultiNPU(2)
		432	+ def test_device_map_invalid_value(self):
		433	+ device_map = {'npu:0': 'npu:1', 'npu:2': 'npu:1'}
		434	+ with self.assertRaises(ValueError):
		435	+ _to_device_map(device_map)
		436	+
		437	+ @skipIfUnsupportMultiNPU(2)
		438	+ def test_device_map_effective(self):
		439	+ device_map = {'npu:0': 'npu:1', 'npu:2': 'npu:3'}
		440	+ result = _to_device_map(device_map)
		441	+ self.assertEqual(len(result), 2)
		442	+ self.assertEqual(result[torch.device('npu:0')], torch.device('npu:1'))
		443	+ self.assertEqual(result[torch.device('npu:2')], torch.device('npu:3'))
		444	+
		445	+ @skipIfUnsupportMultiNPU(2)
		446	+ def test_to_device_invalid_device_type(self):
		447	+ with self.assertRaises(ValueError):
		448	+ _to_device('cpu:0')
		449	+
		450	+ @skipIfUnsupportMultiNPU(2)
		451	+ def test_to_device_valid_npu_device(self):
		452	+ device = _to_device('npu:0')
		453	+ self.assertEqual(device.type, _get_privateuse1_backend_name())
		454	+ self.assertEqual(device.index, 0)
		455	+
392		456
393	if __name__ == '__main__':	457	if __name__ == '__main__':
394	run_tests()	458	run_tests()

Atest/distributed/test_rendezvous.py+47-0

@@ -0,0 +1,47 @@
	1	+import os
	2	+import logging
	3	+from datetime import timedelta
	4	+from unittest.mock import patch
	5	+
	6	+from torch.distributed.rendezvous import register_rendezvous_handler as register_rendezvous_handler
	7	+from torch._C._distributed_c10d import _DEFAULT_PG_TIMEOUT
	8	+from torch.distributed import Store, PrefixStore
	9	+from torch.distributed.elastic.rendezvous.api import RendezvousParameters, RendezvousHandler, RendezvousInfo, RendezvousStoreInfo
	10	+from torch.distributed.elastic.rendezvous.api import rendezvous_handler_registry as handler_registry
	11	+from torch.distributed.elastic.rendezvous.utils import parse_rendezvous_endpoint
	12	+from torch_npu.distributed.run import parse_args as torch_parse_cmd_args
	13	+from torch_npu.testing.testcase import TestCase, run_tests
	14	+from torch_npu.distributed import ParallelStore
	15	+from torch_npu.distributed.rendezvous import _create_c10d_store
	16	+from torch_npu.distributed.rendezvous import _rendezvous_error
	17	+from torch_npu.distributed.rendezvous import _torchelastic_use_agent_store
	18	+from torch_npu.distributed.rendezvous import _parallel_rendezvous_handler
	19	+from torch_npu.distributed.rendezvous import _create_parallel_handler
	20	+from torch_npu.distributed.rendezvous import _rendezvous_init
	21	+
	22	+
	23	+class TestRendezvous(TestCase):
	24	+ def test_create_c10d_store_without_agent_store(self):
	25	+ with patch.dict(os.environ, {
	26	+ "TORCH_NPU_ELASTIC_USE_AGENT_STORE": "False",
	27	+ "PROXY_AGENT_PID_USE_LOCAL_SOCKET_PATH": "12345"}):
	28	+ store = _create_c10d_store("localhost", 12345, 0, 2, timedelta(seconds=600))
	29	+ self.assertIsInstance(store, ParallelStore)
	30	+
	31	+ def test_create_c10d_store_invalid_port(self):
	32	+ with self.assertRaises(ValueError) as context:
	33	+ _create_c10d_store("localhost", 70000, 0, 2, timedelta(seconds=600))
	34	+ self.assertIn("port must have value from 0 to 65535", str(context.exception))
	35	+
	36	+ def test_parallel_rendezvous_handler_missing_rank(self):
	37	+ with self.assertRaises(ValueError) as context:
	38	+ list(_parallel_rendezvous_handler("parallel://localhost:12345?world_size=2"))
	39	+ self.assertIn("rank parameter missing", str(context.exception))
	40	+
	41	+ def test_torchelastic_use_agent_store_true(self):
	42	+ with patch.dict(os.environ, {"TORCH_NPU_ELASTIC_USE_AGENT_STORE": "True"}):
	43	+ self.assertTrue(_torchelastic_use_agent_store())
	44	+
	45	+
	46	+if __name__ == '__main__':
	47	+ run_tests()

Atest/distributed/test_run.py+46-0

@@ -0,0 +1,46 @@
	1	+from unittest.mock import patch
	2	+
	3	+from torch.distributed import run as torch_run
	4	+from torch.distributed.argparse_util import check_env, env
	5	+from torch.distributed.run import get_args_parser
	6	+from torch.distributed.elastic.multiprocessing.errors import record
	7	+import torch_npu
	8	+from torch_npu.distributed.run import parse_args
	9	+from torch_npu.distributed.run import _main
	10	+from torch_npu.testing.testcase import TestCase, run_tests
	11	+
	12	+
	13	+class TestRun(TestCase):
	14	+ def test_main_function_default_args(self):
	15	+ with patch('torch_npu.distributed.run.torch_run.run') as mock_run:
	16	+ _main(["dummy_script.py"])
	17	+ mock_run.assert_called_once()
	18	+ call_args = mock_run.call_args[0][0]
	19	+ self.assertEqual(call_args.rdzv_backend, "parallel")
	20	+ self.assertIsNotNone(call_args.rdzv_endpoint)
	21	+
	22	+ def test_parse_args_default_tiered_parallel_tcpstore(self):
	23	+ args = ["--nproc_per_node", "1", "dummy_script.py"]
	24	+ parsed_args = parse_args(args)
	25	+ self.assertEqual(parsed_args.enable_tiered_parallel_tcpstore, "false")
	26	+
	27	+ def test_main_function_with_existing_rdzv_endpoint(self):
	28	+ with patch('torch_npu.distributed.run.torch_run.run') as mock_run:
	29	+ import argparse
	30	+ args = argparse.Namespace(nproc_per_node=1,
	31	+ master_addr='localhost',
	32	+ master_port=12345,
	33	+ rdzv_backend=None,
	34	+ rdzv_endpoint="existing_endpoint:54321")
	35	+
	36	+ with patch('torch_npu.distributed.run.parse_args', return_value=args):
	37	+ _main(None)
	38	+
	39	+ mock_run.assert_called_once()
	40	+ call_args = mock_run.call_args[0][0]
	41	+ self.assertEqual(call_args.rdzv_backend, 'parallel')
	42	+ self.assertEqual(call_args.rdzv_endpoint, "existing_endpoint:54321")
	43	+
	44	+
	45	+if __name__ == "__main__":
	46	+ 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()

Atest/npu/test_sanitizer.py+50-0

@@ -0,0 +1,50 @@
	1	+import os
	2	+import atexit
	3	+from unittest.mock import MagicMock
	4	+from unittest.mock import patch
	5	+
	6	+import torch.cuda._sanitizer as csan
	7	+import torch_npu
	8	+import torch_npu.utils._npu_trace as npu_trace
	9	+import torch_npu.npu._stream_check as stream_check
	10	+import torch_npu.npu._kernel_check as kernel_check
	11	+from torch_npu.utils.utils import _print_warn_log
	12	+import torch_npu.npu._sanitizer as sanitizer
	13	+from torch_npu.testing.testcase import TestCase, run_tests
	14	+
	15	+
	16	+class TestSanitizer(TestCase):
	17	+ def test_del_method_with_dispatch(self):
	18	+ mock_dispatch = MagicMock()
	19	+ sanitizer.npu_sanitizer.dispatch = mock_dispatch
	20	+ sanitizer.npu_sanitizer.__del__()
	21	+ mock_dispatch.__exit__.assert_called_once_with(None, None, None)
	22	+
	23	+ def test_enable_kernel_check_no_debug_path(self):
	24	+ with patch.dict(os.environ, {}, clear=True):
	25	+ result = sanitizer.npu_sanitizer.enable_kernel_check()
	26	+ self.assertFalse(result)
	27	+
	28	+ def test_enable_stream_check_mode(self):
	29	+ with patch.dict(os.environ, {}, clear=True):
	30	+ with patch('torch.cuda._sanitizer.EventHandler') as mock_event_handler, \
	31	+ patch('torch_npu.npu._stream_check.NPUSanitizerDispatchMode') as mock_dispach, \
	32	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_event_creation'), \
	33	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_event_deletion'), \
	34	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_event_record'), \
	35	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_event_wait'), \
	36	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_memory_allocation'), \
	37	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_memory_deallocation'), \
	38	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_stream_creation'), \
	39	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_device_synchronization'), \
	40	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_stream_synchronization'), \
	41	+ patch('torch_npu.utils._npu_trace.register_callback_for_npu_event_synchronization'):
	42	+ mock_dispatch_instance = mock_dispach.return_value
	43	+ mock_dispatch_instance.__enter__.return_value = None
	44	+ sanitizer.npu_sanitizer.enable()
	45	+ self.assertTrue(sanitizer.npu_sanitizer.enable)
	46	+ self.assertEqual(sanitizer.npu_sanitizer.mode, sanitizer.SanitizerMode.STREAM)
	47	+
	48	+
	49	+if __name__ == "__main__":
	50	+ run_tests()

Atest/npu/test_stream_check.py+67-0

@@ -0,0 +1,67 @@
	1	+import sys
	2	+import logging
	3	+from unittest import mock
	4	+
	5	+import torch
	6	+import torch.cuda._sanitizer as csan
	7	+from torch.utils._python_dispatch import TorchDispatchMode
	8	+import torch_npu
	9	+import torch_npu.npu._stream_check as stream_check
	10	+from torch_npu.testing.testcase import TestCase, run_tests
	11	+
	12	+
	13	+class TestStreamCheck(TestCase):
	14	+ def test_parse_methods_with_valid_inputs(self):
	15	+ mock_event_handler = mock.MagicMock()
	16	+ mode = stream_check.NPUSanitizerDispatchMode(mock_event_handler)
	17	+
	18	+ mock_schema = mock.MagicMock()
	19	+ args = (torch.tensor([1.0]), torch.tensor([2.0]))
	20	+ kwargs = {'test_args': torch.tensor([3.0])}
	21	+
	22	+ mode.args_handler = mock.MagicMock()
	23	+ mode.parse_inputs(mock_schema, args, kwargs)
	24	+ mode.args_handler.parse_inputs.assert_called_once_with(mock_schema, args, kwargs)
	25	+ mock_outputs = [torch.tensor([4.0])]
	26	+ mode.parse_outputs(mock_outputs)
	27	+ mode.args_handler.parse_outputs.assert_called_once_with(mock_outputs)
	28	+
	29	+ def test_torch_dispatch_success(self):
	30	+ mock_event_handler = mock.MagicMock()
	31	+ mode = stream_check.NPUSanitizerDispatchMode(mock_event_handler)
	32	+ mock_func = mock.MagicMock()
	33	+ mock_func.__name__ = "aten::add"
	34	+ mock_func._schema = mock.MagicMock()
	35	+ mock_args = (torch.tensor([1.0]), torch.tensor([2.0]))
	36	+ mock_kwargs = {}
	37	+
	38	+ with mock.patch('torch_npu.npu.current_stream') as mock_stream:
	39	+ mock_stream_instance = mock.MagicMock()
	40	+ mock_stream_instance.npu_stream = 1
	41	+ mock_stream.return_value = mock_stream_instance
	42	+
	43	+ with mock.patch.object(mode, 'parse_inputs') as mock_parse_inputs, \
	44	+ mock.patch.object(mode, 'parse_outputs') as mock_parse_outputs, \
	45	+ mock.patch.object(mode, 'check_errors') as mock_check_errors:
	46	+ result = mode.__torch_dispatch__(mock_func, [], mock_args, mock_kwargs)
	47	+ mock_parse_inputs.assert_called_once()
	48	+ mock_parse_outputs.assert_called_once()
	49	+ mock_check_errors.assert_called_once()
	50	+
	51	+ def test_enable_autograd_with_matching_api(self):
	52	+ mock_event_handler = mock.MagicMock()
	53	+ mode = stream_check.NPUSanitizerDispatchMode(mock_event_handler)
	54	+ with mock.patch('torch._C._dispatch_tls_set_dispatch_key_excluded') as mock_set_dispatch:
	55	+ mode.enable_autograd("adaptive_avg_pool2d")
	56	+ mock_set_dispatch.assert_called_once_with(torch._C.DispatchKey.AutogradFunctionality, False)
	57	+
	58	+ def test_init_with_event_handler(self):
	59	+ mock_event_handler = mock.MagicMock()
	60	+ mode = stream_check.NPUSanitizerDispatchMode(mock_event_handler)
	61	+ self.assertEqual(mode.event_handler, mock_event_handler)
	62	+ self.assertIsNone(mode.args_handler)
	63	+ self.assertEqual(mode.npu_adjust_autograd, ["adaptive_avg_pool2d", "batch_norm", "log_softmax", "nll_loss", "to"])
	64	+
	65	+
	66	+if __name__ == "__main__":
	67	+ 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()

Atest/optim/test_npu_fused_rmsprop_tf.py+92-0

@@ -0,0 +1,92 @@
	1	+from collections import defaultdict
	2	+import torch
	3	+from torch_npu.utils import npu_combine_tensors
	4	+from torch_npu.utils._error_code import ErrCode, pta_error
	5	+from torch_npu.optim.npu_fused_optim_base import NpuFusedOptimizerBase
	6	+import torch_npu.optim.npu_fused_rmsprop_tf as npu_fused_rmsprop_tf
	7	+from torch_npu.testing.testcase import TestCase, run_tests
	8	+
	9	+
	10	+class TestNpuFusedRmsPropTf(TestCase):
	11	+ def test_init_param_state_with_momentum_centered(self):
	12	+ param = torch.tensor([1.0, 2.0])
	13	+ optimizer = npu_fused_rmsprop_tf.NpuFusedRMSpropTF([param], momentum=0.9, centered=True)
	14	+ optimizer._init_param_state(param, momentum=0.9, centered=True)
	15	+ state = optimizer.state[param]
	16	+ self.assertIn('step', state)
	17	+ self.assertIn('square_avg', state)
	18	+ self.assertIn('momentum_buffer', state)
	19	+ self.assertIn('grad_avg', state)
	20	+ self.assertEqual(state['step'], 0)
	21	+ self.assertTrue(torch.allclose(state['square_avg'], torch.ones_like(param)))
	22	+
	23	+ def test_maybe_init_combined_states(self):
	24	+ param = torch.tensor([1.0, 2.0])
	25	+ optimizer = npu_fused_rmsprop_tf.NpuFusedRMSpropTF([param])
	26	+ optimizer.params_lists_indexed_by_group = [[param]]
	27	+ optimizer._maybe_init_combined_states()
	28	+ self.assertTrue(optimizer.is_states_combined)
	29	+ self.assertIsNotNone(optimizer.combined_param_states_indexed_by_group[0])
	30	+
	31	+ def test_maybe_init_combined_states_already_combined(self):
	32	+ param = torch.tensor([1.0, 2.0])
	33	+ optimizer = npu_fused_rmsprop_tf.NpuFusedRMSpropTF([param])
	34	+ optimizer.is_states_combined = True
	35	+ optimizer._maybe_init_combined_states()
	36	+ self.assertTrue(optimizer.is_states_combined)
	37	+
	38	+ def test_setstate_backward_compatibility(self):
	39	+ param = torch.tensor([1.0, 2, 0])
	40	+ optimizer = npu_fused_rmsprop_tf.NpuFusedRMSpropTF([param])
	41	+
	42	+ old_state = {
	43	+ 'param_groups': [
	44	+ {
	45	+ 'params': [param],
	46	+ 'lr': 1e-2,
	47	+ 'weight_decay': 0,
	48	+ 'momentum': 0,
	49	+ 'centered': False
	50	+ }
	51	+ ]
	52	+ }
	53	+
	54	+ optimizer.__setstate__(old_state)
	55	+
	56	+ group = optimizer.param_groups[0]
	57	+ self.assertEqual(group['momentum'], 0)
	58	+ self.assertEqual(group['centered'], False)
	59	+
	60	+ def test_init_param_state(self):
	61	+ optimizer = npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], momentum=0.9, centered=True)
	62	+ p = torch.tensor([1.0])
	63	+ optimizer._init_param_state(p, momentum=0.9, centered=True)
	64	+ state = optimizer.state[p]
	65	+ self.assertIn('step', state)
	66	+ self.assertIn('square_avg', state)
	67	+ self.assertIn('momentum_buffer', state)
	68	+ self.assertIn('grad_avg', state)
	69	+
	70	+ def test_invalid_alpha(self):
	71	+ with self.assertRaises(ValueError):
	72	+ npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], alpha=-0.9)
	73	+
	74	+ def test_invalid_weight_decay(self):
	75	+ with self.assertRaises(ValueError):
	76	+ npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], weight_decay=-1e-2)
	77	+
	78	+ def test_invalid_momentum(self):
	79	+ with self.assertRaises(ValueError):
	80	+ npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], momentum=-0.1)
	81	+
	82	+ def test_invalid_epsilon(self):
	83	+ with self.assertRaises(ValueError):
	84	+ npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], eps=-1e-10)
	85	+
	86	+ def test_invalid_learning_rate(self):
	87	+ with self.assertRaises(ValueError):
	88	+ npu_fused_rmsprop_tf.NpuFusedRMSpropTF([torch.tensor([1.0])], lr=-1e-2)
	89	+
	90	+
	91	+if __name__ == "__main__":
	92	+ 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()

Atest/profiler/analysis/prof_common_func/test_binary_decoder.py+24-0

@@ -0,0 +1,24 @@
	1	+import torch_npu.npu._sanitizer as sanitizer
	2	+import torch_npu.profiler.analysis.prof_common_func._binary_decoder as binary_decoder
	3	+from torch_npu.testing.testcase import TestCase, run_tests
	4	+
	5	+
	6	+class TestBinaryDecoder(TestCase):
	7	+ def test_decode_multiple_structures(self):
	8	+ class MockBean:
	9	+ def __init__(self, data):
	10	+ self.data = data
	11	+
	12	+ def __eq__(self, other):
	13	+ return isinstance(other, MockBean) and self.data == other.data
	14	+
	15	+ test_bytes = b'\x01\x02\x03\x04\x05\x06'
	16	+ result = binary_decoder.BinaryDecoder.decode(test_bytes, MockBean, 2)
	17	+ self.assertEqual(len(result), 3)
	18	+ self.assertEqual(result[0].data, b'\x01\x02')
	19	+ self.assertEqual(result[1].data, b'\x03\x04')
	20	+ self.assertEqual(result[2].data, b'\x05\x06')
	21	+
	22	+
	23	+if __name__ == "__main__":
	24	+ 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_dynamic_profiler.py+89-0

	@@ -3,6 +3,8 @@ import stat
3	import copy	3	import copy
4	import json	4	import json
5	import time	5	import time
		6	+import unittest.mock as mock
		7	+from unittest.mock import patch, MagicMock
6		8
7	import torch	9	import torch
8	from torch_npu.utils._path_manager import PathManager	10	from torch_npu.utils._path_manager import PathManager
	@@ -12,6 +14,7 @@ from torch_npu.testing.testcase import TestCase, run_tests
12	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_config_context import ConfigContext	14	from torch_npu.profiler._dynamic_profiler._dynamic_profiler_config_context import ConfigContext
13	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
14	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
15		18
16		19
17	class SmallModel(torch.nn.Module):	20	class SmallModel(torch.nn.Module):
	@@ -560,6 +563,92 @@ class TestDynamicProfiler(TestCase):
560	return True	563	return True
561	return False	564	return False
562		565
		566	+ def test_out_log_dyno_model(self):
		567	+ original_model = DynamicProfilerUtils.DYNAMIC_PROFILER_MODEL
		568	+ DynamicProfilerUtils.DYNAMIC_PROFILER_MODEL = DynamicProfilerUtils.DynamicProfilerConfigModel.DYNO_CONFIG
		569	+ original_stdout_log = DynamicProfilerUtils.stdout_log
		570	+ log_calls = []
		571	+
		572	+ def log_function(infos, level):
		573	+ log_calls.append((infos, level))
		574	+
		575	+ DynamicProfilerUtils.stdout_log = log_function
		576	+
		577	+ try:
		578	+ DynamicProfilerUtils.out_log("test information", DynamicProfilerUtils.LoggerLevelEnum.INFO)
		579	+ DynamicProfilerUtils.out_log("test warning", DynamicProfilerUtils.LoggerLevelEnum.WARNING)
		580	+ DynamicProfilerUtils.out_log("test error", DynamicProfilerUtils.LoggerLevelEnum.ERROR)
		581	+
		582	+ self.assertEqual(len(log_calls), 3)
		583	+ self.assertEqual(log_calls[0], ("test information", DynamicProfilerUtils.LoggerLevelEnum.INFO))
		584	+ self.assertEqual(log_calls[1], ("test warning", DynamicProfilerUtils.LoggerLevelEnum.WARNING))
		585	+ self.assertEqual(log_calls[2], ("test error", DynamicProfilerUtils.LoggerLevelEnum.ERROR))
		586	+ finally:
		587	+ DynamicProfilerUtils.DYNAMIC_PROFILER_MODEL = original_model
		588	+ DynamicProfilerUtils.stdout_log = original_stdout_log
		589	+
		590	+ def test_clean_shm_for_killed_pid_time_none(self):
		591	+ shm_instance = DynamicProfilerShareMemory(self.results_path, self.cfg_path, 0)
		592	+ with mock.patch(
		593	+ 'torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor_shm.DynamicProfilerShareMemory._get_pid_st_ctime',
		594	+ return_value=None):
		595	+ try:
		596	+ shm_instance._clean_shm_for_killed()
		597	+ except Exception:
		598	+ self.fail("_clean_shm_for_killed should not raise exception when pid_time is None")
		599	+
		600	+ def test_create_shm_over_py38_retry_failure(self):
		601	+ from multiprocessing import shared_memory
		602	+ with mock.patch('multiprocessing.resource_tracker.register', lambda args, *kwargs: None):
		603	+ with mock.patch.object(shared_memory.SharedMemory, '__init__', side_effect=FileNotFoundError("Test error")):
		604	+ with self.assertRaises(RuntimeError):
		605	+ shm_instance = DynamicProfilerShareMemory(self.results_path, self.cfg_path, 0)
		606	+ shm_instance._create_shm_over_py38()
		607	+
		608	+ def test_get_pid_st_ctime_exception(self):
		609	+ shm_instance = DynamicProfilerShareMemory(self.results_path, self.cfg_path, 0)
		610	+ with mock.patch('os.open', side_effect=Exception("Test exception")):
		611	+ result = shm_instance._get_pid_st_ctime(12345)
		612	+ self.assertIsNone(result)
		613	+
		614	+ def test_call_dyno_monitor_with_proxy(self):
		615	+ from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor import DynamicProfilerMonitor
		616	+
		617	+ monitor = DynamicProfilerMonitor()
		618	+ mock_proxy = MagicMock()
		619	+ with patch('torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor.PyDynamic' \
		620	+ 'MonitorProxySingleton') as mock_singleton:
		621	+ mock_singleton_instance = MagicMock()
		622	+ mock_singleton_instance.get_proxy.return_value = mock_proxy
		623	+ mock_singleton.return_value = mock_singleton_instance
		624	+ test_data = {'key': 'value', 'number': 123}
		625	+ monitor._call_dyno_monitor(test_data)
		626	+ mock_proxy.enable_dyno_npu_monitor.assert_called_once()
		627	+
		628	+ def test_clean_resource_with_process(self):
		629	+ from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor import DynamicProfilerMonitor
		630	+ monitor = DynamicProfilerMonitor()
		631	+ monitor._process = MagicMock()
		632	+ with patch.object(monitor, '_shared_loop_flag') as mock_flag:
		633	+ monitor.clean_resource()
		634	+ mock_flag.value = False
		635	+ monitor._process.join.assert_called_once()
		636	+
		637	+ def test_shm_to_prof_conf_context_read_time_exception(self):
		638	+ from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor import DynamicProfilerMonitor
		639	+
		640	+ monitor = DynamicProfilerMonitor()
		641	+ with patch.object(monitor._shm_obj, 'read_bytes', side_effect=Exception("Read error")):
		642	+ result = monitor.shm_to_prof_conf_context()
		643	+ self.assertIsNone(result)
		644	+
		645	+ def test_shm_to_prof_conf_context_none_shm(self):
		646	+ from torch_npu.profiler._dynamic_profiler._dynamic_profiler_monitor import DynamicProfilerMonitor
		647	+ monitor = DynamicProfilerMonitor()
		648	+ monitor._shm_obj = None
		649	+ result = monitor.shm_to_prof_conf_context()
		650	+ self.assertIsNone(result)
		651	+
563		652
564	if __name__ == "__main__":	653	if __name__ == "__main__":
565	run_tests()	654	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()

Atest/profiler/test_non_intrusive_profile.py+42-0

@@ -0,0 +1,42 @@
	1	+import os
	2	+import sys
	3	+import torch
	4	+
	5	+from torch_npu.utils._path_manager import PathManager
	6	+from torch_npu.profiler._dynamic_profiler._dynamic_profiler_utils import DynamicProfilerUtils
	7	+from torch_npu.profiler.dynamic_profile import init as dp_init
	8	+from torch_npu.profiler.dynamic_profile import step as dp_step
	9	+from torch_npu.profiler.analysis.prof_common_func._constant import print_error_msg, print_warn_msg
	10	+import torch_npu.profiler._non_intrusive_profile as none_intrusive_profile
	11	+from torch_npu.testing.testcase import TestCase, run_tests
	12	+from torch_npu.profiler._non_intrusive_profile import _NonIntrusiveProfile
	13	+
	14	+
	15	+class TestNoneInstrusiveProfile(TestCase):
	16	+ def test_step_wrapper(self):
	17	+ optimizer = torch.optim.SGD([torch.tensor([1.0])], lr=0.01)
	18	+
	19	+ def mock_step_func(args, *kwargs):
	20	+ return "wrapped_result"
	21	+
	22	+ wrapped_func = none_intrusive_profile._NonIntrusiveProfile.step_wrapper(mock_step_func)
	23	+ result = wrapped_func(optimizer)
	24	+ self.assertEqual(result, "wrapped_result")
	25	+
	26	+ def test_check_last_optimizer(self):
	27	+ optimizer1 = torch.optim.SGD([torch.tensor([1.0])], lr=0.01)
	28	+ optimizer2 = torch.optim.Adam([torch.tensor([1.0])], lr=0.01)
	29	+
	30	+ _NonIntrusiveProfile.OPTIMIZER_ID = id(optimizer1)
	31	+ self.assertTrue(_NonIntrusiveProfile.check_last_optimizer(optimizer1))
	32	+ self.assertFalse(_NonIntrusiveProfile.check_last_optimizer(optimizer2))
	33	+
	34	+ def test_patch_step_function(self):
	35	+ optimizer = torch.optim.SGD([torch.tensor([1.0])], lr=0.01)
	36	+ none_intrusive_profile._NonIntrusiveProfile.patch_step_function(optimizer)
	37	+ self.assertTrue(hasattr(optimizer.__class__.step, 'step_hooked'))
	38	+
	39	+
	40	+if __name__ == "__main__":
	41	+ run_tests()
	42	+