[Python-checkins] cpython (merge 3.5 -> 3.6): Issue 28668: Merge 3.5
xavier.degaye
python-checkins at python.org
Tue Nov 15 11:10:25 EST 2016
https://hg.python.org/cpython/rev/e5404ba1b19e
changeset: 105124:e5404ba1b19e
branch: 3.6
parent: 105120:aa52ea2a7731
parent: 105123:a377e6987821
user: Xavier de Gaye <xdegaye at users.sourceforge.net>
date: Tue Nov 15 17:03:12 2016 +0100
summary:
Issue 28668: Merge 3.5
files:
Lib/test/support/__init__.py | 18 +++++++++++++++++-
Lib/test/test_logging.py | 2 ++
2 files changed, 19 insertions(+), 1 deletions(-)
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -90,7 +90,7 @@
"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_android_level",
+ "check__all__", "requires_android_level", "requires_multiprocessing_queue",
# sys
"is_jython", "is_android", "check_impl_detail", "unix_shell",
# network
@@ -1752,6 +1752,22 @@
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:
diff --git a/Lib/test/test_logging.py b/Lib/test/test_logging.py
--- a/Lib/test/test_logging.py
+++ b/Lib/test/test_logging.py
@@ -3124,6 +3124,7 @@
self.assertEqual(mock_handle.call_count, 5 * self.repeat,
'correct number of handled log messages')
+ @support.requires_multiprocessing_queue
@patch.object(logging.handlers.QueueListener, 'handle')
def test_handle_called_with_mp_queue(self, mock_handle):
for i in range(self.repeat):
@@ -3140,6 +3141,7 @@
except queue.Empty:
return []
+ @support.requires_multiprocessing_queue
def test_no_messages_in_queue_after_stop(self):
"""
Five messages are logged then the QueueListener is stopped. This
--
Repository URL: https://hg.python.org/cpython
More information about the Python-checkins
mailing list