Backport skipIf() and skipUnless() decorators of Python 2.7 unittest

7 files changed, 49 insertions, 27 deletions

diff --git a/asyncio/test_utils.py b/asyncio/test_utils.py
index d19868f..6a52ce9 100644
--- a/asyncio/test_utils.py
+++ b/asyncio/test_utils.py
@@ -2,6 +2,7 @@
 
 import collections
 import contextlib
+import functools
 import io
 import mock
 import os
@@ -274,3 +275,24 @@ class TestLoop(base_events.BaseEventLoop):
 
     def _write_to_self(self):
         pass
+
+
+try:
+    skipIf = unittest.skipIf
+    skipUnless = unittest.skipUnless
+except AttributeError:
+    # Python 2.6
+    def skip_wrapper(cond, message, func):
+        def wrapper(*args, **kw):
+            if cond:
+                return func(*args, **kw)
+            else:
+                print("Skip %s: %s" % (func, message))
+        return wrapper
+
+    def skipIf(cond, message):
+        return functools.partial(skip_wrapper, not cond, message)
+
+    def skipUnless(cond, message):
+        return functools.partial(skip_wrapper, cond, message)
+
diff --git a/tests/test_base_events.py b/tests/test_base_events.py
index cd87935..a160615 100644
--- a/tests/test_base_events.py
+++ b/tests/test_base_events.py
@@ -619,7 +619,7 @@ class BaseEventLoopWithSelectorTests(unittest.TestCase):
         self.assertRaises(
             OSError, self.loop.run_until_complete, coro)
 
