Using appropriate checks in tests

Lib/test/test_os.py

 # As a test suite for the os module, this is woefully inadequate, but this
 # does add tests for a few functions which have been determined to be more
 # portable than they had been thought to be.
 
 import os
 import errno
 import unittest
 import warnings
 import sys
 import signal
 import subprocess
 import time
 import shutil
 from test import support
 import contextlib
 import mmap
 import platform
 import re
 import uuid
 import asyncore
 import asynchat
 import socket
 import itertools
 import stat
 import locale
 import codecs
+import decimal
+import fractions
+import pickle
 try:
     import threading
 except ImportError:
     threading = None
+try:
+    import resource
+except ImportError:
+    resource = None
+try:
+    import fcntl
+except ImportError:
+    fcntl = None
+
 from test.script_helper import assert_python_ok
 
 with warnings.catch_warnings():
     warnings.simplefilter("ignore", DeprecationWarning)
     os.stat_float_times(True)
 st = os.stat(__file__)
 stat_supports_subsecond = (
     # check if float and int timestamps are different
     (st.st_atime != st[7])
     or (st.st_mtime != st[8])
     or (st.st_ctime != st[9]))
 
 # Detect whether we're on a Linux system that uses the (now outdated
 # and unmaintained) linuxthreads threading library.  There's an issue
 # when combining linuxthreads with a failed execv call: see
 # http://bugs.python.org/issue4970.
 if hasattr(sys, 'thread_info') and sys.thread_info.version:
     USING_LINUXTHREADS = sys.thread_info.version.startswith("linuxthreads")
 else:
     USING_LINUXTHREADS = False
+
+# Issue #14110: Some tests fail on FreeBSD if the user is in the wheel group.
+HAVE_WHEEL_GROUP = sys.platform.startswith('freebsd') and os.getgid() == 0
 
 # Tests creating TESTFN
 class FileTests(unittest.TestCase):
     def setUp(self):
         if os.path.exists(support.TESTFN):
             os.unlink(support.TESTFN)
     tearDown = setUp
 
     def test_access(self):
         f = os.open(support.TESTFN, os.O_CREAT|os.O_RDWR)
@@ skipping 102 matching lines @@
         os.mkdir(support.TESTFN)
         self.fname = os.path.join(support.TESTFN, "f1")
         f = open(self.fname, 'wb')
         f.write(b"ABC")
         f.close()
 
     def tearDown(self):
         os.unlink(self.fname)
         os.rmdir(support.TESTFN)
 
+    @unittest.skipUnless(hasattr(os, 'stat'), 'test needs os.stat()')
     def check_stat_attributes(self, fname):
-        if not hasattr(os, "stat"):
-            return
-
         result = os.stat(fname)
 
         # Make sure direct access works
         self.assertEqual(result[stat.ST_SIZE], 3)
         self.assertEqual(result.st_size, 3)
 
         # Make sure all the attributes are there
         members = dir(result)
         for name in dir(stat):
             if name[:3] == 'ST_':
@@ skipping 56 matching lines @@
 
     def test_stat_attributes_bytes(self):
         try:
             fname = self.fname.encode(sys.getfilesystemencoding())
         except UnicodeEncodeError:
             self.skipTest("cannot encode %a for the filesystem" % self.fname)
         with warnings.catch_warnings():
             warnings.simplefilter("ignore", DeprecationWarning)
             self.check_stat_attributes(fname)
 
+    def test_stat_result_pickle(self):
+        result = os.stat(self.fname)
+        p = pickle.dumps(result)
+        self.assertIn(b'\x03cos\nstat_result\n', p)
+        unpickled = pickle.loads(p)
+        self.assertEqual(result, unpickled)
+
+    @unittest.skipUnless(hasattr(os, 'statvfs'), 'test needs os.statvfs()')
     def test_statvfs_attributes(self):
-        if not hasattr(os, "statvfs"):
-            return
-
         try:
             result = os.statvfs(self.fname)
         except OSError as e:
             # On AtheOS, glibc always returns ENOSYS
             if e.errno == errno.ENOSYS:
-                return
+                self.skipTest('os.statvfs() failed with ENOSYS')
 
         # Make sure direct access works
         self.assertEqual(result.f_bfree, result[3])
 
         # Make sure all the attributes are there.
         members = ('bsize', 'frsize', 'blocks', 'bfree', 'bavail', 'files',
                     'ffree', 'favail', 'flag', 'namemax')
         for value, member in enumerate(members):
             self.assertEqual(getattr(result, 'f_' + member), result[value])
 
@@ skipping 15 matching lines @@
             result2 = os.statvfs_result((10,))
             self.fail("No exception raised")
         except TypeError:
             pass
 
         # Use the constructor with a too-long tuple.
         try:
             result2 = os.statvfs_result((0,1,2,3,4,5,6,7,8,9,10,11,12,13,14))
         except TypeError:
             pass
+
+    @unittest.skipUnless(hasattr(os, 'statvfs'),
+                         "need os.statvfs()")
+    def test_statvfs_result_pickle(self):
+        try:
+            result = os.statvfs(self.fname)
+        except OSError as e:
+            # On AtheOS, glibc always returns ENOSYS
+            if e.errno == errno.ENOSYS:
+                self.skipTest('os.statvfs() failed with ENOSYS')
+
+        p = pickle.dumps(result)
+        self.assertIn(b'\x03cos\nstatvfs_result\n', p)
+        unpickled = pickle.loads(p)
+        self.assertEqual(result, unpickled)
 
     def test_utime_dir(self):
         delta = 1000000
         st = os.stat(support.TESTFN)
         # round to int, because some systems may support sub-second
         # time stamps in stat, but not in utime.
         os.utime(support.TESTFN, (st.st_atime, int(st.st_mtime-delta)))
         st2 = os.stat(support.TESTFN)
         self.assertEqual(st2.st_mtime, int(st.st_mtime-delta))
 
