diff options
Diffstat (limited to 'Lib/test/test_multiprocessing.py')
| -rw-r--r-- | Lib/test/test_multiprocessing.py | 30 |
1 files changed, 12 insertions, 18 deletions
diff --git a/Lib/test/test_multiprocessing.py b/Lib/test/test_multiprocessing.py index 73d12dc..d8e91df 100644 --- a/Lib/test/test_multiprocessing.py +++ b/Lib/test/test_multiprocessing.py @@ -36,6 +36,12 @@ import multiprocessing.pool from multiprocessing import util +try: + from multiprocessing.sharedctypes import Value, copy + HAS_SHAREDCTYPES = True +except ImportError: + HAS_SHAREDCTYPES = False + # # # @@ -75,16 +81,6 @@ except ImportError: Structure = object c_int = c_double = None -try: - from ctypes import Value -except ImportError: - Value = None - -try: - from ctypes import copy as ctypes_copy -except ImportError: - ctypes_copy = None - # # Creates a wrapper for a function which records the time it takes to finish # @@ -803,6 +799,8 @@ class _TestEvent(BaseTestCase): # # +@unittest.skipUnless(HAS_SHAREDCTYPES, + "requires multiprocessing.sharedctypes") class _TestValue(BaseTestCase): ALLOWED_TYPES = ('processes',) @@ -819,7 +817,6 @@ class _TestValue(BaseTestCase): sv.value = cv[2] - @unittest.skipIf(c_int is None, "requires _ctypes") def test_value(self, raw=False): if raw: values = [self.RawValue(code, value) @@ -838,11 +835,9 @@ class _TestValue(BaseTestCase): for sv, cv in zip(values, self.codes_values): self.assertEqual(sv.value, cv[2]) - @unittest.skipIf(c_int is None, "requires _ctypes") def test_rawvalue(self): self.test_value(raw=True) - @unittest.skipIf(c_int is None, "requires _ctypes") def test_getobj_getlock(self): val1 = self.Value('i', 5) lock1 = val1.get_lock() @@ -1594,6 +1589,8 @@ class _Foo(Structure): ('y', c_double) ] +@unittest.skipUnless(HAS_SHAREDCTYPES, + "requires multiprocessing.sharedctypes") class _TestSharedCTypes(BaseTestCase): ALLOWED_TYPES = ('processes',) @@ -1607,14 +1604,13 @@ class _TestSharedCTypes(BaseTestCase): for i in range(len(arr)): arr[i] *= 2 - @unittest.skipIf(Value is None, "requires ctypes.Value") def test_sharedctypes(self, lock=False): x = Value('i', 7, lock=lock) y = Value(c_double, 1.0/3.0, lock=lock) foo = Value(_Foo, 3, 2, lock=lock) arr = self.Array('d', list(range(10)), lock=lock) string = self.Array('c', 20, lock=lock) - string.value = 'hello' + string.value = latin('hello') p = self.Process(target=self._double, args=(x, y, foo, arr, string)) p.start() @@ -1628,14 +1624,12 @@ class _TestSharedCTypes(BaseTestCase): self.assertAlmostEqual(arr[i], i*2) self.assertEqual(string.value, latin('hellohello')) - @unittest.skipIf(Value is None, "requires ctypes.Value") def test_synchronize(self): self.test_sharedctypes(lock=True) - @unittest.skipIf(ctypes_copy is None, "requires ctypes.copy") def test_copy(self): foo = _Foo(2, 5.0) - bar = ctypes_copy(foo) + bar = copy(foo) foo.x = 0 foo.y = 0 self.assertEqual(bar.x, 2) |