-    @unittest.skipUnless(IPV6_ENABLED, 'IPv6 not supported or enabled')
+    @test_utils.skipUnless(IPV6_ENABLED, 'IPv6 not supported or enabled')
     def test_create_datagram_endpoint_no_matching_family(self):
         coro = self.loop.create_datagram_endpoint(
             protocols.DatagramProtocol,
diff --git a/tests/test_events.py b/tests/test_events.py
index 9d26430..2e83884 100644
--- a/tests/test_events.py
+++ b/tests/test_events.py
@@ -316,7 +316,7 @@ class EventLoopTestsMixin(object):
         self.loop.run_forever()
         self.assertEqual(results, ['hello', 'world'])
 
-    @unittest.skipIf(concurrent is None, 'need concurrent.futures')
+    @test_utils.skipIf(concurrent is None, 'need concurrent.futures')
     def test_run_in_executor(self):
         def run(arg):
             return (arg, thread.get_ident())
@@ -420,7 +420,7 @@ class EventLoopTestsMixin(object):
         conn.close()
         listener.close()
 
-    @unittest.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')
+    @test_utils.skipUnless(hasattr(signal, 'SIGKILL'), 'No SIGKILL')
     def test_add_signal_handler(self):
         non_local = {'caught': 0}
 
@@ -463,7 +463,7 @@ class EventLoopTestsMixin(object):
         # Removing again returns False.
         self.assertFalse(self.loop.remove_signal_handler(signal.SIGINT))
 
-    @unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
+    @test_utils.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
     def test_signal_handling_while_selecting(self):
         # Test with a signal actually arriving during a select() call.
         non_local = {'caught': 0}
@@ -478,7 +478,7 @@ class EventLoopTestsMixin(object):
         self.loop.run_forever()
         self.assertEqual(non_local['caught'], 1)
 
-    @unittest.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
+    @test_utils.skipUnless(hasattr(signal, 'SIGALRM'), 'No SIGALRM')
     def test_signal_handling_args(self):
         some_args = (42,)
         non_local = {'caught': 0}
@@ -533,7 +533,7 @@ class EventLoopTestsMixin(object):
             self.assertGreater(pr.nbytes, 0)
             tr.close()
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
     def test_create_ssl_connection(self):
         with test_utils.run_test_server(use_ssl=True) as httpd:
             f = self.loop.create_connection(
@@ -629,7 +629,7 @@ class EventLoopTestsMixin(object):
         self.assertEqual(host, '127.0.0.1')
         return server, host, port
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
     def test_create_server_ssl(self):
         non_local = {'proto': None}
 
@@ -675,8 +675,8 @@ class EventLoopTestsMixin(object):
         # stop serving
         server.close()
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
-    @unittest.skipUnless(HAS_SNI, 'No SNI support in ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipUnless(HAS_SNI, 'No SNI support in ssl module')
     def test_create_server_ssl_verify_failed(self):
         non_local = {'proto': None}
 
@@ -704,8 +704,8 @@ class EventLoopTestsMixin(object):
         self.assertIsNone(non_local['proto'].transport)
         server.close()
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
-    @unittest.skipUnless(HAS_SNI, 'No SNI support in ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipUnless(HAS_SNI, 'No SNI support in ssl module')
     def test_create_server_ssl_match_failed(self):
         non_local = {'proto': None}
 
@@ -736,8 +736,8 @@ class EventLoopTestsMixin(object):
         non_local['proto'].transport.close()
         server.close()
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
-    @unittest.skipUnless(HAS_SNI, 'No SNI support in ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipUnless(HAS_SNI, 'No SNI support in ssl module')
     def test_create_server_ssl_verified(self):
         non_local = {'proto': None}
 
@@ -808,7 +808,7 @@ class EventLoopTestsMixin(object):
 
         server.close()
 
-    @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
+    @test_utils.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled')
     def test_create_server_dual_stack(self):
         f_proto = futures.Future(loop=self.loop)
 
@@ -921,7 +921,7 @@ class EventLoopTestsMixin(object):
         self.assertIsNone(loop._csock)
         self.assertIsNone(loop._ssock)
 
-    @unittest.skipUnless(sys.platform != 'win32',
+    @test_utils.skipUnless(sys.platform != 'win32',
                          "Don't support pipes for Windows")
     def test_read_pipe(self):
         non_local = {'proto': None}
@@ -959,7 +959,7 @@ class EventLoopTestsMixin(object):
         # extra info is available
         self.assertIsNotNone(non_local['proto'].transport.get_extra_info('pipe'))
 
-    @unittest.skipUnless(sys.platform != 'win32',
+    @test_utils.skipUnless(sys.platform != 'win32',
                          "Don't support pipes for Windows")
     def test_write_pipe(self):
         non_local = {'proto': None, 'transport': None}
@@ -1002,7 +1002,7 @@ class EventLoopTestsMixin(object):
         self.loop.run_until_complete(non_local['proto'].done)
         self.assertEqual('CLOSED', non_local['proto'].state)
 
-    @unittest.skipUnless(sys.platform != 'win32',
+    @test_utils.skipUnless(sys.platform != 'win32',
                          "Don't support pipes for Windows")
     def test_write_pipe_disconnect_on_close(self):
         non_local = {'proto': None, 'transport': None}
@@ -1234,7 +1234,7 @@ class SubprocessTestsMixin(object):
         self.loop.run_until_complete(non_local['proto'].completed)
         self.check_terminated(non_local['proto'].returncode)
 
-    @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
+    @test_utils.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
     def test_subprocess_send_signal(self):
         non_local = {'proto': None, 'transp': None}
 
diff --git a/tests/test_futures.py b/tests/test_futures.py
index 91ae85b..c475958 100644
--- a/tests/test_futures.py
+++ b/tests/test_futures.py
@@ -218,7 +218,7 @@ class FutureTests(unittest.TestCase):
         del fut
         self.assertFalse(m_log.error.called)
 
-    @unittest.skipIf(concurrent is None, 'need concurrent.futures')
+    @test_utils.skipIf(concurrent is None, 'need concurrent.futures')
     def test_wrap_future(self):
 
         def run(arg):
@@ -236,7 +236,7 @@ class FutureTests(unittest.TestCase):
         f2 = futures.wrap_future(f1)
         self.assertIs(f1, f2)
 
-    @unittest.skipIf(concurrent is None, 'need concurrent.futures')
+    @test_utils.skipIf(concurrent is None, 'need concurrent.futures')
     @mock.patch('asyncio.futures.events')
     def test_wrap_future_use_global_loop(self, m_events):
         def run(arg):
@@ -246,7 +246,7 @@ class FutureTests(unittest.TestCase):
         f2 = futures.wrap_future(f1)
         self.assertIs(m_events.get_event_loop.return_value, f2._loop)
 
-    @unittest.skipIf(concurrent is None, 'need concurrent.futures')
+    @test_utils.skipIf(concurrent is None, 'need concurrent.futures')
     def test_wrap_future_cancel(self):
         f1 = concurrent.futures.Future()
         f2 = futures.wrap_future(f1, loop=self.loop)
@@ -255,7 +255,7 @@ class FutureTests(unittest.TestCase):
         self.assertTrue(f1.cancelled())
         self.assertTrue(f2.cancelled())
 
-    @unittest.skipIf(concurrent is None, 'need concurrent.futures')
+    @test_utils.skipIf(concurrent is None, 'need concurrent.futures')
     def test_wrap_future_cancel2(self):
         f1 = concurrent.futures.Future()
         f2 = futures.wrap_future(f1, loop=self.loop)
diff --git a/tests/test_selector_events.py b/tests/test_selector_events.py
index bf0a38c..13292b3 100644
--- a/tests/test_selector_events.py
+++ b/tests/test_selector_events.py
@@ -50,7 +50,7 @@ class BaseSelectorEventLoopTests(unittest.TestCase):
         self.assertIsInstance(
             self.loop._make_socket_transport(m, m), _SelectorSocketTransport)
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
     def test_make_ssl_transport(self):
         m = mock.Mock()
         self.loop.add_reader = mock.Mock()
@@ -1035,7 +1035,7 @@ class SelectorSocketTransportTests(unittest.TestCase):
         tr.close()
 
 
-@unittest.skipIf(ssl is None, 'No ssl module')
+@test_utils.skipIf(ssl is None, 'No ssl module')
 class SelectorSslTransportTests(unittest.TestCase):
 
     def setUp(self):
@@ -1346,7 +1346,7 @@ class SelectorSslTransportTests(unittest.TestCase):
         self.assertEqual(tr._conn_lost, 1)
         self.assertEqual(1, self.loop.remove_reader_count[1])
 
-    @unittest.skipIf(ssl is None or not HAS_SNI, 'No SNI support')
+    @test_utils.skipIf(ssl is None or not HAS_SNI, 'No SNI support')
     def test_server_hostname(self):
         _SelectorSslTransport(
             self.loop, self.sock, self.protocol, self.sslcontext,
diff --git a/tests/test_streams.py b/tests/test_streams.py
index 3a26b75..fd1373d 100644
--- a/tests/test_streams.py
+++ b/tests/test_streams.py
@@ -49,7 +49,7 @@ class StreamReaderTests(unittest.TestCase):
 
             writer.close()
 
-    @unittest.skipIf(ssl is None, 'No ssl module')
+    @test_utils.skipIf(ssl is None, 'No ssl module')
     def test_open_connection_no_loop_ssl(self):
         with test_utils.run_test_server(use_ssl=True) as httpd:
             try:
diff --git a/tests/test_unix_events.py b/tests/test_unix_events.py
index 71ed707..a03a7f4 100644
--- a/tests/test_unix_events.py
+++ b/tests/test_unix_events.py
@@ -24,7 +24,7 @@ from asyncio import test_utils
 from asyncio import unix_events
 
 
-@unittest.skipUnless(signal, 'Signals are not supported')
+@test_utils.skipUnless(signal, 'Signals are not supported')
 class SelectorEventLoopTests(unittest.TestCase):
 
     def setUp(self):