@@ skipping 127 matching lines @@
         def set_time(filename, atime, mtime):
             dirname = os.path.dirname(filename)
             dirfd = os.open(dirname, os.O_RDONLY)
             try:
                 os.utime(os.path.basename(filename), dir_fd=dirfd,
                              times=(atime, mtime))
             finally:
                 os.close(dirfd)
         self._test_utime_subsecond(set_time)
 
-    # Restrict test to Win32, since there is no guarantee other
+    # Restrict tests to Win32, since there is no guarantee other
     # systems support centiseconds
-    if sys.platform == 'win32':
-        def get_file_system(path):
+    def get_file_system(path):
+        if sys.platform == 'win32':
             root = os.path.splitdrive(os.path.abspath(path))[0] + '\\'
             import ctypes
             kernel32 = ctypes.windll.kernel32
             buf = ctypes.create_unicode_buffer("", 100)
             if kernel32.GetVolumeInformationW(root, None, 0, None, None, None, buf, len(buf)):
                 return buf.value
 
-        if get_file_system(support.TESTFN) == "NTFS":
-            def test_1565150(self):
-                t1 = 1159195039.25
-                os.utime(self.fname, (t1, t1))
-                self.assertEqual(os.stat(self.fname).st_mtime, t1)
-
-            def test_large_time(self):
-                t1 = 5000000000 # some day in 2128
-                os.utime(self.fname, (t1, t1))
-                self.assertEqual(os.stat(self.fname).st_mtime, t1)
-
-        def test_1686475(self):
-            # Verify that an open file can be stat'ed
-            try:
-                os.stat(r"c:\pagefile.sys")
-            except FileNotFoundError:
-                pass # file does not exist; cannot run test
-            except OSError as e:
-                self.fail("Could not stat pagefile.sys")
-
-        @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
-        def test_15261(self):
-            # Verify that stat'ing a closed fd does not cause crash
-            r, w = os.pipe()
-            try:
-                os.stat(r)          # should not raise error
-            finally:
-                os.close(r)
-                os.close(w)
-            with self.assertRaises(OSError) as ctx:
-                os.stat(r)
-            self.assertEqual(ctx.exception.errno, errno.EBADF)
+    @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+    @unittest.skipUnless(get_file_system(support.TESTFN) == "NTFS",
+                         "requires NTFS")
+    def test_1565150(self):
+        t1 = 1159195039.25
+        os.utime(self.fname, (t1, t1))
+        self.assertEqual(os.stat(self.fname).st_mtime, t1)
+
+    @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+    @unittest.skipUnless(get_file_system(support.TESTFN) == "NTFS",
+                         "requires NTFS")
+    def test_large_time(self):
+        t1 = 5000000000 # some day in 2128
+        os.utime(self.fname, (t1, t1))
+        self.assertEqual(os.stat(self.fname).st_mtime, t1)
+
+    @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+    def test_1686475(self):
+        # Verify that an open file can be stat'ed
+        try:
+            os.stat(r"c:\pagefile.sys")
+        except FileNotFoundError:
+            self.skipTest(r'c:\pagefile.sys does not exist')
+        except OSError as e:
+            self.fail("Could not stat pagefile.sys")
+
+    @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+    @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
+    def test_15261(self):
+        # Verify that stat'ing a closed fd does not cause crash
+        r, w = os.pipe()
+        try:
+            os.stat(r)          # should not raise error
+        finally:
+            os.close(r)
+            os.close(w)
+        with self.assertRaises(OSError) as ctx:
+            os.stat(r)
+        self.assertEqual(ctx.exception.errno, errno.EBADF)
 
 from test import mapping_tests
 
 class EnvironTests(mapping_tests.BasicTestMappingProtocol):
     """check that os.environ object conform to mapping protocol"""
     type2test = None
 
     def setUp(self):
         self.__save = dict(os.environ)
         if os.supports_bytes_environ:
@@ skipping 123 matching lines @@
     @support.requires_mac_ver(10, 6)
     def test_unset_error(self):
         if sys.platform == "win32":
             # an environment variable is limited to 32,767 characters
             key = 'x' * 50000
             self.assertRaises(ValueError, os.environ.__delitem__, key)
         else:
             # "=" is not allowed in a variable name
             key = 'key='
             self.assertRaises(OSError, os.environ.__delitem__, key)
+
+    def test_key_type(self):
+        missing = 'missingkey'
+        self.assertNotIn(missing, os.environ)
+
+        with self.assertRaises(KeyError) as cm:
+            os.environ[missing]
+        self.assertIs(cm.exception.args[0], missing)
+        self.assertTrue(cm.exception.__suppress_context__)
+
+        with self.assertRaises(KeyError) as cm:
+            del os.environ[missing]
+        self.assertIs(cm.exception.args[0], missing)
+        self.assertTrue(cm.exception.__suppress_context__)
+
 
 class WalkTests(unittest.TestCase):
     """Tests for os.walk()."""
 
     def setUp(self):
         import os
         from os.path import join
 
         # Build:
         #     TESTFN/
@@ skipping 23 matching lines @@
 
         # Create stuff.
         os.makedirs(sub11_path)
         os.makedirs(sub2_path)
         os.makedirs(t2_path)
         for path in tmp1_path, tmp2_path, tmp3_path, tmp4_path:
             f = open(path, "w")
             f.write("I'm " + path + " and proud of it.  Blame test_os.\n")
             f.close()
         if support.can_symlink():
