diff options
author | xdegaye <xdegaye@gmail.com> | 2017-12-01 07:08:49 (GMT) |
---|---|---|
committer | GitHub <noreply@github.com> | 2017-12-01 07:08:49 (GMT) |
commit | bf2b65e413ca5ec705c7e2f463d0d75e947588a4 (patch) | |
tree | f7146c74b318ca4d3ed66551e9f9e68b07be7fa7 /Lib/test/support | |
parent | a561862048555d555fa4850eaf832ae5474c7e1f (diff) | |
download | cpython-bf2b65e413ca5ec705c7e2f463d0d75e947588a4.zip cpython-bf2b65e413ca5ec705c7e2f463d0d75e947588a4.tar.gz cpython-bf2b65e413ca5ec705c7e2f463d0d75e947588a4.tar.bz2 |
bpo-28668: test.support.requires_multiprocessing_queue is removed (GH-4560)
Skip tests with test.support.import_module('multiprocessing.synchronize')
instead when the semaphore implementation is broken or missing.
Diffstat (limited to 'Lib/test/support')
-rw-r--r-- | Lib/test/support/__init__.py | 19 |
1 files changed, 1 insertions, 18 deletions
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py index f0e1507..22868d4 100644 --- a/Lib/test/support/__init__.py +++ b/Lib/test/support/__init__.py @@ -87,8 +87,7 @@ __all__ = [ "bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute", "requires_IEEE_754", "skip_unless_xattr", "requires_zlib", "anticipate_failure", "load_package_tests", "detect_api_mismatch", - "check__all__", "requires_multiprocessing_queue", - "skip_unless_bind_unix_socket", + "check__all__", "skip_unless_bind_unix_socket", # sys "is_jython", "is_android", "check_impl_detail", "unix_shell", "setswitchinterval", @@ -1791,22 +1790,6 @@ def impl_detail(msg=None, **guards): msg = msg.format(' or '.join(guardnames)) return unittest.skip(msg) -_have_mp_queue = None -def requires_multiprocessing_queue(test): - """Skip decorator for tests that use multiprocessing.Queue.""" - global _have_mp_queue - if _have_mp_queue is None: - import multiprocessing - # Without a functioning shared semaphore implementation attempts to - # instantiate a Queue will result in an ImportError (issue #3770). - try: - multiprocessing.Queue() - _have_mp_queue = True - except ImportError: - _have_mp_queue = False - msg = "requires a functioning shared semaphore implementation" - return test if _have_mp_queue else unittest.skip(msg)(test) - def _parse_guards(guards): # Returns a tuple ({platform_name: run_me}, default_value) if not guards: |