已合并
[test]add ut for native APIs #35009
mamba_ni创建于 5月7日
[test]add ut for native APIs #35009
已合并
共 17 个文件变更+774-0
| @@ -0,0 +1,70 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestAmpAutocast(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + self.device = torch.device(self.device_name) | ||
| 27 | + | ||
| 28 | + def tearDown(self): | ||
| 29 | + # Ensure autocast is disabled after each test | ||
| 30 | + torch.set_autocast_enabled(self.device_name, False) | ||
| 31 | + super().tearDown() | ||
| 32 | + | ||
| 33 | + def test_npu_autocast_context_manager(self): | ||
| 34 | + """Verify autocast context manager enters and exits without error.""" | ||
| 35 | + with torch.amp.autocast(self.device_name): | ||
| 36 | + pass | ||
| 37 | + | ||
| 38 | + def test_npu_autocast_enabled_inside(self): | ||
| 39 | + """Verify autocast is enabled inside context.""" | ||
| 40 | + with torch.amp.autocast(self.device_name): | ||
| 41 | + self.assertTrue(torch.is_autocast_enabled(self.device_name)) | ||
| 42 | + | ||
| 43 | + def test_npu_autocast_disabled_outside(self): | ||
| 44 | + """Verify autocast is disabled after exiting context.""" | ||
| 45 | + with torch.amp.autocast(self.device_name): | ||
| 46 | + pass | ||
| 47 | + self.assertFalse(torch.is_autocast_enabled(self.device_name)) | ||
| 48 | + | ||
| 49 | + def test_npu_autocast_disabled_context(self): | ||
| 50 | + """Verify autocast with enabled=False keeps it disabled inside.""" | ||
| 51 | + with torch.amp.autocast(self.device_name, enabled=False): | ||
| 52 | + self.assertFalse(torch.is_autocast_enabled(self.device_name)) | ||
| 53 | + | ||
| 54 | + def test_npu_autocast_float16(self): | ||
| 55 | + """Verify autocast with float16 dtype produces valid tensor.""" | ||
| 56 | + with torch.amp.autocast(self.device_name, dtype=torch.float16): | ||
| 57 | + x = torch.randn(4, 4, device=self.device) | ||
| 58 | + result = x @ x.T | ||
| 59 | + self.assertIsInstance(result, torch.Tensor) | ||
| 60 | + | ||
| 61 | + def test_npu_autocast_bfloat16(self): | ||
| 62 | + """Verify autocast with bfloat16 dtype produces valid tensor.""" | ||
| 63 | + with torch.amp.autocast(self.device_name, dtype=torch.bfloat16): | ||
| 64 | + x = torch.randn(4, 4, device=self.device) | ||
| 65 | + result = x @ x.T | ||
| 66 | + self.assertIsInstance(result, torch.Tensor) | ||
| 67 | + | ||
| 68 | + | ||
| 69 | +if __name__ == "__main__": | ||
| 70 | + run_tests() | ||
| @@ -0,0 +1,35 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCpuStreamFn(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_cpu_stream_context_manager(self): | ||
| 28 | + """Verify torch.cpu.stream works as a context manager.""" | ||
| 29 | + s = torch.cpu.Stream() | ||
| 30 | + with torch.cpu.stream(s): | ||
| 31 | + pass | ||
| 32 | + | ||
| 33 | + | ||
| 34 | +if __name__ == "__main__": | ||
| 35 | + run_tests() | ||
| @@ -0,0 +1,71 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaEvent(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_create_event(self): | ||
| 28 | + """Verify Event instance can be created with default args.""" | ||
| 29 | + e = torch.npu.Event() | ||
| 30 | + self.assertIsInstance(e, torch.npu.Event) | ||
| 31 | + | ||
| 32 | + def test_npu_event_query_returns_bool(self): | ||
| 33 | + """Verify query() returns a boolean value.""" | ||
| 34 | + e = torch.npu.Event() | ||
| 35 | + result = e.query() | ||
| 36 | + self.assertIsInstance(result, bool) | ||
| 37 | + | ||
| 38 | + def test_npu_event_wait(self): | ||
| 39 | + """Verify wait() completes without error.""" | ||
| 40 | + e = torch.npu.Event() | ||
| 41 | + e.wait() | ||
| 42 | + | ||
| 43 | + def test_npu_event_record(self): | ||
| 44 | + """Verify record() on a stream completes without error.""" | ||
| 45 | + e = torch.npu.Event() | ||
| 46 | + s = torch.npu.default_stream() | ||
| 47 | + e.record(s) | ||
| 48 | + | ||
| 49 | + def test_npu_event_enable_timing(self): | ||
| 50 | + """Verify Event with enable_timing=True can be created.""" | ||
| 51 | + e = torch.npu.Event(enable_timing=True) | ||
| 52 | + self.assertIsInstance(e, torch.npu.Event) | ||
| 53 | + | ||
| 54 | + def test_npu_event_blocking(self): | ||
| 55 | + """Verify Event with blocking=True can be created.""" | ||
| 56 | + e = torch.npu.Event(blocking=True) | ||
| 57 | + self.assertIsInstance(e, torch.npu.Event) | ||
| 58 | + | ||
| 59 | + def test_npu_event_query_after_record(self): | ||
| 60 | + """Verify query() returns True after record + synchronize.""" | ||
| 61 | + e = torch.npu.Event() | ||
| 62 | + s = torch.npu.default_stream() | ||
| 63 | + e.record(s) | ||
| 64 | + s.synchronize() | ||
| 65 | + result = e.query() | ||
| 66 | + # Event should be marked as completed after stream synchronization | ||
| 67 | + self.assertTrue(result) | ||
| 68 | + | ||
| 69 | + | ||
| 70 | +if __name__ == "__main__": | ||
| 71 | + run_tests() | ||
| @@ -0,0 +1,51 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaCurrentDevice(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + self.device = torch.device(self.device_name) | ||
| 27 | + self._orig_device = torch.npu.current_device() | ||
| 28 | + | ||
| 29 | + def tearDown(self): | ||
| 30 | + torch.npu.set_device(self._orig_device) | ||
| 31 | + super().tearDown() | ||
| 32 | + | ||
| 33 | + def test_npu_returns_int(self): | ||
| 34 | + """Verify current_device returns an int.""" | ||
| 35 | + result = torch.npu.current_device() | ||
| 36 | + self.assertIsInstance(result, int) | ||
| 37 | + | ||
| 38 | + def test_npu_returns_valid_index(self): | ||
| 39 | + """Verify current_device returns a valid device index.""" | ||
| 40 | + result = torch.npu.current_device() | ||
| 41 | + self.assertGreaterEqual(result, 0) | ||
| 42 | + self.assertLess(result, torch.npu.device_count()) | ||
| 43 | + | ||
| 44 | + def test_npu_after_set_device(self): | ||
| 45 | + """Verify current_device matches after set_device.""" | ||
| 46 | + torch.npu.set_device(0) | ||
| 47 | + self.assertEqual(torch.npu.current_device(), 0) | ||
| 48 | + | ||
| 49 | + | ||
| 50 | +if __name__ == "__main__": | ||
| 51 | + run_tests() | ||
| @@ -0,0 +1,39 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaDefaultStream(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_stream(self): | ||
| 28 | + """Verify default_stream returns a Stream instance.""" | ||
| 29 | + s = torch.npu.default_stream() | ||
| 30 | + self.assertIsInstance(s, torch.npu.Stream) | ||
| 31 | + | ||
| 32 | + def test_npu_with_device_arg(self): | ||
| 33 | + """Verify default_stream accepts a device index argument.""" | ||
| 34 | + s = torch.npu.default_stream(0) | ||
| 35 | + self.assertIsInstance(s, torch.npu.Stream) | ||
| 36 | + | ||
| 37 | + | ||
| 38 | +if __name__ == "__main__": | ||
| 39 | + run_tests() | ||
| @@ -0,0 +1,49 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaDevice(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_device_object(self): | ||
| 28 | + """Verify torch.npu.device returns a device instance.""" | ||
| 29 | + result = torch.npu.device(0) | ||
| 30 | + self.assertIsInstance(result, torch.npu.device) | ||
| 31 | + | ||
| 32 | + def test_npu_as_context_manager(self): | ||
| 33 | + """Verify torch.npu.device works as a context manager for device switching.""" | ||
| 34 | + with torch.npu.device(0): | ||
| 35 | + pass | ||
| 36 | + | ||
| 37 | + def test_npu_with_int_arg(self): | ||
| 38 | + """Verify device accepts an integer device index.""" | ||
| 39 | + d = torch.npu.device(0) | ||
| 40 | + self.assertIsNotNone(d) | ||
| 41 | + | ||
| 42 | + def test_npu_with_torch_device_arg(self): | ||
| 43 | + """Verify device accepts a torch.device object.""" | ||
| 44 | + d = torch.npu.device(torch.device("npu", 0)) | ||
| 45 | + self.assertIsNotNone(d) | ||
| 46 | + | ||
| 47 | + | ||
| 48 | +if __name__ == "__main__": | ||
| 49 | + run_tests() | ||
| @@ -0,0 +1,39 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaDeviceCount(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_int(self): | ||
| 28 | + """Verify device_count returns an int.""" | ||
| 29 | + result = torch.npu.device_count() | ||
| 30 | + self.assertIsInstance(result, int) | ||
| 31 | + | ||
| 32 | + def test_npu_positive_count(self): | ||
| 33 | + """Verify device_count is positive on NPU system.""" | ||
| 34 | + result = torch.npu.device_count() | ||
| 35 | + self.assertGreater(result, 0) | ||
| 36 | + | ||
| 37 | + | ||
| 38 | +if __name__ == "__main__": | ||
| 39 | + run_tests() | ||
| @@ -0,0 +1,41 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaDeviceOf(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_device_of_object(self): | ||
| 28 | + """Verify device_of returns a device_of instance for an NPU tensor.""" | ||
| 29 | + t = torch.tensor([1.0, 2.0], device=self.device_name) | ||
| 30 | + result = torch.npu.device_of(t) | ||
| 31 | + self.assertIsInstance(result, torch.npu.device_of) | ||
| 32 | + | ||
| 33 | + def test_npu_as_context_manager(self): | ||
| 34 | + """Verify device_of works as a context manager for device switching.""" | ||
| 35 | + t = torch.tensor([1.0, 2.0], device=self.device_name) | ||
| 36 | + with torch.npu.device_of(t): | ||
| 37 | + pass | ||
| 38 | + | ||
| 39 | + | ||
| 40 | +if __name__ == "__main__": | ||
| 41 | + run_tests() | ||
| @@ -0,0 +1,40 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaEmptyCache(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_none(self): | ||
| 28 | + """Verify empty_cache returns None.""" | ||
| 29 | + result = torch.cuda.empty_cache() | ||
| 30 | + self.assertIsNone(result) | ||
| 31 | + | ||
| 32 | + def test_npu_multiple_calls(self): | ||
| 33 | + """Verify consecutive empty_cache calls do not raise.""" | ||
| 34 | + torch.cuda.empty_cache() | ||
| 35 | + torch.cuda.empty_cache() | ||
| 36 | + torch.cuda.empty_cache() | ||
| 37 | + | ||
| 38 | + | ||
| 39 | +if __name__ == "__main__": | ||
| 40 | + run_tests() | ||
| @@ -0,0 +1,53 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaGetRngStateAll(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_list(self): | ||
| 28 | + """Verify get_rng_state_all returns a list.""" | ||
| 29 | + result = torch.cuda.get_rng_state_all() | ||
| 30 | + self.assertIsInstance(result, list) | ||
| 31 | + | ||
| 32 | + def test_npu_list_length_matches_device_count(self): | ||
| 33 | + """Verify returned list length equals device count.""" | ||
| 34 | + result = torch.cuda.get_rng_state_all() | ||
| 35 | + self.assertEqual(len(result), torch.cuda.device_count()) | ||
| 36 | + | ||
| 37 | + def test_npu_elements_are_tensors(self): | ||
| 38 | + """Verify each element in the returned list is a Tensor.""" | ||
| 39 | + result = torch.cuda.get_rng_state_all() | ||
| 40 | + for state in result: | ||
| 41 | + self.assertIsInstance(state, torch.Tensor) | ||
| 42 | + | ||
| 43 | + def test_npu_state_changes_after_seed(self): | ||
| 44 | + """Verify RNG state list length is consistent after reseeding.""" | ||
| 45 | + torch.cuda.manual_seed_all(42) | ||
| 46 | + state1 = torch.cuda.get_rng_state_all() | ||
| 47 | + torch.cuda.manual_seed_all(123) | ||
| 48 | + state2 = torch.cuda.get_rng_state_all() | ||
| 49 | + self.assertEqual(len(state1), len(state2)) | ||
| 50 | + | ||
| 51 | + | ||
| 52 | +if __name__ == "__main__": | ||
| 53 | + run_tests() | ||
| @@ -0,0 +1,41 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaGraphPoolHandle(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_tuple(self): | ||
| 28 | + """Verify graph_pool_handle returns a tuple.""" | ||
| 29 | + result = torch.npu.graph_pool_handle() | ||
| 30 | + self.assertIsInstance(result, tuple) | ||
| 31 | + | ||
| 32 | + def test_npu_multiple_calls(self): | ||
| 33 | + """Verify repeated calls return valid tuples without leaking state.""" | ||
| 34 | + h1 = torch.npu.graph_pool_handle() | ||
| 35 | + h2 = torch.npu.graph_pool_handle() | ||
| 36 | + self.assertIsInstance(h1, tuple) | ||
| 37 | + self.assertIsInstance(h2, tuple) | ||
| 38 | + | ||
| 39 | + | ||
| 40 | +if __name__ == "__main__": | ||
| 41 | + run_tests() | ||
| @@ -0,0 +1,39 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaIsCurrentStreamCapturing(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_bool(self): | ||
| 28 | + """Verify is_current_stream_capturing returns a boolean.""" | ||
| 29 | + result = torch.npu.is_current_stream_capturing() | ||
| 30 | + self.assertIsInstance(result, bool) | ||
| 31 | + | ||
| 32 | + def test_npu_not_capturing_by_default(self): | ||
| 33 | + """Verify stream is not in capture state by default.""" | ||
| 34 | + result = torch.npu.is_current_stream_capturing() | ||
| 35 | + self.assertFalse(result) | ||
| 36 | + | ||
| 37 | + | ||
| 38 | +if __name__ == "__main__": | ||
| 39 | + run_tests() | ||
| @@ -0,0 +1,39 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaIsInitialized(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_bool(self): | ||
| 28 | + """Verify is_initialized returns a boolean.""" | ||
| 29 | + result = torch.npu.is_initialized() | ||
| 30 | + self.assertIsInstance(result, bool) | ||
| 31 | + | ||
| 32 | + def test_npu_initialized_after_tensor_creation(self): | ||
| 33 | + """Verify is_initialized returns bool after NPU tensor creation.""" | ||
| 34 | + _ = torch.tensor(1.0, device=self.device_name) | ||
| 35 | + self.assertIsInstance(torch.npu.is_initialized(), bool) | ||
| 36 | + | ||
| 37 | + | ||
| 38 | +if __name__ == "__main__": | ||
| 39 | + run_tests() | ||
| @@ -0,0 +1,33 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaMakeGraphedCallables(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_make_graphed_callables_callable(self): | ||
| 28 | + """Verify make_graphed_callables is a callable in torch.npu namespace.""" | ||
| 29 | + self.assertTrue(callable(torch.npu.make_graphed_callables)) | ||
| 30 | + | ||
| 31 | + | ||
| 32 | +if __name__ == "__main__": | ||
| 33 | + run_tests() | ||
| @@ -0,0 +1,47 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaManualSeed(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_none(self): | ||
| 28 | + """Verify manual_seed returns None on success.""" | ||
| 29 | + result = torch.cuda.manual_seed(42) | ||
| 30 | + self.assertIsNone(result) | ||
| 31 | + | ||
| 32 | + def test_npu_seed_zero(self): | ||
| 33 | + """Verify seed=0 (boundary value) does not raise.""" | ||
| 34 | + torch.cuda.manual_seed(0) | ||
| 35 | + | ||
| 36 | + def test_npu_large_seed(self): | ||
| 37 | + """Verify large seed value (2**31-1) does not raise.""" | ||
| 38 | + torch.cuda.manual_seed(2**31 - 1) | ||
| 39 | + | ||
| 40 | + def test_npu_repeated_seed(self): | ||
| 41 | + """Verify setting the same seed twice does not raise.""" | ||
| 42 | + torch.cuda.manual_seed(123) | ||
| 43 | + torch.cuda.manual_seed(123) | ||
| 44 | + | ||
| 45 | + | ||
| 46 | +if __name__ == "__main__": | ||
| 47 | + run_tests() | ||
| @@ -0,0 +1,42 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaManualSeedAll(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_none(self): | ||
| 28 | + """Verify manual_seed_all returns None on success.""" | ||
| 29 | + result = torch.cuda.manual_seed_all(42) | ||
| 30 | + self.assertIsNone(result) | ||
| 31 | + | ||
| 32 | + def test_npu_seed_zero(self): | ||
| 33 | + """Verify seed=0 (boundary value) does not raise.""" | ||
| 34 | + torch.cuda.manual_seed_all(0) | ||
| 35 | + | ||
| 36 | + def test_npu_large_seed(self): | ||
| 37 | + """Verify large seed value (2**31-1) does not raise.""" | ||
| 38 | + torch.cuda.manual_seed_all(2**31 - 1) | ||
| 39 | + | ||
| 40 | + | ||
| 41 | +if __name__ == "__main__": | ||
| 42 | + run_tests() | ||
| @@ -0,0 +1,45 @@ | |||
| 1 | +# Owner(s): ["module: tests"] | ||
| 2 | + | ||
| 3 | +import torch_npu # noqa: F401 | ||
| 4 | + | ||
| 5 | +import torch | ||
| 6 | + | ||
| 7 | + | ||
| 8 | +try: | ||
| 9 | + from torch_npu.testing.testcase import run_tests, TestCase | ||
| 10 | +except ImportError: | ||
| 11 | + import sys | ||
| 12 | + import unittest | ||
| 13 | + from unittest import TestCase | ||
| 14 | + | ||
| 15 | + def run_tests(): | ||
| 16 | + unittest.main(argv=sys.argv) | ||
| 17 | + | ||
| 18 | + | ||
| 19 | +class TestCudaMaxMemoryAllocated(TestCase): | ||
| 20 | + def setUp(self): | ||
| 21 | + super().setUp() | ||
| 22 | + self.device_name = torch._C._get_privateuse1_backend_name() | ||
| 23 | + self.assertEqual( | ||
| 24 | + self.device_name, "npu", f"Expected device 'npu', got '{self.device_name}'" | ||
| 25 | + ) | ||
| 26 | + | ||
| 27 | + def test_npu_returns_int(self): | ||
| 28 | + """Verify max_memory_allocated returns an int.""" | ||
| 29 | + result = torch.cuda.max_memory_allocated() | ||
| 30 | + self.assertIsInstance(result, int) | ||
| 31 | + | ||
| 32 | + def test_npu_non_negative(self): | ||
| 33 | + """Verify reported value is non-negative.""" | ||
| 34 | + result = torch.cuda.max_memory_allocated() | ||
| 35 | + self.assertGreaterEqual(result, 0) | ||
| 36 | + | ||
| 37 | + def test_npu_gte_memory_allocated(self): | ||
| 38 | + """Verify peak memory >= current memory_allocated.""" | ||
| 39 | + max_val = torch.cuda.max_memory_allocated() | ||
| 40 | + cur_val = torch.cuda.memory_allocated() | ||
| 41 | + self.assertGreaterEqual(max_val, cur_val) | ||
| 42 | + | ||
| 43 | + | ||
| 44 | +if __name__ == "__main__": | ||
| 45 | + run_tests() | ||