-            if os.name == 'nt':
-                def symlink_to_dir(src, dest):
-                    os.symlink(src, dest, True)
-            else:
-                symlink_to_dir = os.symlink
-            symlink_to_dir(os.path.abspath(t2_path), link_path)
-            symlink_to_dir('broken', broken_link_path)
+            os.symlink(os.path.abspath(t2_path), link_path)
+            os.symlink('broken', broken_link_path, True)
             sub2_tree = (sub2_path, ["link"], ["broken_link", "tmp3"])
         else:
             sub2_tree = (sub2_path, [], ["tmp3"])
 
         # Walk top-down.
         all = list(os.walk(walk_path))
         self.assertEqual(len(all), 4)
         # We can't know which order SUB1 and SUB2 will appear in.
         # Not flipped:  TESTFN, SUB1, SUB11, SUB2
         #     flipped:  TESTFN, SUB2, SUB1, SUB11
@@ skipping 156 matching lines @@
     def test_exist_ok_existing_directory(self):
         path = os.path.join(support.TESTFN, 'dir1')
         mode = 0o777
         old_mask = os.umask(0o022)
         os.makedirs(path, mode)
         self.assertRaises(OSError, os.makedirs, path, mode)
         self.assertRaises(OSError, os.makedirs, path, mode, exist_ok=False)
         self.assertRaises(OSError, os.makedirs, path, 0o776, exist_ok=True)
         os.makedirs(path, mode=mode, exist_ok=True)
         os.umask(old_mask)
+
+    @unittest.skipUnless(hasattr(os, 'chown'), 'test needs os.chown')
+    def test_chown_uid_gid_arguments_must_be_index(self):
+        stat = os.stat(support.TESTFN)
+        uid = stat.st_uid
+        gid = stat.st_gid
+        for value in (-1.0, -1j, decimal.Decimal(-1), fractions.Fraction(-2, 2)):
+            self.assertRaises(TypeError, os.chown, support.TESTFN, value, gid)
+            self.assertRaises(TypeError, os.chown, support.TESTFN, uid, value)
+        self.assertIsNone(os.chown(support.TESTFN, uid, gid))
+        self.assertIsNone(os.chown(support.TESTFN, -1, -1))
 
     def test_exist_ok_s_isgid_directory(self):
         path = os.path.join(support.TESTFN, 'dir1')
         S_ISGID = stat.S_ISGID
         mode = 0o777
         old_mask = os.umask(0o022)
         try:
             existing_testfn_mode = stat.S_IMODE(
                     os.lstat(support.TESTFN).st_mode)
             try:
@@ skipping 113 matching lines @@
         out = assert_python_ok('-c', code)
         stdout = out[1]
         self.assertEqual(len(stdout), 16)
         return stdout
 
     def test_urandom_subprocess(self):
         data1 = self.get_urandom_subprocess(16)
         data2 = self.get_urandom_subprocess(16)
         self.assertNotEqual(data1, data2)
 
+    @unittest.skipUnless(resource, "test requires the resource module")
+    def test_urandom_failure(self):
+        # Check urandom() failing when it is not able to open /dev/random.
+        # We spawn a new process to make the test more robust (if getrlimit()
+        # failed to restore the file descriptor limit after this, the whole
+        # test suite would crash; this actually happened on the OS X Tiger
+        # buildbot).
+        code = """if 1:
+            import errno
+            import os
+            import resource
+
+            soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
+            resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
+            try:
+                os.urandom(16)
+            except OSError as e:
+                assert e.errno == errno.EMFILE, e.errno
+            else:
+                raise AssertionError("OSError not raised")
+            """
+        assert_python_ok('-c', code)
+
+
 @contextlib.contextmanager
 def _execvpe_mockup(defpath=None):
     """
     Stubs out execv and execve functions when used as context manager.
     Records exec calls. The mock execv and execve functions always raise an
     exception as they would normally never return.
     """
     # A list of tuples containing (function name, first arg, args)
     # of calls to execv or execve that have been made.
     calls = []
@@ skipping 78 matching lines @@
             self.assertEqual(len(calls), 1)
             self.assertSequenceEqual(calls[0],
                 ('execve', native_fullpath, (arguments, env_path)))
 
     def test_internal_execvpe_str(self):
         self._test_internal_execvpe(str)
         if os.name != "nt":
             self._test_internal_execvpe(bytes)
 
 
+@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
 class Win32ErrorTests(unittest.TestCase):
     def test_rename(self):
         self.assertRaises(OSError, os.rename, support.TESTFN, support.TESTFN+".bak")
 
     def test_remove(self):
         self.assertRaises(OSError, os.remove, support.TESTFN)
 
     def test_chdir(self):
         self.assertRaises(OSError, os.chdir, support.TESTFN)
 
@@ skipping 26 matching lines @@
 
     def check(self, f, *args):
         try:
             f(support.make_bad_fd(), *args)
         except OSError as e:
             self.assertEqual(e.errno, errno.EBADF)
         else:
             self.fail("%r didn't raise a OSError with a bad file descriptor"
                       % f)
 
+    @unittest.skipUnless(hasattr(os, 'isatty'), 'test needs os.isatty()')
     def test_isatty(self):
-        if hasattr(os, "isatty"):
-            self.assertEqual(os.isatty(support.make_bad_fd()), False)
-
+        self.assertEqual(os.isatty(support.make_bad_fd()), False)
+
+    @unittest.skipUnless(hasattr(os, 'closerange'), 'test needs os.closerange()')
     def test_closerange(self):
