Issue #12196: Make test.support's requires_linux_version a decorator.

Lib/test/support.py

@@ -37,8 +37,8 @@ __all__ = [
     "Error", "TestFailed", "ResourceDenied", "import_module",
     "verbose", "use_resources", "max_memuse", "record_original_stdout",
     "get_original_stdout", "unload", "unlink", "rmtree", "forget",
-    "is_resource_enabled", "requires", "linux_version", "requires_mac_ver",
+    "is_resource_enabled", "requires", "requires_linux_version",
-    "find_unused_port", "bind_port",
+    "requires_mac_ver", "find_unused_port", "bind_port",
     "IPV6_ENABLED", "is_jython", "TESTFN", "HOST", "SAVEDCWD", "temp_cwd",
     "findfile", "sortdict", "check_syntax_error", "open_urlresource",
     "check_warnings", "CleanImport", "EnvironmentVarGuard", "TransientResource",
@@ -292,13 +292,32 @@ def requires(resource, msg=None):
             msg = "Use of the `%s' resource not enabled" % resource
         raise ResourceDenied(msg)
 
-def linux_version():
+def requires_linux_version(*min_version):
-    try:
+    """Decorator raising SkipTest if the OS is Linux and the kernel version is
-        # platform.release() is something like '2.6.33.7-desktop-2mnb'
+    less than min_version.
-        version_string = platform.release().split('-')[0]
+
-        return tuple(map(int, version_string.split('.')))
+    For example, @requires_linux_version(2, 6, 35) raises SkipTest if the Linux
-    except ValueError:
+    kernel version is less than 2.6.35.
-        return 0, 0, 0
+    """
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kw):
+            if sys.platform.startswith('linux'):
+                version_txt = platform.release().split('-', 1)[0]
+                try:
+                    version = tuple(map(int, version_txt.split('.')))
+                except ValueError:
+                    pass
+                else:
+                    if version < min_version:
+                        min_version_txt = '.'.join(map(str, min_version))
+                        raise unittest.SkipTest(
+                            "Linux kernel %s or higher required, not %s"
+                            % (min_version_txt, version_txt))
+            return func(*args, **kw)
+        wrapper.min_version = min_version
+        return wrapper
+    return decorator
 
 def requires_mac_ver(*min_version):
     """Decorator raising SkipTest if the OS is Mac OS X and the OS X

Lib/test/test_posix.py

@@ -309,11 +309,8 @@ class PosixTester(unittest.TestCase):
                 fp2.close()
 
     @unittest.skipUnless(hasattr(os, 'O_CLOEXEC'), "needs os.O_CLOEXEC")
+    @support.requires_linux_version(2, 6, 23)
     def test_oscloexec(self):
-        version = support.linux_version()
-        if sys.platform == 'linux2' and version < (2, 6, 23):
-            self.skipTest("Linux kernel 2.6.23 or higher required, "
-                          "not %s.%s.%s" % version)
         fd = os.open(support.TESTFN, os.O_RDONLY|os.O_CLOEXEC)
         self.addCleanup(os.close, fd)
         self.assertTrue(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC)
@@ -479,11 +476,8 @@ class PosixTester(unittest.TestCase):
             os.close(writer)
 
     @unittest.skipUnless(hasattr(os, 'pipe2'), "test needs os.pipe2()")
+    @support.requires_linux_version(2, 6, 27)
     def test_pipe2(self):
-        version = support.linux_version()
-        if sys.platform == 'linux2' and version < (2, 6, 27):
-            self.skipTest("Linux kernel 2.6.27 or higher required, "
-                          "not %s.%s.%s" % version)
         self.assertRaises(TypeError, os.pipe2, 'DEADBEEF')
         self.assertRaises(TypeError, os.pipe2, 0, 0)
 

Lib/test/test_socket.py

@@ -1023,11 +1023,8 @@ class NonBlockingTCPTests(ThreadedTCPSocketTest):
         pass
 
     if hasattr(socket, "SOCK_NONBLOCK"):
+        @support.requires_linux_version(2, 6, 28)
         def testInitNonBlocking(self):
-            v = support.linux_version()
-            if v < (2, 6, 28):
-                self.skipTest("Linux kernel 2.6.28 or higher required, not %s"
-                              % ".".join(map(str, v)))
             # reinit server socket
             self.serv.close()
             self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM |
@@ -2001,11 +1998,8 @@ class ContextManagersTest(ThreadedTCPSocketTest):
                      "SOCK_CLOEXEC not defined")
 @unittest.skipUnless(fcntl, "module fcntl not available")
 class CloexecConstantTest(unittest.TestCase):
+    @support.requires_linux_version(2, 6, 28)
     def test_SOCK_CLOEXEC(self):
-        v = support.linux_version()
-        if v < (2, 6, 28):
-            self.skipTest("Linux kernel 2.6.28 or higher required, not %s"
-                          % ".".join(map(str, v)))
         with socket.socket(socket.AF_INET,
                            socket.SOCK_STREAM | socket.SOCK_CLOEXEC) as s:
             self.assertTrue(s.type & socket.SOCK_CLOEXEC)
@@ -2023,11 +2017,8 @@ class NonblockConstantTest(unittest.TestCase):
             self.assertFalse(s.type & socket.SOCK_NONBLOCK)
             self.assertEqual(s.gettimeout(), None)
 
+    @support.requires_linux_version(2, 6, 28)
     def test_SOCK_NONBLOCK(self):
-        v = support.linux_version()
-        if v < (2, 6, 28):
-            self.skipTest("Linux kernel 2.6.28 or higher required, not %s"
-                          % ".".join(map(str, v)))
         # a lot of it seems silly and redundant, but I wanted to test that
         # changing back and forth worked ok
         with socket.socket(socket.AF_INET,