Merged revisions 85299 via svnmerge from

svn+ssh://pythondev@svn.python.org/python/branches/py3k

........
  r85299 | brian.curtin | 2010-10-06 20:12:19 -0500 (Wed, 06 Oct 2010) | 9 lines

  Fix #8584. Fix the skipping of a number of tests due to incorrect importing.

  This undoes a previous attempt to fix some of the skips. Value and copy
  come from sharedctypes rather than plain ctypes, and the test skipping is
  then done at the class level rather than test-by-test.

  There are zero skipped tests after the fix.
........
This commit is contained in:
Brian Curtin 2010-10-07 02:27:41 +00:00
parent 76a66aab19
commit a06e9b8f6b
1 changed files with 12 additions and 18 deletions

View File

@ -34,6 +34,12 @@ import multiprocessing.pool
from multiprocessing import util from multiprocessing import util
try:
from multiprocessing.sharedctypes import Value, copy
HAS_SHAREDCTYPES = True
except ImportError:
HAS_SHAREDCTYPES = False
# #
# #
# #
@ -72,16 +78,6 @@ except ImportError:
Structure = object Structure = object
c_int = c_double = None 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 # Creates a wrapper for a function which records the time it takes to finish
# #
@ -800,6 +796,8 @@ class _TestEvent(BaseTestCase):
# #
# #
@unittest.skipUnless(HAS_SHAREDCTYPES,
"requires multiprocessing.sharedctypes")
class _TestValue(BaseTestCase): class _TestValue(BaseTestCase):
ALLOWED_TYPES = ('processes',) ALLOWED_TYPES = ('processes',)
@ -816,7 +814,6 @@ class _TestValue(BaseTestCase):
sv.value = cv[2] sv.value = cv[2]
@unittest.skipIf(c_int is None, "requires _ctypes")
def test_value(self, raw=False): def test_value(self, raw=False):
if raw: if raw:
values = [self.RawValue(code, value) values = [self.RawValue(code, value)
@ -835,11 +832,9 @@ class _TestValue(BaseTestCase):
for sv, cv in zip(values, self.codes_values): for sv, cv in zip(values, self.codes_values):
self.assertEqual(sv.value, cv[2]) self.assertEqual(sv.value, cv[2])
@unittest.skipIf(c_int is None, "requires _ctypes")
def test_rawvalue(self): def test_rawvalue(self):
self.test_value(raw=True) self.test_value(raw=True)
@unittest.skipIf(c_int is None, "requires _ctypes")
def test_getobj_getlock(self): def test_getobj_getlock(self):
val1 = self.Value('i', 5) val1 = self.Value('i', 5)
lock1 = val1.get_lock() lock1 = val1.get_lock()
@ -1593,6 +1588,8 @@ class _Foo(Structure):
('y', c_double) ('y', c_double)
] ]
@unittest.skipUnless(HAS_SHAREDCTYPES,
"requires multiprocessing.sharedctypes")
class _TestSharedCTypes(BaseTestCase): class _TestSharedCTypes(BaseTestCase):
ALLOWED_TYPES = ('processes',) ALLOWED_TYPES = ('processes',)
@ -1606,14 +1603,13 @@ class _TestSharedCTypes(BaseTestCase):
for i in range(len(arr)): for i in range(len(arr)):
arr[i] *= 2 arr[i] *= 2
@unittest.skipIf(Value is None, "requires ctypes.Value")
def test_sharedctypes(self, lock=False): def test_sharedctypes(self, lock=False):
x = Value('i', 7, lock=lock) x = Value('i', 7, lock=lock)
y = Value(c_double, 1.0/3.0, lock=lock) y = Value(c_double, 1.0/3.0, lock=lock)
foo = Value(_Foo, 3, 2, lock=lock) foo = Value(_Foo, 3, 2, lock=lock)
arr = self.Array('d', range(10), lock=lock) arr = self.Array('d', range(10), lock=lock)
string = self.Array('c', 20, 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 = self.Process(target=self._double, args=(x, y, foo, arr, string))
p.start() p.start()
@ -1627,14 +1623,12 @@ class _TestSharedCTypes(BaseTestCase):
self.assertAlmostEqual(arr[i], i*2) self.assertAlmostEqual(arr[i], i*2)
self.assertEqual(string.value, latin('hellohello')) self.assertEqual(string.value, latin('hellohello'))
@unittest.skipIf(Value is None, "requires ctypes.Value")
def test_synchronize(self): def test_synchronize(self):
self.test_sharedctypes(lock=True) self.test_sharedctypes(lock=True)
@unittest.skipIf(ctypes_copy is None, "requires ctypes.copy")
def test_copy(self): def test_copy(self):
foo = _Foo(2, 5.0) foo = _Foo(2, 5.0)
bar = ctypes_copy(foo) bar = copy(foo)
foo.x = 0 foo.x = 0
foo.y = 0 foo.y = 0
self.assertEqual(bar.x, 2) self.assertEqual(bar.x, 2)