-        if hasattr(os, "closerange"):
-            fd = support.make_bad_fd()
-            # Make sure none of the descriptors we are about to close are
-            # currently valid (issue 6542).
-            for i in range(10):
-                try: os.fstat(fd+i)
-                except OSError:
-                    pass
-                else:
-                    break
-            if i < 2:
-                raise unittest.SkipTest(
-                    "Unable to acquire a range of invalid file descriptors")
-            self.assertEqual(os.closerange(fd, fd + i-1), None)
-
+        fd = support.make_bad_fd()
+        # Make sure none of the descriptors we are about to close are
+        # currently valid (issue 6542).
+        for i in range(10):
+            try: os.fstat(fd+i)
+            except OSError:
+                pass
+            else:
+                break
+        if i < 2:
+            raise unittest.SkipTest(
+                "Unable to acquire a range of invalid file descriptors")
+        self.assertEqual(os.closerange(fd, fd + i-1), None)
+
+    @unittest.skipUnless(hasattr(os, 'dup2'), 'test needs os.dup2()')
     def test_dup2(self):
-        if hasattr(os, "dup2"):
-            self.check(os.dup2, 20)
-
+        self.check(os.dup2, 20)
+
+    @unittest.skipUnless(hasattr(os, 'fchmod'), 'test needs os.fchmod()')
     def test_fchmod(self):
-        if hasattr(os, "fchmod"):
-            self.check(os.fchmod, 0)
-
+        self.check(os.fchmod, 0)
+
+    @unittest.skipUnless(hasattr(os, 'fchown'), 'test needs os.fchown()')
     def test_fchown(self):
-        if hasattr(os, "fchown"):
-            self.check(os.fchown, -1, -1)
-
+        self.check(os.fchown, -1, -1)
+
+    @unittest.skipUnless(hasattr(os, 'fpathconf'), 'test needs os.fpathconf()')
     def test_fpathconf(self):
-        if hasattr(os, "fpathconf"):
-            self.check(os.pathconf, "PC_NAME_MAX")
-            self.check(os.fpathconf, "PC_NAME_MAX")
-
+        self.check(os.pathconf, "PC_NAME_MAX")
+        self.check(os.fpathconf, "PC_NAME_MAX")
+
+    @unittest.skipUnless(hasattr(os, 'ftruncate'), 'test needs os.ftruncate()')
     def test_ftruncate(self):
-        if hasattr(os, "ftruncate"):
-            self.check(os.truncate, 0)
-            self.check(os.ftruncate, 0)
-
+        self.check(os.truncate, 0)
+        self.check(os.ftruncate, 0)
+
+    @unittest.skipUnless(hasattr(os, 'lseek'), 'test needs os.lseek()')
     def test_lseek(self):
-        if hasattr(os, "lseek"):
-            self.check(os.lseek, 0, 0)
-
+        self.check(os.lseek, 0, 0)
+
+    @unittest.skipUnless(hasattr(os, 'read'), 'test needs os.read()')
     def test_read(self):
-        if hasattr(os, "read"):
-            self.check(os.read, 1)
-
+        self.check(os.read, 1)
+
+    @unittest.skipUnless(hasattr(os, 'readv'), 'test needs os.readv()')
+    def test_readv(self):
+        buf = bytearray(10)
+        self.check(os.readv, [buf])
+
+    @unittest.skipUnless(hasattr(os, 'tcsetpgrp'), 'test needs os.tcsetpgrp()')
     def test_tcsetpgrpt(self):
-        if hasattr(os, "tcsetpgrp"):
-            self.check(os.tcsetpgrp, 0)
-
+        self.check(os.tcsetpgrp, 0)
+
+    @unittest.skipUnless(hasattr(os, 'write'), 'test needs os.write()')
     def test_write(self):
-        if hasattr(os, "write"):
-            self.check(os.write, b" ")
+        self.check(os.write, b" ")
+
+    @unittest.skipUnless(hasattr(os, 'writev'), 'test needs os.writev()')
+    def test_writev(self):
+        self.check(os.writev, [b'abc'])
 
 
 class LinkTests(unittest.TestCase):
     def setUp(self):
         self.file1 = support.TESTFN
         self.file2 = os.path.join(support.TESTFN + "2")
 
     def tearDown(self):
         for file in (self.file1, self.file2):
             if os.path.exists(file):
@@ skipping 19 matching lines @@
     def test_unicode_name(self):
         try:
             os.fsencode("\xf1")
         except UnicodeError:
             raise unittest.SkipTest("Unable to encode for this platform.")
 
         self.file1 += "\xf1"
         self.file2 = self.file1 + "2"
         self._test_link(self.file1, self.file2)
 
