traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Issue #12981: test_multiprocessing: catch ImportError when importing 

multiprocessing.reduction, which may not be available (e.g. if the OS doesn't
support FD passing over Unix domain sockets).
This commit is contained in:
Charles-François Natali 2011-09-21 18:44:49 +02:00
parent c23484b21f
commit f8413b29c0
1 changed files with 13 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Lib/test/test_multiprocessing.py
View File

@ -33,7 +33,13 @@ import multiprocessing.managers
import multiprocessing.heap import multiprocessing.heap
import multiprocessing.pool import multiprocessing.pool
from multiprocessing import util, reduction from multiprocessing import util
try:
from multiprocessing import reduction
HAS_REDUCTION = True
except ImportError:
HAS_REDUCTION = False
try: try:
from multiprocessing.sharedctypes import Value, copy from multiprocessing.sharedctypes import Value, copy
@ -1526,6 +1532,7 @@ class _TestConnection(BaseTestCase):
os.write(fd, data) os.write(fd, data)
os.close(fd) os.close(fd)
@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
def test_fd_transfer(self): def test_fd_transfer(self):
if self.TYPE != 'processes': if self.TYPE != 'processes':
self.skipTest("only makes sense with processes") self.skipTest("only makes sense with processes")
@ -1543,6 +1550,7 @@ class _TestConnection(BaseTestCase):
with open(test_support.TESTFN, "rb") as f: with open(test_support.TESTFN, "rb") as f:
self.assertEqual(f.read(), b"foo") self.assertEqual(f.read(), b"foo")
@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
@unittest.skipIf(sys.platform == "win32", @unittest.skipIf(sys.platform == "win32",
"test semantics don't make sense on Windows") "test semantics don't make sense on Windows")
@unittest.skipIf(MAXFD <= 256, @unittest.skipIf(MAXFD <= 256,
@ -1578,6 +1586,7 @@ class _TestConnection(BaseTestCase):
def _send_data_without_fd(self, conn): def _send_data_without_fd(self, conn):
os.write(conn.fileno(), b"\0") os.write(conn.fileno(), b"\0")
@unittest.skipUnless(HAS_REDUCTION, "test needs multiprocessing.reduction")
@unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows") @unittest.skipIf(sys.platform == "win32", "doesn't make sense on Windows")
def test_missing_fd_transfer(self): def test_missing_fd_transfer(self):
# Check that exception is raised when received data is not # Check that exception is raised when received data is not
@ -1899,10 +1908,12 @@ class _TestImportStar(BaseTestCase):
'multiprocessing', 'multiprocessing.connection', 'multiprocessing', 'multiprocessing.connection',
'multiprocessing.heap', 'multiprocessing.managers', 'multiprocessing.heap', 'multiprocessing.managers',
'multiprocessing.pool', 'multiprocessing.process', 'multiprocessing.pool', 'multiprocessing.process',
'multiprocessing.reduction',
'multiprocessing.synchronize', 'multiprocessing.util' 'multiprocessing.synchronize', 'multiprocessing.util'
] ]
if HAS_REDUCTION:
modules.append('multiprocessing.reduction')
if c_int is not None: if c_int is not None:
# This module requires _ctypes # This module requires _ctypes
modules.append('multiprocessing.sharedctypes') modules.append('multiprocessing.sharedctypes')