-if sys.platform != 'win32':
-    class Win32ErrorTests(unittest.TestCase):
-        pass
-
-    class PosixUidGidTests(unittest.TestCase):
-        if hasattr(os, 'setuid'):
-            def test_setuid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.setuid, 0)
-                self.assertRaises(OverflowError, os.setuid, 1<<32)
-
-        if hasattr(os, 'setgid'):
-            def test_setgid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.setgid, 0)
-                self.assertRaises(OverflowError, os.setgid, 1<<32)
-
-        if hasattr(os, 'seteuid'):
-            def test_seteuid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.seteuid, 0)
-                self.assertRaises(OverflowError, os.seteuid, 1<<32)
-
-        if hasattr(os, 'setegid'):
-            def test_setegid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.setegid, 0)
-                self.assertRaises(OverflowError, os.setegid, 1<<32)
-
-        if hasattr(os, 'setreuid'):
-            def test_setreuid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.setreuid, 0, 0)
-                self.assertRaises(OverflowError, os.setreuid, 1<<32, 0)
-                self.assertRaises(OverflowError, os.setreuid, 0, 1<<32)
-
-            def test_setreuid_neg1(self):
-                # Needs to accept -1.  We run this in a subprocess to avoid
-                # altering the test runner's process state (issue8045).
-                subprocess.check_call([
-                        sys.executable, '-c',
-                        'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
-
-        if hasattr(os, 'setregid'):
-            def test_setregid(self):
-                if os.getuid() != 0:
-                    self.assertRaises(OSError, os.setregid, 0, 0)
-                self.assertRaises(OverflowError, os.setregid, 1<<32, 0)
-                self.assertRaises(OverflowError, os.setregid, 0, 1<<32)
-
-            def test_setregid_neg1(self):
-                # Needs to accept -1.  We run this in a subprocess to avoid
-                # altering the test runner's process state (issue8045).
-                subprocess.check_call([
-                        sys.executable, '-c',
-                        'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
-
-    class Pep383Tests(unittest.TestCase):
-        def setUp(self):
-            if support.TESTFN_UNENCODABLE:
-                self.dir = support.TESTFN_UNENCODABLE
-            elif support.TESTFN_NONASCII:
-                self.dir = support.TESTFN_NONASCII
-            else:
-                self.dir = support.TESTFN
-            self.bdir = os.fsencode(self.dir)
-
-            bytesfn = []
-            def add_filename(fn):
-                try:
-                    fn = os.fsencode(fn)
-                except UnicodeEncodeError:
-                    return
-                bytesfn.append(fn)
-            add_filename(support.TESTFN_UNICODE)
-            if support.TESTFN_UNENCODABLE:
-                add_filename(support.TESTFN_UNENCODABLE)
-            if support.TESTFN_NONASCII:
-                add_filename(support.TESTFN_NONASCII)
-            if not bytesfn:
-                self.skipTest("couldn't create any non-ascii filename")
-
-            self.unicodefn = set()
-            os.mkdir(self.dir)
+@unittest.skipIf(sys.platform == "win32", "Posix specific tests")
+class PosixUidGidTests(unittest.TestCase):
+    @unittest.skipUnless(hasattr(os, 'setuid'), 'test needs os.setuid()')
+    def test_setuid(self):
+        if os.getuid() != 0:
+            self.assertRaises(OSError, os.setuid, 0)
+        self.assertRaises(OverflowError, os.setuid, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'setgid'), 'test needs os.setgid()')
+    def test_setgid(self):
+        if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
+            self.assertRaises(OSError, os.setgid, 0)
+        self.assertRaises(OverflowError, os.setgid, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'seteuid'), 'test needs os.seteuid()')
+    def test_seteuid(self):
+        if os.getuid() != 0:
+            self.assertRaises(OSError, os.seteuid, 0)
+        self.assertRaises(OverflowError, os.seteuid, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'setegid'), 'test needs os.setegid()')
+    def test_setegid(self):
+        if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
+            self.assertRaises(OSError, os.setegid, 0)
+        self.assertRaises(OverflowError, os.setegid, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'setreuid'), 'test needs os.setreuid()')
+    def test_setreuid(self):
+        if os.getuid() != 0:
+            self.assertRaises(OSError, os.setreuid, 0, 0)
+        self.assertRaises(OverflowError, os.setreuid, 1<<32, 0)
+        self.assertRaises(OverflowError, os.setreuid, 0, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'setreuid'), 'test needs os.setreuid()')
+    def test_setreuid_neg1(self):
+        # Needs to accept -1.  We run this in a subprocess to avoid
+        # altering the test runner's process state (issue8045).
+        subprocess.check_call([
+                sys.executable, '-c',
+                'import os,sys;os.setreuid(-1,-1);sys.exit(0)'])
+
+    @unittest.skipUnless(hasattr(os, 'setregid'), 'test needs os.setregid()')
+    def test_setregid(self):
+        if os.getuid() != 0 and not HAVE_WHEEL_GROUP:
+            self.assertRaises(OSError, os.setregid, 0, 0)
+        self.assertRaises(OverflowError, os.setregid, 1<<32, 0)
+        self.assertRaises(OverflowError, os.setregid, 0, 1<<32)
+
+    @unittest.skipUnless(hasattr(os, 'setregid'), 'test needs os.setregid()')
+    def test_setregid_neg1(self):
+        # Needs to accept -1.  We run this in a subprocess to avoid
+        # altering the test runner's process state (issue8045).
+        subprocess.check_call([
+                sys.executable, '-c',
+                'import os,sys;os.setregid(-1,-1);sys.exit(0)'])
+
+@unittest.skipIf(sys.platform == "win32", "Posix specific tests")
+class Pep383Tests(unittest.TestCase):
+    def setUp(self):
+        if support.TESTFN_UNENCODABLE:
+            self.dir = support.TESTFN_UNENCODABLE
+        elif support.TESTFN_NONASCII:
+            self.dir = support.TESTFN_NONASCII
+        else:
+            self.dir = support.TESTFN
+        self.bdir = os.fsencode(self.dir)
+
+        bytesfn = []
+        def add_filename(fn):
             try:
-                for fn in bytesfn:
-                    support.create_empty_file(os.path.join(self.bdir, fn))
-                    fn = os.fsdecode(fn)
-                    if fn in self.unicodefn:
-                        raise ValueError("duplicate filename")
-                    self.unicodefn.add(fn)
-            except:
-                shutil.rmtree(self.dir)
-                raise
-
-        def tearDown(self):
+                fn = os.fsencode(fn)
+            except UnicodeEncodeError:
+                return
+            bytesfn.append(fn)
+        add_filename(support.TESTFN_UNICODE)
+        if support.TESTFN_UNENCODABLE:
+            add_filename(support.TESTFN_UNENCODABLE)
+        if support.TESTFN_NONASCII:
+            add_filename(support.TESTFN_NONASCII)
+        if not bytesfn:
+            self.skipTest("couldn't create any non-ascii filename")
+
+        self.unicodefn = set()
+        os.mkdir(self.dir)
+        try:
+            for fn in bytesfn:
+                support.create_empty_file(os.path.join(self.bdir, fn))
+                fn = os.fsdecode(fn)
+                if fn in self.unicodefn:
+                    raise ValueError("duplicate filename")
+                self.unicodefn.add(fn)
+        except:
             shutil.rmtree(self.dir)
-
-        def test_listdir(self):
-            expected = self.unicodefn
-            found = set(os.listdir(self.dir))
-            self.assertEqual(found, expected)
-            # test listdir without arguments
-            current_directory = os.getcwd()
-            try:
-                os.chdir(os.sep)
-                self.assertEqual(set(os.listdir()), set(os.listdir(os.sep)))
-            finally:
-                os.chdir(current_directory)
-
-        def test_open(self):
-            for fn in self.unicodefn:
-                f = open(os.path.join(self.dir, fn), 'rb')
-                f.close()
-
-        @unittest.skipUnless(hasattr(os, 'statvfs'),
-                             "need os.statvfs()")
-        def test_statvfs(self):
-            # issue #9645
-            for fn in self.unicodefn:
-                # should not fail with file not found error
-                fullname = os.path.join(self.dir, fn)
-                os.statvfs(fullname)
-
-        def test_stat(self):
-            for fn in self.unicodefn:
-                os.stat(os.path.join(self.dir, fn))
-else:
-    class PosixUidGidTests(unittest.TestCase):
-        pass
-    class Pep383Tests(unittest.TestCase):
-        pass
+            raise
+
+    def tearDown(self):
+        shutil.rmtree(self.dir)
+
+    def test_listdir(self):
+        expected = self.unicodefn
+        found = set(os.listdir(self.dir))
+        self.assertEqual(found, expected)
+        # test listdir without arguments
+        current_directory = os.getcwd()
+        try:
+            os.chdir(os.sep)
+            self.assertEqual(set(os.listdir()), set(os.listdir(os.sep)))
+        finally:
+            os.chdir(current_directory)
+
+    def test_open(self):
+        for fn in self.unicodefn:
+            f = open(os.path.join(self.dir, fn), 'rb')
+            f.close()
+
+    @unittest.skipUnless(hasattr(os, 'statvfs'),
+                            "need os.statvfs()")
+    def test_statvfs(self):
+        # issue #9645
+        for fn in self.unicodefn:
+            # should not fail with file not found error
+            fullname = os.path.join(self.dir, fn)
+            os.statvfs(fullname)
+
+    def test_stat(self):
+        for fn in self.unicodefn:
+            os.stat(os.path.join(self.dir, fn))
 
 @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
 class Win32KillTests(unittest.TestCase):
     def _kill(self, sig):
         # Start sys.executable as a subprocess and communicate from the
         # subprocess to the parent that the interpreter is ready. When it
         # becomes ready, send *sig* via os.kill to the subprocess and check
         # that the return code is equal to *sig*.
         import ctypes
         from ctypes import wintypes
@@ skipping 98 matching lines @@
         # by subprocesses.
         SetConsoleCtrlHandler(NULL, 0)
 
         self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
 
     def test_CTRL_BREAK_EVENT(self):
         self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
 
 
 @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
+class Win32ListdirTests(unittest.TestCase):
+    """Test listdir on Windows."""
+
+    def setUp(self):
+        self.created_paths = []
+        for i in range(2):
+            dir_name = 'SUB%d' % i
+            dir_path = os.path.join(support.TESTFN, dir_name)
+            file_name = 'FILE%d' % i
+            file_path = os.path.join(support.TESTFN, file_name)
+            os.makedirs(dir_path)
+            with open(file_path, 'w') as f:
+                f.write("I'm %s and proud of it. Blame test_os.\n" % file_path)
+            self.created_paths.extend([dir_name, file_name])
+        self.created_paths.sort()
+
+    def tearDown(self):
+        shutil.rmtree(support.TESTFN)
+
+    def test_listdir_no_extended_path(self):
+        """Test when the path is not an "extended" path."""
+        # unicode
+        self.assertEqual(
+                sorted(os.listdir(support.TESTFN)),
+                self.created_paths)
+        # bytes
+        self.assertEqual(
+                sorted(os.listdir(os.fsencode(support.TESTFN))),
+                [os.fsencode(path) for path in self.created_paths])
+
+    def test_listdir_extended_path(self):
+        """Test when the path starts with '\\\\?\\'."""
+        # See: http://msdn.microsoft.com/en-us/library/windows/desktop/aa365247(v=vs.85).aspx#maxpath
+        # unicode
+        path = '\\\\?\\' + os.path.abspath(support.TESTFN)
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                self.created_paths)
+        # bytes
+        path = b'\\\\?\\' + os.fsencode(os.path.abspath(support.TESTFN))
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                [os.fsencode(path) for path in self.created_paths])
+
+
+@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
 @support.skip_unless_symlink
 class Win32SymlinkTests(unittest.TestCase):
     filelink = 'filelinktest'
     filelink_target = os.path.abspath(__file__)
     dirlink = 'dirlinktest'
     dirlink_target = os.path.dirname(filelink_target)
     missing_link = 'missing link'
 
     def setUp(self):
         assert os.path.exists(self.dirlink_target)
         assert os.path.exists(self.filelink_target)
         assert not os.path.exists(self.dirlink)
         assert not os.path.exists(self.filelink)
         assert not os.path.exists(self.missing_link)
 
     def tearDown(self):
         if os.path.exists(self.filelink):
             os.remove(self.filelink)
         if os.path.exists(self.dirlink):
             os.rmdir(self.dirlink)
         if os.path.lexists(self.missing_link):
             os.remove(self.missing_link)
 
     def test_directory_link(self):
-        os.symlink(self.dirlink_target, self.dirlink, True)
+        os.symlink(self.dirlink_target, self.dirlink)
         self.assertTrue(os.path.exists(self.dirlink))
         self.assertTrue(os.path.isdir(self.dirlink))
         self.assertTrue(os.path.islink(self.dirlink))
         self.check_stat(self.dirlink, self.dirlink_target)
 
     def test_file_link(self):
         os.symlink(self.filelink_target, self.filelink)
         self.assertTrue(os.path.exists(self.filelink))
         self.assertTrue(os.path.isfile(self.filelink))
         self.assertTrue(os.path.islink(self.filelink))
@@ skipping 71 matching lines @@
                 os.chdir(level3)
                 self.assertEqual(os.stat(file1),
                                  os.stat(os.path.relpath(link)))
             finally:
                 os.chdir(orig_dir)
         except OSError as err:
             self.fail(err)
         finally:
             os.remove(file1)
             shutil.rmtree(level1)
+
+
+@support.skip_unless_symlink
+class NonLocalSymlinkTests(unittest.TestCase):
+
+    def setUp(self):
+        """
+        Create this structure:
+
+        base
+         \___ some_dir
+        """
+        os.makedirs('base/some_dir')
+
+    def tearDown(self):
+        shutil.rmtree('base')
+
+    def test_directory_link_nonlocal(self):
+        """
+        The symlink target should resolve relative to the link, not relative
+        to the current directory.
+
+        Then, link base/some_link -> base/some_dir and ensure that some_link
+        is resolved as a directory.
+
+        In issue13772, it was discovered that directory detection failed if
+        the symlink target was not specified relative to the current
+        directory, which was a defect in the implementation.
+        """
+        src = os.path.join('base', 'some_link')
+        os.symlink('some_dir', src)
+        assert os.path.isdir(src)
 
 
 class FSEncodingTests(unittest.TestCase):
     def test_nop(self):
         self.assertEqual(os.fsencode(b'abc\xff'), b'abc\xff')
         self.assertEqual(os.fsdecode('abc\u0141'), 'abc\u0141')
 
     def test_identity(self):
         # assert fsdecode(fsencode(x)) == x
         for fn in ('unicode\u0141', 'latin\xe9', 'ascii'):
@@ skipping 151 matching lines @@
 
 
 @unittest.skipUnless(threading is not None, "test needs threading module")
 @unittest.skipUnless(hasattr(os, 'sendfile'), "test needs os.sendfile()")
 class TestSendfile(unittest.TestCase):
 
     DATA = b"12345abcde" * 16 * 1024  # 160 KB
     SUPPORT_HEADERS_TRAILERS = not sys.platform.startswith("linux") and \
                                not sys.platform.startswith("solaris") and \
                                not sys.platform.startswith("sunos")
+    requires_headers_trailers = unittest.skipUnless(SUPPORT_HEADERS_TRAILERS,
+            'requires headers and trailers support')
 
     @classmethod
     def setUpClass(cls):
         with open(support.TESTFN, "wb") as f:
             f.write(cls.DATA)
 
     @classmethod
     def tearDownClass(cls):
         support.unlink(support.TESTFN)
 
@@ skipping 98 matching lines @@
         data = self.server.handler_instance.get_data()
         self.assertEqual(data, b'')
 
     def test_invalid_offset(self):
         with self.assertRaises(OSError) as cm:
             os.sendfile(self.sockno, self.fileno, -1, 4096)
         self.assertEqual(cm.exception.errno, errno.EINVAL)
 
     # --- headers / trailers tests
 
-    if SUPPORT_HEADERS_TRAILERS:
-
-        def test_headers(self):
-            total_sent = 0
-            sent = os.sendfile(self.sockno, self.fileno, 0, 4096,
-                               headers=[b"x" * 512])
+    @requires_headers_trailers
+    def test_headers(self):
+        total_sent = 0
+        sent = os.sendfile(self.sockno, self.fileno, 0, 4096,
+                            headers=[b"x" * 512])
+        total_sent += sent
+        offset = 4096
+        nbytes = 4096
+        while 1:
+            sent = self.sendfile_wrapper(self.sockno, self.fileno,
+                                                    offset, nbytes)
+            if sent == 0:
+                break
             total_sent += sent
-            offset = 4096
-            nbytes = 4096
-            while 1:
-                sent = self.sendfile_wrapper(self.sockno, self.fileno,
-                                                     offset, nbytes)
-                if sent == 0:
-                    break
-                total_sent += sent
-                offset += sent
-
-            expected_data = b"x" * 512 + self.DATA
-            self.assertEqual(total_sent, len(expected_data))
+            offset += sent
+
+        expected_data = b"x" * 512 + self.DATA
+        self.assertEqual(total_sent, len(expected_data))
+        self.client.close()
+        self.server.wait()
+        data = self.server.handler_instance.get_data()
+        self.assertEqual(hash(data), hash(expected_data))
+
+    @requires_headers_trailers
+    def test_trailers(self):
+        TESTFN2 = support.TESTFN + "2"
+        file_data = b"abcdef"
+        with open(TESTFN2, 'wb') as f:
+            f.write(file_data)
+        with open(TESTFN2, 'rb')as f:
+            self.addCleanup(os.remove, TESTFN2)
+            os.sendfile(self.sockno, f.fileno(), 0, len(file_data),
+                        trailers=[b"1234"])
             self.client.close()
             self.server.wait()
             data = self.server.handler_instance.get_data()
-            self.assertEqual(hash(data), hash(expected_data))
-
-        def test_trailers(self):
-            TESTFN2 = support.TESTFN + "2"
-            with open(TESTFN2, 'wb') as f:
-                f.write(b"abcde")
-            with open(TESTFN2, 'rb')as f:
-                self.addCleanup(os.remove, TESTFN2)
-                os.sendfile(self.sockno, f.fileno(), 0, 4096,
-                            trailers=[b"12345"])
-                self.client.close()
-                self.server.wait()
-                data = self.server.handler_instance.get_data()
-                self.assertEqual(data, b"abcde12345")
-
-        if hasattr(os, "SF_NODISKIO"):
-            def test_flags(self):
-                try:
-                    os.sendfile(self.sockno, self.fileno, 0, 4096,
-                                flags=os.SF_NODISKIO)
-                except OSError as err:
-                    if err.errno not in (errno.EBUSY, errno.EAGAIN):
-                        raise
+            self.assertEqual(data, b"abcdef1234")
+
+    @requires_headers_trailers
+    @unittest.skipUnless(hasattr(os, 'SF_NODISKIO'),
+                         'test needs os.SF_NODISKIO')
+    def test_flags(self):
+        try:
+            os.sendfile(self.sockno, self.fileno, 0, 4096,
+                        flags=os.SF_NODISKIO)
+        except OSError as err:
+            if err.errno not in (errno.EBUSY, errno.EAGAIN):
+                raise
 
 
 def supports_extended_attributes():
     if not hasattr(os, "setxattr"):
         return False
     try:
         with open(support.TESTFN, "wb") as fp:
             try:
                 os.setxattr(fp.fileno(), b"user.test", b"")
             except OSError:
@@ skipping 254 matching lines @@
 
         for filenames, func, *func_args in funcs:
             for name in filenames:
                 try:
                     func(name, *func_args)
                 except OSError as err:
                     self.assertIs(err.filename, name)
                 else:
                     self.fail("No exception thrown by {}".format(func))
 
+class CPUCountTests(unittest.TestCase):
+    def test_cpu_count(self):
+        cpus = os.cpu_count()
+        if cpus is not None:
+            self.assertIsInstance(cpus, int)
+            self.assertGreater(cpus, 0)
+        else:
+            self.skipTest("Could not determine the number of CPUs")
+
+
+class FDInheritanceTests(unittest.TestCase):
+    def test_get_set_inheritable(self):
+        fd = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd)
+        self.assertEqual(os.get_inheritable(fd), False)
+
+        os.set_inheritable(fd, True)
+        self.assertEqual(os.get_inheritable(fd), True)
+
+    @unittest.skipIf(fcntl is None, "need fcntl")
+    def test_get_inheritable_cloexec(self):
+        fd = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd)
+        self.assertEqual(os.get_inheritable(fd), False)
+
+        # clear FD_CLOEXEC flag
+        flags = fcntl.fcntl(fd, fcntl.F_GETFD)
+        flags &= ~fcntl.FD_CLOEXEC
+        fcntl.fcntl(fd, fcntl.F_SETFD, flags)
+
+        self.assertEqual(os.get_inheritable(fd), True)
+
+    @unittest.skipIf(fcntl is None, "need fcntl")
+    def test_set_inheritable_cloexec(self):
+        fd = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd)
+        self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
+                         fcntl.FD_CLOEXEC)
+
+        os.set_inheritable(fd, True)
+        self.assertEqual(fcntl.fcntl(fd, fcntl.F_GETFD) & fcntl.FD_CLOEXEC,
+                         0)
+
+    def test_open(self):
+        fd = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd)
+        self.assertEqual(os.get_inheritable(fd), False)
+
+    @unittest.skipUnless(hasattr(os, 'pipe'), "need os.pipe()")
+    def test_pipe(self):
+        rfd, wfd = os.pipe()
+        self.addCleanup(os.close, rfd)
+        self.addCleanup(os.close, wfd)
+        self.assertEqual(os.get_inheritable(rfd), False)
+        self.assertEqual(os.get_inheritable(wfd), False)
+
+    def test_dup(self):
+        fd1 = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd1)
+
+        fd2 = os.dup(fd1)
+        self.addCleanup(os.close, fd2)
+        self.assertEqual(os.get_inheritable(fd2), False)
+
+    @unittest.skipUnless(hasattr(os, 'dup2'), "need os.dup2()")
+    def test_dup2(self):
+        fd = os.open(__file__, os.O_RDONLY)
+        self.addCleanup(os.close, fd)
+
+        # inheritable by default
+        fd2 = os.open(__file__, os.O_RDONLY)
+        try:
+            os.dup2(fd, fd2)
+            self.assertEqual(os.get_inheritable(fd2), True)
+        finally:
+            os.close(fd2)
+
+        # force non-inheritable
+        fd3 = os.open(__file__, os.O_RDONLY)
+        try:
+            os.dup2(fd, fd3, inheritable=False)
+            self.assertEqual(os.get_inheritable(fd3), False)
+        finally:
+            os.close(fd3)
+
+    @unittest.skipUnless(hasattr(os, 'openpty'), "need os.openpty()")
+    def test_openpty(self):
+        master_fd, slave_fd = os.openpty()
+        self.addCleanup(os.close, master_fd)
+        self.addCleanup(os.close, slave_fd)
+        self.assertEqual(os.get_inheritable(master_fd), False)
+        self.assertEqual(os.get_inheritable(slave_fd), False)
+
+
 @support.reap_threads
 def test_main():
     support.run_unittest(
         FileTests,
         StatAttributeTests,
         EnvironTests,
         WalkTests,
         FwalkTests,
         MakedirTests,
         DevNullTests,
         URandomTests,
         ExecTests,
         Win32ErrorTests,
         TestInvalidFD,
         PosixUidGidTests,
         Pep383Tests,
         Win32KillTests,
+        Win32ListdirTests,
         Win32SymlinkTests,
+        NonLocalSymlinkTests,
         FSEncodingTests,
         DeviceEncodingTests,
         PidTests,
         LoginTests,
         LinkTests,
         TestSendfile,
         ProgramPriorityTests,
         ExtendedAttributeTests,
         Win32DeprecatedBytesAPI,
         TermsizeTests,
         OSErrorTests,
         RemoveDirsTests,
+        CPUCountTests,
+        FDInheritanceTests,
     )
 
 if __name__ == "__main__":
     test_main()