diff --git a/Lib/os.py b/Lib/os.py
index 9bb00f45f01..a5e1d805556 100644
--- a/Lib/os.py
+++ b/Lib/os.py
@@ -58,6 +58,11 @@ def _get_exports_list(module):
         __all__.append('_exit')
     except ImportError:
         pass
+    try:
+        from posix import _clearenv
+        __all__.append('_clearenv')
+    except ImportError:
+        pass
     import posixpath as path
 
     try:
@@ -131,6 +136,8 @@ def _add(str, fn):
     _add("HAVE_UNLINKAT",   "unlink")
     _add("HAVE_UNLINKAT",   "rmdir")
     _add("HAVE_UTIMENSAT",  "utime")
+    if _exists("statx"):
+        _set.add(statx)
     supports_dir_fd = _set
 
     _set = set()
@@ -152,6 +159,8 @@ def _add(str, fn):
     _add("HAVE_FPATHCONF",  "pathconf")
     if _exists("statvfs") and _exists("fstatvfs"): # mac os x10.3
         _add("HAVE_FSTATVFS", "statvfs")
+    if _exists("statx"):
+        _set.add(statx)
     supports_fd = _set
 
     _set = set()
@@ -190,6 +199,8 @@ def _add(str, fn):
     _add("HAVE_FSTATAT",    "stat")
     _add("HAVE_UTIMENSAT",  "utime")
     _add("MS_WINDOWS",      "stat")
+    if _exists("statx"):
+        _set.add(statx)
     supports_follow_symlinks = _set
 
     del _set
@@ -208,14 +219,17 @@ def _add(str, fn):
 # Super directory utilities.
 # (Inspired by Eric Raymond; the doc strings are mostly his)
 
-def makedirs(name, mode=0o777, exist_ok=False):
-    """makedirs(name [, mode=0o777][, exist_ok=False])
+def makedirs(name, mode=0o777, exist_ok=False, *, parent_mode=None):
+    """makedirs(name [, mode=0o777][, exist_ok=False][, parent_mode=None])
 
     Super-mkdir; create a leaf directory and all intermediate ones.  Works
     like mkdir, except that any intermediate path segment (not just the
     rightmost) will be created if it does not exist.  If the target
     directory already exists, raise an OSError if exist_ok is False.
-    Otherwise no exception is raised.  This is recursive.
+    Otherwise no exception is raised.  If parent_mode is not None, it will
+    be used as the mode for any newly-created, intermediate-level
+    directories. Otherwise, intermediate directories are created with the
+    default permissions (respecting umask).  This is recursive.
 
     """
     head, tail = path.split(name)
@@ -223,7 +237,11 @@ def makedirs(name, mode=0o777, exist_ok=False):
         head, tail = path.split(head)
     if head and tail and not path.exists(head):
         try:
-            makedirs(head, exist_ok=exist_ok)
+            if parent_mode is not None:
+                makedirs(head, mode=parent_mode, exist_ok=exist_ok,
+                         parent_mode=parent_mode)
+            else:
+                makedirs(head, exist_ok=exist_ok)
         except FileExistsError:
             # Defeats race condition when another thread created the path
             pass
@@ -417,14 +435,16 @@ def walk(top, topdown=True, onerror=None, followlinks=False):
             # Yield before sub-directory traversal if going top down
             yield top, dirs, nondirs
             # Traverse into sub-directories
-            for dirname in reversed(dirs):
-                new_path = join(top, dirname)
-                # bpo-23605: os.path.islink() is used instead of caching
-                # entry.is_symlink() result during the loop on os.scandir() because
-                # the caller can replace the directory entry during the "yield"
-                # above.
-                if followlinks or not islink(new_path):
-                    stack.append(new_path)
+            if dirs:
+                prefix = join(top, top[:0])  # Add trailing slash
+                for dirname in reversed(dirs):
+                    new_path = prefix + dirname
+                    # bpo-23605: os.path.islink() is used instead of caching
+                    # entry.is_symlink() result during the loop on os.scandir() because
+                    # the caller can replace the directory entry during the "yield"
+                    # above.
+                    if followlinks or not islink(new_path):
+                        stack.append(new_path)
         else:
             # Yield after sub-directory traversal if going bottom up
             stack.append((top, dirs, nondirs))
@@ -766,6 +786,12 @@ def __ror__(self, other):
         new.update(self)
         return new
 
+    if _exists("_clearenv"):
+        def clear(self):
+            _clearenv()
+            self._data.clear()
+
+
 def _create_environ_mapping():
     if name == 'nt':
         # Where Env Var Names Must Be UPPERCASE
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index 504a0ffb401..7f02be1dbeb 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -22,6 +22,7 @@
 
 
 __all__ = [
+    "requires_root_user", "requires_non_root_user",
     # globals
     "PIPE_MAX_SIZE", "verbose", "max_memuse", "use_resources", "failfast",
     # exceptions
@@ -3336,6 +3337,11 @@ def control_characters_c0() -> list[str]:
     return [chr(c) for c in range(0x00, 0x20)] + ["\x7F"]
 
 
+_ROOT_IN_POSIX = hasattr(os, 'geteuid') and os.geteuid() == 0
+requires_root_user = unittest.skipUnless(_ROOT_IN_POSIX, "test needs root privilege")
+requires_non_root_user = unittest.skipIf(_ROOT_IN_POSIX, "test needs non-root account")
+
+
 STATUS_DLL_INIT_FAILED = 0xC0000142
 def skip_on_low_desktop_heap_memory_subprocess(returncode):
     if sys.platform not in ('win32', 'cygwin'):
diff --git a/Lib/test/test_os/__init__.py b/Lib/test/test_os/__init__.py
new file mode 100644
index 00000000000..bc502ef32d2
--- /dev/null
+++ b/Lib/test/test_os/__init__.py
@@ -0,0 +1,6 @@
+import os.path
+from test.support import load_package_tests
+
+
+def load_tests(*args):
+    return load_package_tests(os.path.dirname(__file__), *args)
diff --git a/Lib/test/test_os.py b/Lib/test/test_os/test_os.py
similarity index 88%
rename from Lib/test/test_os.py
rename to Lib/test/test_os/test_os.py
index c9cabd67355..2e3cec6e42b 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os/test_os.py
@@ -7,7 +7,6 @@
 import contextlib
 import decimal
 import errno
-import fnmatch
 import fractions
 import itertools
 import locale
@@ -31,12 +30,14 @@
 import uuid
 import warnings
 from test import support
-from test.support import import_helper
 from test.support import os_helper
 from test.support import socket_helper
 from test.support import infinite_recursion
+from test.support import requires_root_user
+from test.support import requires_non_root_user
 from test.support import warnings_helper
 from platform import win32_is_iot
+from .utils import create_file
 
 try:
     import resource
@@ -46,10 +47,6 @@
     import fcntl
 except ImportError:
     fcntl = None
-try:
-    import _winapi
-except ImportError:
-    _winapi = None
 try:
     import pwd
     all_users = [u.pw_uid for u in pwd.getpwall()]
@@ -72,10 +69,6 @@
 from test.support.os_helper import FakePath
 
 
-root_in_posix = False
-if hasattr(os, 'geteuid'):
-    root_in_posix = (os.geteuid() == 0)
-
 # 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
@@ -93,11 +86,6 @@ def requires_os_func(name):
     return unittest.skipUnless(hasattr(os, name), 'requires os.%s' % name)
 
 
-def create_file(filename, content=b'content'):
-    with open(filename, "xb", 0) as fp:
-        fp.write(content)
-
-
 # On platforms without a native spawnv(), os.py provides a Python fallback
 # built on fork()+exec*() that reports argument conversion failures from the
 # child as exit status 127 instead of raising, so tests of the C
@@ -184,6 +172,20 @@ def test_getcwdb(self):
         self.assertIsInstance(cwd, bytes)
         self.assertEqual(os.fsdecode(cwd), os.getcwd())
 
+    def test_type_fqdn(self):
+        def fqdn(obj):
+            return (obj.__module__, obj.__qualname__)
+
+        native = os.name
+        self.assertEqual(fqdn(os.stat_result), ("os", "stat_result"))
+        self.assertEqual(fqdn(os.times_result), (native, "times_result"))
+        if hasattr(os, "statvfs_result"):
+            self.assertEqual(fqdn(os.statvfs_result), ("os", "statvfs_result"))
+        if hasattr(os, "sched_param"):
+            self.assertEqual(fqdn(os.sched_param), (native, "sched_param"))
+        if hasattr(os, "waitid_result"):
+            self.assertEqual(fqdn(os.waitid_result), (native, "waitid_result"))
+
 
 # Tests creating TESTFN
 class FileTests(unittest.TestCase):
@@ -650,6 +652,15 @@ def setUp(self):
         self.addCleanup(os_helper.unlink, self.fname)
         create_file(self.fname, b"ABC")
 
+    def check_timestamp_agreement(self, result, names):
+        # Make sure that the st_?time and st_?time_ns fields roughly agree
+        # (they should always agree up to around tens-of-microseconds)
+        for name in names:
+            with self.subTest(name=name):
+                floaty = int(getattr(result, name) * 100_000)
+                nanosecondy = getattr(result, name + "_ns") // 10_000
+                self.assertAlmostEqual(floaty, nanosecondy, delta=2, msg=name)
+
     def check_stat_attributes(self, fname):
         result = os.stat(fname)
 
@@ -670,21 +681,15 @@ def trunc(x): return x
                                   result[getattr(stat, name)])
                 self.assertIn(attr, members)
 
-        # Make sure that the st_?time and st_?time_ns fields roughly agree
-        # (they should always agree up to around tens-of-microseconds)
-        for name in 'st_atime st_mtime st_ctime'.split():
-            floaty = int(getattr(result, name) * 100000)
-            nanosecondy = getattr(result, name + "_ns") // 10000
-            self.assertAlmostEqual(floaty, nanosecondy, delta=2)
-
-        # Ensure both birthtime and birthtime_ns roughly agree, if present
+        time_attributes = ['st_atime', 'st_mtime', 'st_ctime']
         try:
-            floaty = int(result.st_birthtime * 100000)
-            nanosecondy = result.st_birthtime_ns // 10000
+            result.st_birthtime
+            result.st_birthtime_ns
         except AttributeError:
             pass
         else:
-            self.assertAlmostEqual(floaty, nanosecondy, delta=2)
+            time_attributes.append('st_birthtime')
+        self.check_timestamp_agreement(result, time_attributes)
 
         try:
             result[200]
@@ -745,6 +750,160 @@ def test_stat_result_pickle(self):
                 unpickled = pickle.loads(p)
                 self.assertEqual(result, unpickled)
 
+    def check_statx_attributes(self, filename):
+        maximal_mask = 0
+        for name in dir(os):
+            if name.startswith('STATX_'):
+                maximal_mask |= getattr(os, name)
+        result = os.statx(filename, maximal_mask)
+        stat_result = os.stat(filename)
+
+        time_attributes = ('stx_atime', 'stx_btime', 'stx_ctime', 'stx_mtime')
+        # gh-83714: stx_btime can be None on tmpfs even if STATX_BTIME mask
+        # is used
+        time_attributes = [name for name in time_attributes
+                           if getattr(result, name) is not None]
+        self.check_timestamp_agreement(result, time_attributes)
+
+        def getmask(name):
+            return getattr(os, name, 0)
+
+        requirements = (
+            ('stx_atime', os.STATX_ATIME),
+            ('stx_atime_ns', os.STATX_ATIME),
+            ('stx_atomic_write_segments_max', getmask('STATX_WRITE_ATOMIC')),
+            ('stx_atomic_write_unit_max', getmask('STATX_WRITE_ATOMIC')),
+            ('stx_atomic_write_unit_max_opt', getmask('STATX_WRITE_ATOMIC')),
+            ('stx_atomic_write_unit_min', getmask('STATX_WRITE_ATOMIC')),
+            ('stx_attributes', 0),
+            ('stx_attributes_mask', 0),
+            ('stx_blksize', 0),
+            ('stx_blocks', os.STATX_BLOCKS),
+            ('stx_btime', os.STATX_BTIME),
+            ('stx_btime_ns', os.STATX_BTIME),
+            ('stx_ctime', os.STATX_CTIME),
+            ('stx_ctime_ns', os.STATX_CTIME),
+            ('stx_dev', 0),
+            ('stx_dev_major', 0),
+            ('stx_dev_minor', 0),
+            ('stx_dio_mem_align', getmask('STATX_DIOALIGN')),
+            ('stx_dio_offset_align', getmask('STATX_DIOALIGN')),
+            ('stx_dio_read_offset_align', getmask('STATX_DIO_READ_ALIGN')),
+            ('stx_gid', os.STATX_GID),
+            ('stx_ino', os.STATX_INO),
+            ('stx_mask', 0),
+            ('stx_mnt_id', getmask('STATX_MNT_ID')),
+            ('stx_mode', os.STATX_TYPE | os.STATX_MODE),
+            ('stx_mtime', os.STATX_MTIME),
+            ('stx_mtime_ns', os.STATX_MTIME),
+            ('stx_nlink', os.STATX_NLINK),
+            ('stx_rdev', 0),
+            ('stx_rdev_major', 0),
+            ('stx_rdev_minor', 0),
+            ('stx_size', os.STATX_SIZE),
+            ('stx_subvol', getmask('STATX_SUBVOL')),
+            ('stx_uid', os.STATX_UID),
+        )
+        optional_members = {
+            'stx_atomic_write_segments_max',
+            'stx_atomic_write_unit_max',
+            'stx_atomic_write_unit_max_opt',
+            'stx_atomic_write_unit_min',
+            'stx_dio_mem_align',
+            'stx_dio_offset_align',
+            'stx_dio_read_offset_align',
+            'stx_mnt_id',
+            'stx_subvol',
+        }
+        float_type = {
+            'stx_atime',
+            'stx_btime',
+            'stx_ctime',
+            'stx_mtime',
+        }
+
+        members = set(name for name in dir(result)
+                      if name.startswith('stx_'))
+        tested = set(name for name, mask in requirements)
+        if members - tested:
+            raise ValueError(f"statx members not tested: {members - tested}")
+
+        for name, mask in requirements:
+            with self.subTest(name=name):
+                try:
+                    x = getattr(result, name)
+                except AttributeError:
+                    if name in optional_members:
+                        continue
+                    else:
+                        raise
+
+                if not(result.stx_mask & mask == mask):
+                    self.assertIsNone(x)
+                    continue
+
+                if name in float_type:
+                    self.assertIsInstance(x, float)
+                else:
+                    self.assertIsInstance(x, int)
+
+                # Compare with stat_result
+                try:
+                    b = getattr(stat_result, "st_" + name[4:])
+                except AttributeError:
+                    pass
+                else:
+                    self.assertEqual(type(x), type(b))
+                    if isinstance(x, float):
+                        self.assertAlmostEqual(x, b)
+                    else:
+                        self.assertEqual(x, b)
+
+        self.assertEqual(result.stx_rdev_major, os.major(result.stx_rdev))
+        self.assertEqual(result.stx_rdev_minor, os.minor(result.stx_rdev))
+        self.assertEqual(result.stx_dev_major, os.major(result.stx_dev))
+        self.assertEqual(result.stx_dev_minor, os.minor(result.stx_dev))
+
+        self.assertEqual(result.stx_attributes & result.stx_attributes_mask,
+                         result.stx_attributes)
+
+    @unittest.skipUnless(hasattr(os, 'statx'), 'test needs os.statx()')
+    def test_statx_attributes(self):
+        self.check_statx_attributes(self.fname)
+
+    @unittest.skipUnless(hasattr(os, 'statx'), 'test needs os.statx()')
+    def test_statx_attributes_bytes(self):
+        try:
+            fname = self.fname.encode(sys.getfilesystemencoding())
+        except UnicodeEncodeError:
+            self.skipTest("cannot encode %a for the filesystem" % self.fname)
+        self.check_statx_attributes(fname)
+
+    @unittest.skipUnless(hasattr(os, 'statx'), 'test needs os.statx()')
+    def test_statx_attributes_pathlike(self):
+        self.check_statx_attributes(FakePath(self.fname))
+
+    @unittest.skipUnless(hasattr(os, 'statx'), 'test needs os.statx()')
+    def test_statx_result(self):
+        result = os.statx(self.fname, os.STATX_BASIC_STATS)
+
+        # Check that attributes are read-only
+        members = [name for name in dir(result)
+                   if name.startswith('stx_')]
+        for name in members:
+            try:
+                setattr(result, name, 1)
+            except AttributeError:
+                pass
+            else:
+                self.fail("No exception raised")
+
+        # statx_result is not a tuple or tuple-like object.
+        with self.assertRaisesRegex(TypeError, 'not subscriptable'):
+            result[0]
+        with self.assertRaisesRegex(TypeError, 'cannot unpack'):
+            _, _ = result
+
     @unittest.skipUnless(hasattr(os, 'statvfs'), 'test needs os.statvfs()')
     def test_statvfs_attributes(self):
         result = os.statvfs(self.fname)
@@ -969,6 +1128,20 @@ def ns_to_sec(ns):
         # issue, os.utime() rounds towards minus infinity.
         return (ns * 1e-9) + 0.5e-9
 
+    @staticmethod
+    def ns_to_sec_decimal(ns):
+        # Convert a number of nanosecond (int) to a number of seconds (Decimal).
+        # Round towards infinity by adding 0.5 nanosecond to avoid rounding
+        # issue, os.utime() rounds towards minus infinity.
+        return decimal.Decimal('1e-9') * ns + decimal.Decimal('0.5e-9')
+
+    @staticmethod
+    def ns_to_sec_fraction(ns):
+        # Convert a number of nanosecond (int) to a number of seconds (Fraction).
+        # Round towards infinity by adding 0.5 nanosecond to avoid rounding
+        # issue, os.utime() rounds towards minus infinity.
+        return fractions.Fraction(ns, 10**9) + fractions.Fraction(1, 2*10**9)
+
     def test_utime_by_indexed(self):
         # pass times as floating-point seconds as the second indexed parameter
         def set_time(filename, ns):
@@ -989,6 +1162,24 @@ def set_time(filename, ns):
             os.utime(filename, times=(atime, mtime))
         self._test_utime(set_time)
 
+    def test_utime_decimal(self):
+        # pass times as Decimal seconds
+        def set_time(filename, ns):
+            atime_ns, mtime_ns = ns
+            atime = self.ns_to_sec_decimal(atime_ns)
+            mtime = self.ns_to_sec_decimal(mtime_ns)
+            os.utime(filename, (atime, mtime))
+        self._test_utime(set_time)
+
+    def test_utime_fraction(self):
+        # pass times as Fraction seconds
+        def set_time(filename, ns):
+            atime_ns, mtime_ns = ns
+            atime = self.ns_to_sec_fraction(atime_ns)
+            mtime = self.ns_to_sec_fraction(mtime_ns)
+            os.utime(filename, (atime, mtime))
+        self._test_utime(set_time)
+
     @unittest.skipUnless(os.utime in os.supports_follow_symlinks,
                          "follow_symlinks support for utime required "
                          "for this test.")
@@ -1085,9 +1276,15 @@ def test_large_time(self):
         if self.get_file_system(self.dirname) != "NTFS":
             self.skipTest("requires NTFS")
 
-        large = 5000000000   # some day in 2128
-        os.utime(self.fname, (large, large))
-        self.assertEqual(os.stat(self.fname).st_mtime, large)
+        times = (
+            5000000000,  # some day in 2128
+            # boundaries of the fast path cutoff in posixmodule.c:fill_time
+            -9223372037, -9223372036, 9223372035, 9223372036,
+        )
+        for large in times:
+            with self.subTest(large=large):
+                os.utime(self.fname, (large, large))
+                self.assertEqual(os.stat(self.fname).st_mtime, large)
 
     def test_utime_invalid_arguments(self):
         # seconds and nanoseconds parameters are mutually exclusive
@@ -1487,6 +1684,14 @@ def test_reload_environ(self):
             self.assertNotIn(b'test_env', os.environb)
             self.assertNotIn('test_env', os.environ)
 
+    def test_clearenv(self):
+        os.environ['REMOVEME'] = '1'
+        os.environ.clear()
+        self.assertEqual(os.environ, {})
+
+        self.assertRaises(TypeError, os.environ.clear, None)
+
+
 class WalkTests(unittest.TestCase):
     """Tests for os.walk()."""
     is_fwalk = False
@@ -1953,6 +2158,94 @@ def test_mode(self):
                 self.assertEqual(os.stat(path).st_mode & 0o777, 0o555)
                 self.assertEqual(os.stat(parent).st_mode & 0o777, 0o775)
 
+    @unittest.skipIf(
+        support.is_emscripten or support.is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_mode_with_parent_mode(self):
+        # Test the parent_mode parameter
+        parent = os.path.join(os_helper.TESTFN, 'dir1')
+        path = os.path.join(parent, 'dir2')
+        with os_helper.temp_umask(0o002):
+            # Specify mode for both leaf and parent directories
+            os.makedirs(path, 0o770, parent_mode=0o750)
+            self.assertTrue(os.path.exists(path))
+            self.assertTrue(os.path.isdir(path))
+            if os.name != 'nt':
+                # Leaf directory gets the mode parameter
+                self.assertEqual(os.stat(path).st_mode & 0o777, 0o770)
+                # Parent directory gets the parent_mode parameter
+                self.assertEqual(os.stat(parent).st_mode & 0o777, 0o750)
+
+    @unittest.skipIf(
+        support.is_emscripten or support.is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_parent_mode_deep_hierarchy(self):
+        # Test parent_mode with deep directory hierarchy
+        base = os.path.join(os_helper.TESTFN, 'dir1', 'dir2', 'dir3')
+        with os_helper.temp_umask(0o002):
+            os.makedirs(base, 0o755, parent_mode=0o700)
+            self.assertTrue(os.path.exists(base))
+            if os.name != 'nt':
+                # Check that all parent directories have parent_mode
+                level1 = os.path.join(os_helper.TESTFN, 'dir1')
+                level2 = os.path.join(level1, 'dir2')
+                self.assertEqual(os.stat(level1).st_mode & 0o777, 0o700)
+                self.assertEqual(os.stat(level2).st_mode & 0o777, 0o700)
+                # Leaf directory has the regular mode
+                self.assertEqual(os.stat(base).st_mode & 0o777, 0o755)
+
+    @unittest.skipIf(
+        support.is_emscripten or support.is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_parent_mode_same_as_mode(self):
+        # Test emulating Python 3.6 behavior by setting parent_mode=mode
+        parent = os.path.join(os_helper.TESTFN, 'dir1')
+        path = os.path.join(parent, 'dir2')
+        with os_helper.temp_umask(0o002):
+            os.makedirs(path, 0o705, parent_mode=0o705)
+            self.assertTrue(os.path.exists(path))
+            if os.name != 'nt':
+                # Both directories should have the same mode
+                self.assertEqual(os.stat(path).st_mode & 0o777, 0o705)
+                self.assertEqual(os.stat(parent).st_mode & 0o777, 0o705)
+
+    @unittest.skipIf(
+        support.is_emscripten or support.is_wasi,
+        "umask is not implemented on Emscripten/WASI."
+    )
+    @unittest.skipIf(
+        sys.platform == "android",
+        "Android filesystem may not honor requested permissions."
+    )
+    def test_parent_mode_combined_with_umask(self):
+        # parent_mode, like mode, is combined with the process umask; it does
+        # not bypass it.
+        parent = os.path.join(os_helper.TESTFN, 'dir1')
+        path = os.path.join(parent, 'dir2')
+        with os_helper.temp_umask(0o022):
+            os.makedirs(path, 0o777, parent_mode=0o777)
+            self.assertTrue(os.path.isdir(path))
+            if os.name != 'nt':
+                # 0o777 is masked down to 0o755 by the 0o022 umask, for both
+                # the leaf (mode) and the parent (parent_mode).
+                self.assertEqual(os.stat(path).st_mode & 0o777, 0o755)
+                self.assertEqual(os.stat(parent).st_mode & 0o777, 0o755)
+
     @unittest.skipIf(
         support.is_wasi,
         "WASI's umask is a stub."
@@ -2026,15 +2319,9 @@ def test_win32_mkdir_700(self):
         )
 
     def tearDown(self):
-        path = os.path.join(os_helper.TESTFN, 'dir1', 'dir2', 'dir3',
-                            'dir4', 'dir5', 'dir6')
-        # If the tests failed, the bottom-most directory ('../dir6')
-        # may not have been created, so we look for the outermost directory
-        # that exists.
-        while not os.path.exists(path) and path != os_helper.TESTFN:
-            path = os.path.dirname(path)
-
-        os.removedirs(path)
+        # Remove the whole tree regardless of which sub-directories a test
+        # created and regardless of their permission bits.
+        os_helper.rmtree(os_helper.TESTFN)
 
 
 @unittest.skipUnless(hasattr(os, "chown"), "requires os.chown()")
@@ -2071,8 +2358,8 @@ def test_chown_gid(self):
         gid = os.stat(os_helper.TESTFN).st_gid
         self.assertEqual(gid, gid_2)
 
-    @unittest.skipUnless(root_in_posix and len(all_users) > 1,
-                         "test needs root privilege and more than one user")
+    @requires_root_user
+    @unittest.skipUnless(len(all_users) > 1, "test needs more than one user")
     def test_chown_with_root(self):
         uid_1, uid_2 = all_users[:2]
         gid = os.stat(os_helper.TESTFN).st_gid
@@ -2083,8 +2370,8 @@ def test_chown_with_root(self):
         uid = os.stat(os_helper.TESTFN).st_uid
         self.assertEqual(uid, uid_2)
 
-    @unittest.skipUnless(not root_in_posix and len(all_users) > 1,
-                         "test needs non-root account and more than one user")
+    @requires_non_root_user
+    @unittest.skipUnless(len(all_users) > 1, "test needs and more than one user")
     def test_chown_without_permission(self):
         uid_1, uid_2 = all_users[:2]
         gid = os.stat(os_helper.TESTFN).st_gid
@@ -2489,42 +2776,6 @@ def test_execve_with_empty_path(self):
             self.fail('No OSError raised')
 
 
-@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
-class Win32ErrorTests(unittest.TestCase):
-    def setUp(self):
-        try:
-            os.stat(os_helper.TESTFN)
-        except FileNotFoundError:
-            exists = False
-        except OSError as exc:
-            exists = True
-            self.fail("file %s must not exist; os.stat failed with %s"
-                      % (os_helper.TESTFN, exc))
-        else:
-            self.fail("file %s must not exist" % os_helper.TESTFN)
-
-    def test_rename(self):
-        self.assertRaises(OSError, os.rename, os_helper.TESTFN, os_helper.TESTFN+".bak")
-
-    def test_remove(self):
-        self.assertRaises(OSError, os.remove, os_helper.TESTFN)
-
-    def test_chdir(self):
-        self.assertRaises(OSError, os.chdir, os_helper.TESTFN)
-
-    def test_mkdir(self):
-        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
-
-        with open(os_helper.TESTFN, "x") as f:
-            self.assertRaises(OSError, os.mkdir, os_helper.TESTFN)
-
-    def test_utime(self):
-        self.assertRaises(OSError, os.utime, os_helper.TESTFN, None)
-
-    def test_chmod(self):
-        self.assertRaises(OSError, os.chmod, os_helper.TESTFN, 0)
-
-
 @unittest.skipIf(support.is_wasi, "Cannot create invalid FD on WASI.")
 class TestInvalidFD(unittest.TestCase):
     singles = ["fchdir", "dup", "fstat", "fstatvfs", "tcgetpgrp", "ttyname"]
@@ -2635,10 +2886,62 @@ def test_fpathconf_bad_fd(self):
         'musl fpathconf ignores the file descriptor and returns a constant',
         )
     def test_pathconf_negative_fd_uses_fd_semantics(self):
+        if os.pathconf not in os.supports_fd:
+            self.skipTest('needs fpathconf()')
+
         with self.assertRaises(OSError) as ctx:
             os.pathconf(-1, 1)
         self.assertEqual(ctx.exception.errno, errno.EBADF)
 
+    @support.subTests("fd", [-1, -5])
+    def test_negative_fd_ebadf(self, fd):
+        tests = [(os.stat, fd)]
+        if hasattr(os, "statx"):
+            tests.append((os.statx, fd, 0))
+        if os.chdir in os.supports_fd:
+            tests.append((os.chdir, fd))
+        if os.chmod in os.supports_fd:
+            tests.append((os.chmod, fd, 0o777))
+        if hasattr(os, "chown") and os.chown in os.supports_fd:
+            tests.append((os.chown, fd, 0, 0))
+        if os.listdir in os.supports_fd:
+            tests.append((os.listdir, fd))
+        if os.utime in os.supports_fd:
+            tests.append((os.utime, fd, (0, 0)))
+        if hasattr(os, "truncate") and os.truncate in os.supports_fd:
+            tests.append((os.truncate, fd, 0))
+        if hasattr(os, 'statvfs') and os.statvfs in os.supports_fd:
+            tests.append((os.statvfs, fd))
+        if hasattr(os, "setxattr"):
+            tests.append((os.getxattr, fd, b"user.test"))
+            tests.append((os.setxattr, fd, b"user.test", b"1"))
+            tests.append((os.removexattr, fd, b"user.test"))
+            tests.append((os.listxattr, fd))
+        if os.scandir in os.supports_fd:
+            tests.append((os.scandir, fd))
+
+        for func, *args in tests:
+            with self.subTest(func=func, args=args):
+                with self.assertRaises(OSError) as ctx:
+                    func(*args)
+                self.assertEqual(ctx.exception.errno, errno.EBADF)
+
+        if (hasattr(os, "execve") and os.execve in os.supports_fd
+            and support.has_subprocess_support):
+            # glibc fails with EINVAL, musl fails with EBADF
+            with self.assertRaises(OSError) as ctx:
+                os.execve(fd, [sys.executable, "-c", "pass"], os.environ)
+            self.assertIn(ctx.exception.errno, (errno.EBADF, errno.EINVAL))
+
+        if support.MS_WINDOWS:
+            import nt
+            self.assertFalse(nt._path_exists(fd))
+            self.assertFalse(nt._path_lexists(fd))
+            self.assertFalse(nt._path_isdir(fd))
+            self.assertFalse(nt._path_isfile(fd))
+            self.assertFalse(nt._path_islink(fd))
+            self.assertFalse(nt._path_isjunction(fd))
+
     @unittest.skipUnless(hasattr(os, 'ftruncate'), 'test needs os.ftruncate()')
     def test_ftruncate(self):
         self.check(os.truncate, 0)
@@ -2869,224 +3172,6 @@ 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
-        import msvcrt
-
-        # Since we can't access the contents of the process' stdout until the
-        # process has exited, use PeekNamedPipe to see what's inside stdout
-        # without waiting. This is done so we can tell that the interpreter
-        # is started and running at a point where it could handle a signal.
-        PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
-        PeekNamedPipe.restype = wintypes.BOOL
-        PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
-                                  ctypes.POINTER(ctypes.c_char), # stdout buf
-                                  wintypes.DWORD, # Buffer size
-                                  ctypes.POINTER(wintypes.DWORD), # bytes read
-                                  ctypes.POINTER(wintypes.DWORD), # bytes avail
-                                  ctypes.POINTER(wintypes.DWORD)) # bytes left
-        msg = "running"
-        proc = subprocess.Popen([sys.executable, "-c",
-                                 "import sys;"
-                                 "sys.stdout.write('{}');"
-                                 "sys.stdout.flush();"
-                                 "input()".format(msg)],
-                                stdout=subprocess.PIPE,
-                                stderr=subprocess.PIPE,
-                                stdin=subprocess.PIPE)
-        self.addCleanup(proc.stdout.close)
-        self.addCleanup(proc.stderr.close)
-        self.addCleanup(proc.stdin.close)
-
-        count, max = 0, 100
-        while count < max and proc.poll() is None:
-            # Create a string buffer to store the result of stdout from the pipe
-            buf = ctypes.create_string_buffer(len(msg))
-            # Obtain the text currently in proc.stdout
-            # Bytes read/avail/left are left as NULL and unused
-            rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
-                                 buf, ctypes.sizeof(buf), None, None, None)
-            self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
-            if buf.value:
-                self.assertEqual(msg, buf.value.decode())
-                break
-            time.sleep(0.1)
-            count += 1
-        else:
-            self.fail("Did not receive communication from the subprocess")
-
-        os.kill(proc.pid, sig)
-        self.assertEqual(proc.wait(), sig)
-
-    def test_kill_sigterm(self):
-        # SIGTERM doesn't mean anything special, but make sure it works
-        self._kill(signal.SIGTERM)
-
-    def test_kill_int(self):
-        # os.kill on Windows can take an int which gets set as the exit code
-        self._kill(100)
-
-    @unittest.skipIf(mmap is None, "requires mmap")
-    def _kill_with_event(self, event, name):
-        tagname = "test_os_%s" % uuid.uuid1()
-        m = mmap.mmap(-1, 1, tagname)
-        m[0] = 0
-
-        # Run a script which has console control handling enabled.
-        script = os.path.join(os.path.dirname(__file__),
-                              "win_console_handler.py")
-        cmd = [sys.executable, script, tagname]
-        proc = subprocess.Popen(cmd,
-                                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
-
-        with proc:
-            # Let the interpreter startup before we send signals. See #3137.
-            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
-                if proc.poll() is None:
-                    break
-            else:
-                # Forcefully kill the process if we weren't able to signal it.
-                proc.kill()
-                self.fail("Subprocess didn't finish initialization")
-
-            os.kill(proc.pid, event)
-
-            try:
-                # proc.send_signal(event) could also be done here.
-                # Allow time for the signal to be passed and the process to exit.
-                proc.wait(timeout=support.SHORT_TIMEOUT)
-            except subprocess.TimeoutExpired:
-                # Forcefully kill the process if we weren't able to signal it.
-                proc.kill()
-                self.fail("subprocess did not stop on {}".format(name))
-
-    @unittest.skip("subprocesses aren't inheriting Ctrl+C property")
-    @support.requires_subprocess()
-    def test_CTRL_C_EVENT(self):
-        from ctypes import wintypes
-        import ctypes
-
-        # Make a NULL value by creating a pointer with no argument.
-        NULL = ctypes.POINTER(ctypes.c_int)()
-        SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
-        SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
-                                          wintypes.BOOL)
-        SetConsoleCtrlHandler.restype = wintypes.BOOL
-
-        # Calling this with NULL and FALSE causes the calling process to
-        # handle Ctrl+C, rather than ignore it. This property is inherited
-        # by subprocesses.
-        SetConsoleCtrlHandler(NULL, 0)
-
-        self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
-
-    @support.requires_subprocess()
-    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(os_helper.TESTFN, dir_name)
-            file_name = 'FILE%d' % i
-            file_path = os.path.join(os_helper.TESTFN, file_name)
-            os.makedirs(dir_path)
-            with open(file_path, 'w', encoding='utf-8') 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(os_helper.TESTFN)
-
-    def test_listdir_no_extended_path(self):
-        """Test when the path is not an "extended" path."""
-        # unicode
-        self.assertEqual(
-                sorted(os.listdir(os_helper.TESTFN)),
-                self.created_paths)
-
-        # bytes
-        self.assertEqual(
-                sorted(os.listdir(os.fsencode(os_helper.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(os_helper.TESTFN)
-        self.assertEqual(
-                sorted(os.listdir(path)),
-                self.created_paths)
-
-        # bytes
-        path = b'\\\\?\\' + os.fsencode(os.path.abspath(os_helper.TESTFN))
-        self.assertEqual(
-                sorted(os.listdir(path)),
-                [os.fsencode(path) for path in self.created_paths])
-
-
-@unittest.skipUnless(os.name == "nt", "NT specific tests")
-class Win32ListdriveTests(unittest.TestCase):
-    """Test listdrive, listmounts and listvolume on Windows."""
-
-    def setUp(self):
-        # Get drives and volumes from fsutil
-        out = subprocess.check_output(
-            ["fsutil.exe", "volume", "list"],
-            cwd=os.path.join(os.getenv("SystemRoot", "\\Windows"), "System32"),
-            encoding="mbcs",
-            errors="ignore",
-        )
-        lines = out.splitlines()
-        self.known_volumes = {l for l in lines if l.startswith('\\\\?\\')}
-        self.known_drives = {l for l in lines if l[1:] == ':\\'}
-        self.known_mounts = {l for l in lines if l[1:3] == ':\\'}
-
-    def test_listdrives(self):
-        drives = os.listdrives()
-        self.assertIsInstance(drives, list)
-        self.assertSetEqual(
-            self.known_drives,
-            self.known_drives & set(drives),
-        )
-
-    def test_listvolumes(self):
-        volumes = os.listvolumes()
-        self.assertIsInstance(volumes, list)
-        self.assertSetEqual(
-            self.known_volumes,
-            self.known_volumes & set(volumes),
-        )
-
-    def test_listmounts(self):
-        for volume in os.listvolumes():
-            try:
-                mounts = os.listmounts(volume)
-            except OSError as ex:
-                if support.verbose:
-                    print("Skipping", volume, "because of", ex)
-            else:
-                self.assertIsInstance(mounts, list)
-                self.assertSetEqual(
-                    set(mounts),
-                    self.known_mounts & set(mounts),
-                )
-
 
 @unittest.skipUnless(hasattr(os, 'readlink'), 'needs os.readlink()')
 class ReadlinkTests(unittest.TestCase):
@@ -3149,370 +3234,6 @@ def test_bytes(self):
         self.assertIsInstance(path, bytes)
 
 
-@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
-@os_helper.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)
-        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))
-        self.check_stat(self.filelink, self.filelink_target)
-
-    def _create_missing_dir_link(self):
-        'Create a "directory" link to a non-existent target'
-        linkname = self.missing_link
-        if os.path.lexists(linkname):
-            os.remove(linkname)
-        target = r'c:\\target does not exist.29r3c740'
-        assert not os.path.exists(target)
-        target_is_dir = True
-        os.symlink(target, linkname, target_is_dir)
-
-    def test_remove_directory_link_to_missing_target(self):
-        self._create_missing_dir_link()
-        # For compatibility with Unix, os.remove will check the
-        #  directory status and call RemoveDirectory if the symlink
-        #  was created with target_is_dir==True.
-        os.remove(self.missing_link)
-
-    def test_isdir_on_directory_link_to_missing_target(self):
-        self._create_missing_dir_link()
-        self.assertFalse(os.path.isdir(self.missing_link))
-
-    def test_rmdir_on_directory_link_to_missing_target(self):
-        self._create_missing_dir_link()
-        os.rmdir(self.missing_link)
-
-    def check_stat(self, link, target):
-        self.assertEqual(os.stat(link), os.stat(target))
-        self.assertNotEqual(os.lstat(link), os.stat(link))
-
-        bytes_link = os.fsencode(link)
-        self.assertEqual(os.stat(bytes_link), os.stat(target))
-        self.assertNotEqual(os.lstat(bytes_link), os.stat(bytes_link))
-
-    def test_12084(self):
-        level1 = os.path.abspath(os_helper.TESTFN)
-        level2 = os.path.join(level1, "level2")
-        level3 = os.path.join(level2, "level3")
-        self.addCleanup(os_helper.rmtree, level1)
-
-        os.mkdir(level1)
-        os.mkdir(level2)
-        os.mkdir(level3)
-
-        file1 = os.path.abspath(os.path.join(level1, "file1"))
-        create_file(file1)
-
-        orig_dir = os.getcwd()
-        try:
-            os.chdir(level2)
-            link = os.path.join(level2, "link")
-            os.symlink(os.path.relpath(file1), "link")
-            self.assertIn("link", os.listdir(os.getcwd()))
-
-            # Check os.stat calls from the same dir as the link
-            self.assertEqual(os.stat(file1), os.stat("link"))
-
-            # Check os.stat calls from a dir below the link
-            os.chdir(level1)
-            self.assertEqual(os.stat(file1),
-                             os.stat(os.path.relpath(link)))
-
-            # Check os.stat calls from a dir above the link
-            os.chdir(level3)
-            self.assertEqual(os.stat(file1),
-                             os.stat(os.path.relpath(link)))
-        finally:
-            os.chdir(orig_dir)
-
-    @unittest.skipUnless(os.path.lexists(r'C:\Users\All Users')
-                            and os.path.exists(r'C:\ProgramData'),
-                            'Test directories not found')
-    def test_29248(self):
-        # os.symlink() calls CreateSymbolicLink, which creates
-        # the reparse data buffer with the print name stored
-        # first, so the offset is always 0. CreateSymbolicLink
-        # stores the "PrintName" DOS path (e.g. "C:\") first,
-        # with an offset of 0, followed by the "SubstituteName"
-        # NT path (e.g. "\??\C:\"). The "All Users" link, on
-        # the other hand, seems to have been created manually
-        # with an inverted order.
-        target = os.readlink(r'C:\Users\All Users')
-        self.assertTrue(os.path.samefile(target, r'C:\ProgramData'))
-
-    def test_buffer_overflow(self):
-        # Older versions would have a buffer overflow when detecting
-        # whether a link source was a directory. This test ensures we
-        # no longer crash, but does not otherwise validate the behavior
-        segment = 'X' * 27
-        path = os.path.join(*[segment] * 10)
-        test_cases = [
-            # overflow with absolute src
-            ('\\' + path, segment),
-            # overflow dest with relative src
-            (segment, path),
-            # overflow when joining src
-            (path[:180], path[:180]),
-        ]
-        for src, dest in test_cases:
-            try:
-                os.symlink(src, dest)
-            except FileNotFoundError:
-                pass
-            else:
-                try:
-                    os.remove(dest)
-                except OSError:
-                    pass
-            # Also test with bytes, since that is a separate code path.
-            try:
-                os.symlink(os.fsencode(src), os.fsencode(dest))
-            except FileNotFoundError:
-                pass
-            else:
-                try:
-                    os.remove(dest)
-                except OSError:
-                    pass
-
-    def test_appexeclink(self):
-        root = os.path.expandvars(r'%LOCALAPPDATA%\Microsoft\WindowsApps')
-        if not os.path.isdir(root):
-            self.skipTest("test requires a WindowsApps directory")
-
-        aliases = [os.path.join(root, a)
-                   for a in fnmatch.filter(os.listdir(root), '*.exe')]
-
-        for alias in aliases:
-            if support.verbose:
-                print()
-                print("Testing with", alias)
-            st = os.lstat(alias)
-            self.assertEqual(st, os.stat(alias))
-            self.assertFalse(stat.S_ISLNK(st.st_mode))
-            self.assertEqual(st.st_reparse_tag, stat.IO_REPARSE_TAG_APPEXECLINK)
-            self.assertTrue(os.path.isfile(alias))
-            # testing the first one we see is sufficient
-            break
-        else:
-            self.skipTest("test requires an app execution alias")
-
-@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
-class Win32JunctionTests(unittest.TestCase):
-    junction = 'junctiontest'
-    junction_target = os.path.dirname(os.path.abspath(__file__))
-
-    def setUp(self):
-        assert os.path.exists(self.junction_target)
-        assert not os.path.lexists(self.junction)
-
-    def tearDown(self):
-        if os.path.lexists(self.junction):
-            os.unlink(self.junction)
-
-    def test_create_junction(self):
-        _winapi.CreateJunction(self.junction_target, self.junction)
-        self.assertTrue(os.path.lexists(self.junction))
-        self.assertTrue(os.path.exists(self.junction))
-        self.assertTrue(os.path.isdir(self.junction))
-        self.assertNotEqual(os.stat(self.junction), os.lstat(self.junction))
-        self.assertEqual(os.stat(self.junction), os.stat(self.junction_target))
-
-        # bpo-37834: Junctions are not recognized as links.
-        self.assertFalse(os.path.islink(self.junction))
-        self.assertEqual(os.path.normcase("\\\\?\\" + self.junction_target),
-                         os.path.normcase(os.readlink(self.junction)))
-
-    def test_unlink_removes_junction(self):
-        _winapi.CreateJunction(self.junction_target, self.junction)
-        self.assertTrue(os.path.exists(self.junction))
-        self.assertTrue(os.path.lexists(self.junction))
-
-        os.unlink(self.junction)
-        self.assertFalse(os.path.exists(self.junction))
-
-@unittest.skipUnless(sys.platform == "win32", "Win32 specific tests")
-class Win32NtTests(unittest.TestCase):
-    def test_getfinalpathname_handles(self):
-        nt = import_helper.import_module('nt')
-        ctypes = import_helper.import_module('ctypes')
-        # Ruff false positive -- it thinks we're redefining `ctypes` here
-        import ctypes.wintypes  # noqa: F811
-
-        kernel = ctypes.WinDLL('Kernel32.dll', use_last_error=True)
-        kernel.GetCurrentProcess.restype = ctypes.wintypes.HANDLE
-
-        kernel.GetProcessHandleCount.restype = ctypes.wintypes.BOOL
-        kernel.GetProcessHandleCount.argtypes = (ctypes.wintypes.HANDLE,
-                                                 ctypes.wintypes.LPDWORD)
-
-        # This is a pseudo-handle that doesn't need to be closed
-        hproc = kernel.GetCurrentProcess()
-
-        handle_count = ctypes.wintypes.DWORD()
-        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
-        self.assertEqual(1, ok)
-
-        before_count = handle_count.value
-
-        # The first two test the error path, __file__ tests the success path
-        filenames = [
-            r'\\?\C:',
-            r'\\?\NUL',
-            r'\\?\CONIN',
-            __file__,
-        ]
-
-        for _ in range(10):
-            for name in filenames:
-                try:
-                    nt._getfinalpathname(name)
-                except Exception:
-                    # Failure is expected
-                    pass
-                try:
-                    os.stat(name)
-                except Exception:
-                    pass
-
-        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
-        self.assertEqual(1, ok)
-
-        handle_delta = handle_count.value - before_count
-
-        self.assertEqual(0, handle_delta)
-
-    @support.requires_subprocess()
-    def test_stat_unlink_race(self):
-        # bpo-46785: the implementation of os.stat() falls back to reading
-        # the parent directory if CreateFileW() fails with a permission
-        # error. If reading the parent directory fails because the file or
-        # directory are subsequently unlinked, or because the volume or
-        # share are no longer available, then the original permission error
-        # should not be restored.
-        filename =  os_helper.TESTFN
-        self.addCleanup(os_helper.unlink, filename)
-        deadline = time.time() + 5
-        command = textwrap.dedent("""\
-            import os
-            import sys
-            import time
-
-            filename = sys.argv[1]
-            deadline = float(sys.argv[2])
-
-            while time.time() < deadline:
-                try:
-                    with open(filename, "w") as f:
-                        pass
-                except OSError:
-                    pass
-                try:
-                    os.remove(filename)
-                except OSError:
-                    pass
-            """)
-
-        with subprocess.Popen([sys.executable, '-c', command, filename, str(deadline)]) as proc:
-            while time.time() < deadline:
-                try:
-                    os.stat(filename)
-                except FileNotFoundError as e:
-                    assert e.winerror == 2  # ERROR_FILE_NOT_FOUND
-            try:
-                proc.wait(1)
-            except subprocess.TimeoutExpired:
-                proc.terminate()
-
-    @support.requires_subprocess()
-    def test_stat_inaccessible_file(self):
-        filename = os_helper.TESTFN
-        ICACLS = os.path.expandvars(r"%SystemRoot%\System32\icacls.exe")
-
-        with open(filename, "wb") as f:
-            f.write(b'Test data')
-
-        stat1 = os.stat(filename)
-
-        try:
-            # Remove all permissions from the file
-            subprocess.check_output([ICACLS, filename, "/inheritance:r"],
-                                    stderr=subprocess.STDOUT)
-        except subprocess.CalledProcessError as ex:
-            if support.verbose:
-                print(ICACLS, filename, "/inheritance:r", "failed.")
-                print(ex.stdout.decode("oem", "replace").rstrip())
-            try:
-                os.unlink(filename)
-            except OSError:
-                pass
-            self.skipTest("Unable to create inaccessible file")
-
-        def cleanup():
-            # Give delete permission to the owner (us)
-            subprocess.check_output([ICACLS, filename, "/grant", "*WD:(D)"],
-                                    stderr=subprocess.STDOUT)
-            os.unlink(filename)
-
-        self.addCleanup(cleanup)
-
-        if support.verbose:
-            print("File:", filename)
-            print("stat with access:", stat1)
-
-        # First test - we shouldn't raise here, because we still have access to
-        # the directory and can extract enough information from its metadata.
-        stat2 = os.stat(filename)
-
-        if support.verbose:
-            print(" without access:", stat2)
-
-        # We may not get st_dev/st_ino, so ensure those are 0 or match
-        self.assertIn(stat2.st_dev, (0, stat1.st_dev))
-        self.assertIn(stat2.st_ino, (0, stat1.st_ino))
-
-        # st_mode and st_size should match (for a normal file, at least)
-        self.assertEqual(stat1.st_mode, stat2.st_mode)
-        self.assertEqual(stat1.st_size, stat2.st_size)
-
-        # st_ctime and st_mtime should be the same
-        self.assertEqual(stat1.st_ctime, stat2.st_ctime)
-        self.assertEqual(stat1.st_mtime, stat2.st_mtime)
-
-        # st_atime should be the same or later
-        self.assertGreaterEqual(stat1.st_atime, stat2.st_atime)
-
-
 @os_helper.skip_unless_symlink
 class NonLocalSymlinkTests(unittest.TestCase):
 
@@ -3589,6 +3310,7 @@ def test_getppid(self):
         self.assertEqual(error, b'')
         self.assertEqual(int(stdout), os.getpid())
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     def check_waitpid(self, code, exitcode, callback=None):
         if sys.platform == 'win32':
             # On Windows, os.spawnv() simply joins arguments with spaces:
@@ -3691,30 +3413,35 @@ def create_args(self, *, with_env=False, use_bytes=False):
 
         return program, args
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnl')
     def test_spawnl(self):
         program, args = self.create_args()
         exitcode = os.spawnl(os.P_WAIT, program, *args)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnle')
     def test_spawnle(self):
         program, args = self.create_args(with_env=True)
         exitcode = os.spawnle(os.P_WAIT, program, *args, self.env)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnlp')
     def test_spawnlp(self):
         program, args = self.create_args()
         exitcode = os.spawnlp(os.P_WAIT, program, *args)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnlpe')
     def test_spawnlpe(self):
         program, args = self.create_args(with_env=True)
         exitcode = os.spawnlpe(os.P_WAIT, program, *args, self.env)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnv')
     def test_spawnv(self):
         program, args = self.create_args()
@@ -3725,30 +3452,35 @@ def test_spawnv(self):
         exitcode = os.spawnv(os.P_WAIT, FakePath(program), args)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnve')
     def test_spawnve(self):
         program, args = self.create_args(with_env=True)
         exitcode = os.spawnve(os.P_WAIT, program, args, self.env)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnvp')
     def test_spawnvp(self):
         program, args = self.create_args()
         exitcode = os.spawnvp(os.P_WAIT, program, args)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnvpe')
     def test_spawnvpe(self):
         program, args = self.create_args(with_env=True)
         exitcode = os.spawnvpe(os.P_WAIT, program, args, self.env)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnv')
     def test_nowait(self):
         program, args = self.create_args()
         pid = os.spawnv(os.P_NOWAIT, program, args)
         support.wait_process(pid, exitcode=self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnve')
     def test_spawnve_bytes(self):
         # Test bytes handling in parse_arglist and parse_envlist (#28114)
@@ -3756,18 +3488,21 @@ def test_spawnve_bytes(self):
         exitcode = os.spawnve(os.P_WAIT, program, args, self.env)
         self.assertEqual(exitcode, self.exitcode)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnl')
     def test_spawnl_noargs(self):
         program, __ = self.create_args()
         self.assertRaises(ValueError, os.spawnl, os.P_NOWAIT, program)
         self.assertRaises(ValueError, os.spawnl, os.P_NOWAIT, program, '')
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnle')
     def test_spawnle_noargs(self):
         program, __ = self.create_args()
         self.assertRaises(ValueError, os.spawnle, os.P_NOWAIT, program, {})
         self.assertRaises(ValueError, os.spawnle, os.P_NOWAIT, program, '', {})
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnv')
     def test_spawnv_noargs(self):
         program, __ = self.create_args()
@@ -3776,6 +3511,7 @@ def test_spawnv_noargs(self):
         self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, program, ('',))
         self.assertRaises(ValueError, os.spawnv, os.P_NOWAIT, program, [''])
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnve')
     def test_spawnve_noargs(self):
         program, __ = self.create_args()
@@ -3784,7 +3520,7 @@ def test_spawnve_noargs(self):
         self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, program, ('',), {})
         self.assertRaises(ValueError, os.spawnve, os.P_NOWAIT, program, [''], {})
 
-    @unittest.expectedFailureIfWindows('TODO: RUSTPYTHON; AssertionError: "must contain only strings" does not match "expected str, bytes or os.PathLike object, not int"')
+    @unittest.expectedFailureIfWindows("TODO: RUSTPYTHON; AssertionError: \"must contain only strings\" does not match \"expected str, bytes or os.PathLike object, not int\"")
     @requires_native_spawnv
     def test_spawnv_arg_conversion_errors(self):
         # A non-path argv item gets a TypeError naming the argument...
@@ -3852,22 +3588,28 @@ def _test_invalid_env(self, spawn):
         exitcode = spawn(os.P_WAIT, program, args, newenv)
         self.assertEqual(exitcode, 0)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnve')
     def test_spawnve_invalid_env(self):
         self._test_invalid_env(os.spawnve)
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @requires_os_func('spawnvpe')
     def test_spawnvpe_invalid_env(self):
         self._test_invalid_env(os.spawnvpe)
 
 
-# The introduction of this TestCase caused at least two different errors on
-# *nix buildbots. Temporarily skip this to let the buildbots move along.
-@unittest.skip("Skip due to platform/environment differences on *NIX buildbots")
 @unittest.skipUnless(hasattr(os, 'getlogin'), "test needs os.getlogin")
 class LoginTests(unittest.TestCase):
     def test_getlogin(self):
-        user_name = os.getlogin()
+        try:
+            user_name = os.getlogin()
+        except OSError as exc:
+            # See https://man7.org/linux/man-pages/man3/getlogin.3.html#ERRORS.
+            if exc.errno in (errno.ENXIO, errno.ENOENT, errno.ENOTTY):
+                self.skipTest(str(exc))
+            else:
+                raise
         self.assertNotEqual(len(user_name), 0)
 
 
@@ -4031,6 +3773,11 @@ async def test_invalid_offset(self):
             await self.async_sendfile(self.sockno, self.fileno, -1, 4096)
         self.assertEqual(cm.exception.errno, errno.EINVAL)
 
+    async def test_invalid_count(self):
+        with self.assertRaises(ValueError, msg="count cannot be negative"):
+            await self.sendfile_wrapper(self.sockno, self.fileno, offset=0,
+                                        count=-1)
+
     async def test_keywords(self):
         # Keyword arguments should be supported
         await self.async_sendfile(out_fd=self.sockno, in_fd=self.fileno,
@@ -4221,12 +3968,7 @@ def test_does_not_crash(self):
         try:
             size = os.get_terminal_size()
         except OSError as e:
-            known_errnos = [errno.EINVAL, errno.ENOTTY]
-            if sys.platform == "android":
-                # The Android testbed redirects the native stdout to a pipe,
-                # which returns a different error code.
-                known_errnos.append(errno.EACCES)
-            if sys.platform == "win32" or e.errno in known_errnos:
+            if sys.platform == "win32" or e.errno in (errno.EINVAL, errno.ENOTTY):
                 # Under win32 a generic OSError can be thrown if the
                 # handle cannot be retrieved
                 self.skipTest("failed to query terminal size")
@@ -4737,6 +4479,7 @@ def test_oserror_filename(self):
             (self.filenames, os.listdir,),
             (self.filenames, os.rename, "dst"),
             (self.filenames, os.replace, "dst"),
+            (self.filenames, os.utime, None),
         ]
         if os_helper.can_chmod():
             funcs.append((self.filenames, os.chmod, 0o777))
@@ -4777,6 +4520,19 @@ def test_oserror_filename(self):
                 else:
                     self.fail(f"No exception thrown by {func}")
 
+    def test_mkdir(self):
+        filename = os_helper.TESTFN
+        subdir = os.path.join(filename, 'subdir')
+        self.assertRaises(FileNotFoundError, os.mkdir, subdir)
+
+        self.addCleanup(os_helper.unlink, filename)
+        create_file(filename)
+        self.assertRaises(FileExistsError, os.mkdir, filename)
+
+        self.assertRaises((NotADirectoryError, FileNotFoundError),
+                          os.mkdir, subdir)
+
+
 class CPUCountTests(unittest.TestCase):
     def check_cpu_count(self, cpus):
         if cpus is None:
@@ -4992,6 +4748,7 @@ def test_posix_pty_functions(self):
                     signal.signal(signal.SIGHUP, old_handler)
         self.assertEqual(os.ptsname(mother_fd), os.ttyname(son_fd))
 
+    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
     @unittest.skipUnless(hasattr(os, 'spawnl'), "need os.spawnl()")
     @support.requires_subprocess()
     def test_pipe_spawnl(self):
diff --git a/Lib/test/test_posix.py b/Lib/test/test_os/test_posix.py
similarity index 94%
rename from Lib/test/test_posix.py
rename to Lib/test/test_os/test_posix.py
index 265f68902a4..4839b1772a3 100644
--- a/Lib/test/test_posix.py
+++ b/Lib/test/test_os/test_posix.py
@@ -46,7 +46,8 @@ def _supports_sched():
     try:
         posix.sched_getscheduler(0)
     except OSError as e:
-        if e.errno == errno.ENOSYS:
+        # DragonFly BSD requires privileges to use the scheduler API.
+        if e.errno in (errno.ENOSYS, errno.EPERM):
             return False
     return True
 
@@ -70,7 +71,7 @@ def testNoArgFunctions(self):
         NO_ARG_FUNCTIONS = [ "ctermid", "getcwd", "getcwdb", "uname",
                              "times", "getloadavg",
                              "getegid", "geteuid", "getgid", "getgroups",
-                             "getpid", "getpgrp", "getppid", "getuid", "sync",
+                             "getpid", "getpgrp", "getppid", "getuid",
                            ]
 
         for name in NO_ARG_FUNCTIONS:
@@ -80,6 +81,13 @@ def testNoArgFunctions(self):
                     posix_func()
                     self.assertRaises(TypeError, posix_func, 1)
 
+    # gh-102184: sync() can block for a long time.
+    @support.requires_resource('walltime')
+    @unittest.skipUnless(hasattr(posix, 'sync'), 'test needs posix.sync()')
+    def test_sync(self):
+        posix.sync()
+        self.assertRaises(TypeError, posix.sync, 1)
+
     @unittest.skipUnless(hasattr(posix, 'getresuid'),
                          'test needs posix.getresuid()')
     def test_getresuid(self):
@@ -415,8 +423,10 @@ def test_posix_fallocate(self):
             if inst.errno == errno.EINVAL and sys.platform.startswith(
                 ('sunos', 'freebsd', 'openbsd', 'gnukfreebsd')):
                 raise unittest.SkipTest("test may fail on ZFS filesystems")
-            elif inst.errno == errno.EOPNOTSUPP and sys.platform.startswith("netbsd"):
-                raise unittest.SkipTest("test may fail on FFS filesystems")
+            elif inst.errno == errno.EOPNOTSUPP:
+                # ZFS on FreeBSD, FFS on NetBSD, etc.
+                raise unittest.SkipTest(
+                    "the file system does not support posix_fallocate()")
             else:
                 raise
         finally:
@@ -680,22 +690,65 @@ def test_stat_fd_zero_follow_symlinks(self):
                 'cannot use fd and follow_symlinks together'):
             posix.stat(1, follow_symlinks=False)
 
-    def test_stat(self):
-        self.assertTrue(posix.stat(os_helper.TESTFN))
-        self.assertTrue(posix.stat(os.fsencode(os_helper.TESTFN)))
+    def check_statlike_path(self, func):
+        self.assertTrue(func(os_helper.TESTFN))
+        self.assertTrue(func(os.fsencode(os_helper.TESTFN)))
+        self.assertTrue(func(os_helper.FakePath(os_helper.TESTFN)))
 
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, bytearray(os.fsencode(os_helper.TESTFN)))
+                func, bytearray(os.fsencode(os_helper.TESTFN)))
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, None)
+                func, None)
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, list(os_helper.TESTFN))
+                func, list(os_helper.TESTFN))
         self.assertRaisesRegex(TypeError,
                 'should be string, bytes, os.PathLike or integer, not',
-                posix.stat, list(os.fsencode(os_helper.TESTFN)))
+                func, list(os.fsencode(os_helper.TESTFN)))
+
+    def test_stat(self):
+        self.check_statlike_path(posix.stat)
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), 'test needs posix.statx()')
+    def test_statx(self):
+        def func(path, **kwargs):
+            return posix.statx(path, posix.STATX_BASIC_STATS, **kwargs)
+        self.check_statlike_path(func)
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), 'test needs posix.statx()')
+    def test_statx_flags(self):
+        # glibc's fallback implementation of statx via the stat family fails
+        # with EINVAL on the (nonzero) sync flags.  If you see this failure,
+        # update your kernel and/or seccomp syscall filter.
+        valid_flag_names = ('AT_NO_AUTOMOUNT', 'AT_STATX_SYNC_AS_STAT',
+                            'AT_STATX_FORCE_SYNC', 'AT_STATX_DONT_SYNC')
+        for flag_name in valid_flag_names:
+            flag = getattr(posix, flag_name)
+            with self.subTest(msg=flag_name, flags=flag):
+                posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                            flags=flag)
+
+        # These flags are not exposed to Python because their functionality is
+        # implemented via kwargs instead.
+        kwarg_equivalent_flags = (
+            (0x0100, 'AT_SYMLINK_NOFOLLOW', 'follow_symlinks'),
+            (0x0400, 'AT_SYMLINK_FOLLOW', 'follow_symlinks'),
+            (0x1000, 'AT_EMPTY_PATH', 'dir_fd'),
+        )
+        for flag, flag_name, kwarg_name in kwarg_equivalent_flags:
+            with self.subTest(msg=flag_name, flags=flag):
+                with self.assertRaisesRegex(ValueError, kwarg_name):
+                    posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                                flags=flag)
+
+        with self.subTest(msg="AT_STATX_FORCE_SYNC | AT_STATX_DONT_SYNC"):
+            with self.assertRaises(OSError) as ctx:
+                flags = posix.AT_STATX_FORCE_SYNC | posix.AT_STATX_DONT_SYNC
+                posix.statx(os_helper.TESTFN, posix.STATX_BASIC_STATS,
+                            flags=flags)
+            self.assertEqual(ctx.exception.errno, errno.EINVAL)
 
     @unittest.skipUnless(hasattr(posix, 'mkfifo'), "don't have mkfifo()")
     def test_mkfifo(self):
@@ -769,12 +822,29 @@ def test_makedev(self):
             self.assertRaises((ValueError, OverflowError), posix.makedev, x, minor)
             self.assertRaises((ValueError, OverflowError), posix.makedev, major, x)
 
-        if sys.platform == 'linux' and not support.linked_to_musl():
-            NODEV = -1
+        # The following tests are needed to test functions accepting or
+        # returning the special value NODEV (if it is defined). major(), minor()
+        # and makefile() are the only easily reproducible examples, but that
+        # behavior is platform specific -- on some platforms their code has
+        # a special case for NODEV, on others this is just an implementation
+        # artifact.
+        if (hasattr(posix, 'NODEV') and
+            sys.platform.startswith(('linux', 'macos', 'freebsd', 'sunos'))):
+            NODEV = posix.NODEV
             self.assertEqual(posix.major(NODEV), NODEV)
             self.assertEqual(posix.minor(NODEV), NODEV)
             self.assertEqual(posix.makedev(NODEV, NODEV), NODEV)
 
+    def test_nodev(self):
+        # NODEV is not a part of Posix, but is defined on many systems.
+        if (not hasattr(posix, 'NODEV')
+            and (not sys.platform.startswith(('linux', 'macos', 'freebsd',
+                                              'dragonfly', 'netbsd', 'openbsd',
+                                              'sunos'))
+                 or support.linked_to_musl())):
+            self.skipTest('not defined on this platform')
+        self.assertHasAttr(posix, 'NODEV')
+
     def _test_all_chown_common(self, chown_func, first_param, stat_func):
         """Common code for chown, fchown and lchown tests."""
         def check_stat(uid, gid):
@@ -838,7 +908,9 @@ def check_stat(uid, gid):
             self.assertRaises(OSError, chown_func, first_param, 0, -1)
             check_stat(uid, gid)
             if hasattr(os, 'getgroups'):
-                if 0 not in os.getgroups():
+                # Also check the effective gid, which the kernel
+                # accepts for chown even if not in getgroups().
+                if 0 not in os.getgroups() and os.getegid() != 0:
                     self.assertRaises(OSError, chown_func, first_param, -1, 0)
                     check_stat(uid, gid)
         # test illegal types
@@ -1386,6 +1458,7 @@ def test_bug_140634(self):
         del sched_priority, param  # should not crash
         support.gc_collect()  # just to be sure
 
+    @requires_sched
     @unittest.skipUnless(hasattr(posix, "sched_rr_get_interval"), "no function")
     def test_sched_rr_get_interval(self):
         try:
@@ -1631,33 +1704,47 @@ def test_chown_dir_fd(self):
         with self.prepare_file() as (dir_fd, name, fullname):
             posix.chown(name, os.getuid(), os.getgid(), dir_fd=dir_fd)
 
-    @unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
-    def test_stat_dir_fd(self):
+    def check_statlike_dir_fd(self, func, prefix):
         with self.prepare() as (dir_fd, name, fullname):
             with open(fullname, 'w') as outfile:
                 outfile.write("testline\n")
             self.addCleanup(posix.unlink, fullname)
 
-            s1 = posix.stat(fullname)
-            s2 = posix.stat(name, dir_fd=dir_fd)
-            self.assertEqual(s1, s2)
-            s2 = posix.stat(fullname, dir_fd=None)
-            self.assertEqual(s1, s2)
+            def get(result, attr):
+                return getattr(result, prefix + attr)
+
+            s1 = func(fullname)
+            s2 = func(name, dir_fd=dir_fd)
+            self.assertEqual((get(s1, "dev"), get(s1, "ino")),
+                             (get(s2, "dev"), get(s2, "ino")))
+            s2 = func(fullname, dir_fd=None)
+            self.assertEqual((get(s1, "dev"), get(s1, "ino")),
+                             (get(s2, "dev"), get(s2, "ino")))
 
             self.assertRaisesRegex(TypeError, 'should be integer or None, not',
-                    posix.stat, name, dir_fd=posix.getcwd())
+                    func, name, dir_fd=posix.getcwd())
             self.assertRaisesRegex(TypeError, 'should be integer or None, not',
-                    posix.stat, name, dir_fd=float(dir_fd))
+                    func, name, dir_fd=float(dir_fd))
             self.assertRaises(OverflowError,
-                    posix.stat, name, dir_fd=10**20)
+                    func, name, dir_fd=10**20)
 
             for fd in False, True:
                 with self.assertWarnsRegex(RuntimeWarning,
                         'bool is used as a file descriptor') as cm:
                     with self.assertRaises(OSError):
-                        posix.stat('nonexisting', dir_fd=fd)
+                        func('nonexisting', dir_fd=fd)
                 self.assertEqual(cm.filename, __file__)
 
+    @unittest.skipUnless(os.stat in os.supports_dir_fd, "test needs dir_fd support in os.stat()")
+    def test_stat_dir_fd(self):
+        self.check_statlike_dir_fd(posix.stat, prefix="st_")
+
+    @unittest.skipUnless(hasattr(posix, 'statx'), "test needs os.statx()")
+    def test_statx_dir_fd(self):
+        def func(path, **kwargs):
+            return posix.statx(path, os.STATX_INO, **kwargs)
+        self.check_statlike_dir_fd(func, prefix="stx_")
+
     @unittest.skipUnless(os.utime in os.supports_dir_fd, "test needs dir_fd support in os.utime()")
     def test_utime_dir_fd(self):
         with self.prepare_file() as (dir_fd, name, fullname):
@@ -2271,6 +2358,22 @@ def test_pwritev(self):
             self.assertNotHasAttr(os, "pwritev")
             self.assertNotHasAttr(os, "preadv")
 
+    def test_pipe2(self):
+        self._verify_available("HAVE_PIPE2")
+        if self.mac_ver >= (27, 0):
+            self.assertHasAttr(os, "pipe2")
+        else:
+            self.assertNotHasAttr(os, "pipe2")
+
+    def test_dup3(self):
+        self._verify_available("HAVE_DUP3")
+        r, w = os.pipe()
+        self.addCleanup(os.close, r)
+        self.addCleanup(os.close, w)
+        # Must not crash even when dup3 unavailable at runtime.
+        # os.dup2 returns fd2 (here w); do not double-close.
+        os.dup2(r, w, inheritable=False)
+
     def test_stat(self):
         self._verify_available("HAVE_FSTATAT")
         if self.mac_ver >= (10, 10):
diff --git a/Lib/test/test_os/test_windows.py b/Lib/test/test_os/test_windows.py
new file mode 100644
index 00000000000..f1c6283f60d
--- /dev/null
+++ b/Lib/test/test_os/test_windows.py
@@ -0,0 +1,605 @@
+import sys
+import unittest
+
+if sys.platform != "win32":
+    raise unittest.SkipTest("Win32 specific tests")
+
+import _winapi
+import fnmatch
+import mmap
+import os
+import shutil
+import signal
+import stat
+import subprocess
+import textwrap
+import time
+import uuid
+from test import support
+from test.support import import_helper
+from test.support import os_helper
+from .utils import create_file
+
+
+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
+        import msvcrt
+
+        # Since we can't access the contents of the process' stdout until the
+        # process has exited, use PeekNamedPipe to see what's inside stdout
+        # without waiting. This is done so we can tell that the interpreter
+        # is started and running at a point where it could handle a signal.
+        PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe
+        PeekNamedPipe.restype = wintypes.BOOL
+        PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle
+                                  ctypes.POINTER(ctypes.c_char), # stdout buf
+                                  wintypes.DWORD, # Buffer size
+                                  ctypes.POINTER(wintypes.DWORD), # bytes read
+                                  ctypes.POINTER(wintypes.DWORD), # bytes avail
+                                  ctypes.POINTER(wintypes.DWORD)) # bytes left
+        msg = "running"
+        proc = subprocess.Popen([sys.executable, "-c",
+                                 "import sys;"
+                                 "sys.stdout.write('{}');"
+                                 "sys.stdout.flush();"
+                                 "input()".format(msg)],
+                                stdout=subprocess.PIPE,
+                                stderr=subprocess.PIPE,
+                                stdin=subprocess.PIPE)
+        self.addCleanup(proc.stdout.close)
+        self.addCleanup(proc.stderr.close)
+        self.addCleanup(proc.stdin.close)
+
+        count, max = 0, 100
+        while count < max and proc.poll() is None:
+            # Create a string buffer to store the result of stdout from the pipe
+            buf = ctypes.create_string_buffer(len(msg))
+            # Obtain the text currently in proc.stdout
+            # Bytes read/avail/left are left as NULL and unused
+            rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()),
+                                 buf, ctypes.sizeof(buf), None, None, None)
+            self.assertNotEqual(rslt, 0, "PeekNamedPipe failed")
+            if buf.value:
+                self.assertEqual(msg, buf.value.decode())
+                break
+            time.sleep(0.1)
+            count += 1
+        else:
+            self.fail("Did not receive communication from the subprocess")
+
+        os.kill(proc.pid, sig)
+        self.assertEqual(proc.wait(), sig)
+
+    def test_kill_sigterm(self):
+        # SIGTERM doesn't mean anything special, but make sure it works
+        self._kill(signal.SIGTERM)
+
+    def test_kill_int(self):
+        # os.kill on Windows can take an int which gets set as the exit code
+        self._kill(100)
+
+    @unittest.skipIf(mmap is None, "requires mmap")
+    def _kill_with_event(self, event, name):
+        tagname = "test_os_%s" % uuid.uuid1()
+        m = mmap.mmap(-1, 1, tagname)
+        m[0] = 0
+
+        # Run a script which has console control handling enabled.
+        script = os.path.join(os.path.dirname(__file__),
+                              "win_console_handler.py")
+        cmd = [sys.executable, script, tagname]
+        proc = subprocess.Popen(cmd,
+                                creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
+
+        with proc:
+            # Let the interpreter startup before we send signals. See #3137.
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
+                if proc.poll() is None:
+                    break
+            else:
+                # Forcefully kill the process if we weren't able to signal it.
+                proc.kill()
+                self.fail("Subprocess didn't finish initialization")
+
+            os.kill(proc.pid, event)
+
+            try:
+                # proc.send_signal(event) could also be done here.
+                # Allow time for the signal to be passed and the process to exit.
+                proc.wait(timeout=support.SHORT_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                # Forcefully kill the process if we weren't able to signal it.
+                proc.kill()
+                self.fail("subprocess did not stop on {}".format(name))
+
+    @unittest.skip("subprocesses aren't inheriting Ctrl+C property")
+    @support.requires_subprocess()
+    def test_CTRL_C_EVENT(self):
+        from ctypes import wintypes
+        import ctypes
+
+        # Make a NULL value by creating a pointer with no argument.
+        NULL = ctypes.POINTER(ctypes.c_int)()
+        SetConsoleCtrlHandler = ctypes.windll.kernel32.SetConsoleCtrlHandler
+        SetConsoleCtrlHandler.argtypes = (ctypes.POINTER(ctypes.c_int),
+                                          wintypes.BOOL)
+        SetConsoleCtrlHandler.restype = wintypes.BOOL
+
+        # Calling this with NULL and FALSE causes the calling process to
+        # handle Ctrl+C, rather than ignore it. This property is inherited
+        # by subprocesses.
+        SetConsoleCtrlHandler(NULL, 0)
+
+        self._kill_with_event(signal.CTRL_C_EVENT, "CTRL_C_EVENT")
+
+    @support.requires_subprocess()
+    def test_CTRL_BREAK_EVENT(self):
+        self._kill_with_event(signal.CTRL_BREAK_EVENT, "CTRL_BREAK_EVENT")
+
+
+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(os_helper.TESTFN, dir_name)
+            file_name = 'FILE%d' % i
+            file_path = os.path.join(os_helper.TESTFN, file_name)
+            os.makedirs(dir_path)
+            with open(file_path, 'w', encoding='utf-8') 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(os_helper.TESTFN)
+
+    def test_listdir_no_extended_path(self):
+        """Test when the path is not an "extended" path."""
+        # unicode
+        self.assertEqual(
+                sorted(os.listdir(os_helper.TESTFN)),
+                self.created_paths)
+
+        # bytes
+        self.assertEqual(
+                sorted(os.listdir(os.fsencode(os_helper.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(os_helper.TESTFN)
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                self.created_paths)
+
+        # bytes
+        path = b'\\\\?\\' + os.fsencode(os.path.abspath(os_helper.TESTFN))
+        self.assertEqual(
+                sorted(os.listdir(path)),
+                [os.fsencode(path) for path in self.created_paths])
+
+
+@unittest.skipUnless(os.name == "nt", "NT specific tests")
+class Win32ListdriveTests(unittest.TestCase):
+    """Test listdrive, listmounts and listvolume on Windows."""
+
+    def setUp(self):
+        # Get drives and volumes from fsutil
+        out = subprocess.check_output(
+            ["fsutil.exe", "volume", "list"],
+            cwd=os.path.join(os.getenv("SystemRoot", "\\Windows"), "System32"),
+            encoding="mbcs",
+            errors="ignore",
+        )
+        lines = out.splitlines()
+        self.known_volumes = {l for l in lines if l.startswith('\\\\?\\')}
+        self.known_drives = {l for l in lines if l[1:] == ':\\'}
+        self.known_mounts = {l for l in lines if l[1:3] == ':\\'}
+
+    def test_listdrives(self):
+        drives = os.listdrives()
+        self.assertIsInstance(drives, list)
+        self.assertSetEqual(
+            self.known_drives,
+            self.known_drives & set(drives),
+        )
+
+    def test_listvolumes(self):
+        volumes = os.listvolumes()
+        self.assertIsInstance(volumes, list)
+        self.assertSetEqual(
+            self.known_volumes,
+            self.known_volumes & set(volumes),
+        )
+
+    def test_listmounts(self):
+        for volume in os.listvolumes():
+            try:
+                mounts = os.listmounts(volume)
+            except OSError as ex:
+                if support.verbose:
+                    print("Skipping", volume, "because of", ex)
+            else:
+                self.assertIsInstance(mounts, list)
+                self.assertSetEqual(
+                    set(mounts),
+                    self.known_mounts & set(mounts),
+                )
+
+
+@os_helper.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)
+        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))
+        self.check_stat(self.filelink, self.filelink_target)
+
+    def _create_missing_dir_link(self):
+        'Create a "directory" link to a non-existent target'
+        linkname = self.missing_link
+        if os.path.lexists(linkname):
+            os.remove(linkname)
+        target = r'c:\\target does not exist.29r3c740'
+        assert not os.path.exists(target)
+        target_is_dir = True
+        os.symlink(target, linkname, target_is_dir)
+
+    def test_remove_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        # For compatibility with Unix, os.remove will check the
+        #  directory status and call RemoveDirectory if the symlink
+        #  was created with target_is_dir==True.
+        os.remove(self.missing_link)
+
+    def test_isdir_on_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        self.assertFalse(os.path.isdir(self.missing_link))
+
+    def test_rmdir_on_directory_link_to_missing_target(self):
+        self._create_missing_dir_link()
+        os.rmdir(self.missing_link)
+
+    def check_stat(self, link, target):
+        self.assertEqual(os.stat(link), os.stat(target))
+        self.assertNotEqual(os.lstat(link), os.stat(link))
+
+        bytes_link = os.fsencode(link)
+        self.assertEqual(os.stat(bytes_link), os.stat(target))
+        self.assertNotEqual(os.lstat(bytes_link), os.stat(bytes_link))
+
+    def test_12084(self):
+        level1 = os.path.abspath(os_helper.TESTFN)
+        level2 = os.path.join(level1, "level2")
+        level3 = os.path.join(level2, "level3")
+        self.addCleanup(os_helper.rmtree, level1)
+
+        os.mkdir(level1)
+        os.mkdir(level2)
+        os.mkdir(level3)
+
+        file1 = os.path.abspath(os.path.join(level1, "file1"))
+        create_file(file1)
+
+        orig_dir = os.getcwd()
+        try:
+            os.chdir(level2)
+            link = os.path.join(level2, "link")
+            os.symlink(os.path.relpath(file1), "link")
+            self.assertIn("link", os.listdir(os.getcwd()))
+
+            # Check os.stat calls from the same dir as the link
+            self.assertEqual(os.stat(file1), os.stat("link"))
+
+            # Check os.stat calls from a dir below the link
+            os.chdir(level1)
+            self.assertEqual(os.stat(file1),
+                             os.stat(os.path.relpath(link)))
+
+            # Check os.stat calls from a dir above the link
+            os.chdir(level3)
+            self.assertEqual(os.stat(file1),
+                             os.stat(os.path.relpath(link)))
+        finally:
+            os.chdir(orig_dir)
+
+    @unittest.skipUnless(os.path.lexists(r'C:\Users\All Users')
+                            and os.path.exists(r'C:\ProgramData'),
+                            'Test directories not found')
+    def test_29248(self):
+        # os.symlink() calls CreateSymbolicLink, which creates
+        # the reparse data buffer with the print name stored
+        # first, so the offset is always 0. CreateSymbolicLink
+        # stores the "PrintName" DOS path (e.g. "C:\") first,
+        # with an offset of 0, followed by the "SubstituteName"
+        # NT path (e.g. "\??\C:\"). The "All Users" link, on
+        # the other hand, seems to have been created manually
+        # with an inverted order.
+        target = os.readlink(r'C:\Users\All Users')
+        self.assertTrue(os.path.samefile(target, r'C:\ProgramData'))
+
+    def test_buffer_overflow(self):
+        # Older versions would have a buffer overflow when detecting
+        # whether a link source was a directory. This test ensures we
+        # no longer crash, but does not otherwise validate the behavior
+        segment = 'X' * 27
+        path = os.path.join(*[segment] * 10)
+        test_cases = [
+            # overflow with absolute src
+            ('\\' + path, segment),
+            # overflow dest with relative src
+            (segment, path),
+            # overflow when joining src
+            (path[:180], path[:180]),
+        ]
+        for src, dest in test_cases:
+            try:
+                os.symlink(src, dest)
+            except FileNotFoundError:
+                pass
+            else:
+                try:
+                    os.remove(dest)
+                except OSError:
+                    pass
+            # Also test with bytes, since that is a separate code path.
+            try:
+                os.symlink(os.fsencode(src), os.fsencode(dest))
+            except FileNotFoundError:
+                pass
+            else:
+                try:
+                    os.remove(dest)
+                except OSError:
+                    pass
+
+    def test_appexeclink(self):
+        root = os.path.expandvars(r'%LOCALAPPDATA%\Microsoft\WindowsApps')
+        if not os.path.isdir(root):
+            self.skipTest("test requires a WindowsApps directory")
+
+        aliases = [os.path.join(root, a)
+                   for a in fnmatch.filter(os.listdir(root), '*.exe')]
+
+        for alias in aliases:
+            if support.verbose:
+                print()
+                print("Testing with", alias)
+            st = os.lstat(alias)
+            self.assertEqual(st, os.stat(alias))
+            self.assertFalse(stat.S_ISLNK(st.st_mode))
+            self.assertEqual(st.st_reparse_tag, stat.IO_REPARSE_TAG_APPEXECLINK)
+            self.assertTrue(os.path.isfile(alias))
+            # testing the first one we see is sufficient
+            break
+        else:
+            self.skipTest("test requires an app execution alias")
+
+
+class Win32JunctionTests(unittest.TestCase):
+    junction = 'junctiontest'
+    junction_target = os.path.dirname(os.path.abspath(__file__))
+
+    def setUp(self):
+        assert os.path.exists(self.junction_target)
+        assert not os.path.lexists(self.junction)
+
+    def tearDown(self):
+        if os.path.lexists(self.junction):
+            os.unlink(self.junction)
+
+    def test_create_junction(self):
+        _winapi.CreateJunction(self.junction_target, self.junction)
+        self.assertTrue(os.path.lexists(self.junction))
+        self.assertTrue(os.path.exists(self.junction))
+        self.assertTrue(os.path.isdir(self.junction))
+        self.assertNotEqual(os.stat(self.junction), os.lstat(self.junction))
+        self.assertEqual(os.stat(self.junction), os.stat(self.junction_target))
+
+        # bpo-37834: Junctions are not recognized as links.
+        self.assertFalse(os.path.islink(self.junction))
+        self.assertEqual(os.path.normcase("\\\\?\\" + self.junction_target),
+                         os.path.normcase(os.readlink(self.junction)))
+
+    def test_unlink_removes_junction(self):
+        _winapi.CreateJunction(self.junction_target, self.junction)
+        self.assertTrue(os.path.exists(self.junction))
+        self.assertTrue(os.path.lexists(self.junction))
+
+        os.unlink(self.junction)
+        self.assertFalse(os.path.exists(self.junction))
+
+
+class Win32NtTests(unittest.TestCase):
+    def test_getfinalpathname_handles(self):
+        nt = import_helper.import_module('nt')
+        ctypes = import_helper.import_module('ctypes')
+        # Ruff false positive -- it thinks we're redefining `ctypes` here
+        import ctypes.wintypes  # noqa: F811
+
+        kernel = ctypes.WinDLL('Kernel32.dll', use_last_error=True)
+        kernel.GetCurrentProcess.restype = ctypes.wintypes.HANDLE
+
+        kernel.GetProcessHandleCount.restype = ctypes.wintypes.BOOL
+        kernel.GetProcessHandleCount.argtypes = (ctypes.wintypes.HANDLE,
+                                                 ctypes.wintypes.LPDWORD)
+
+        # This is a pseudo-handle that doesn't need to be closed
+        hproc = kernel.GetCurrentProcess()
+
+        handle_count = ctypes.wintypes.DWORD()
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        before_count = handle_count.value
+
+        # The first two test the error path, __file__ tests the success path
+        filenames = [
+            r'\\?\C:',
+            r'\\?\NUL',
+            r'\\?\CONIN',
+            __file__,
+        ]
+
+        for _ in range(10):
+            for name in filenames:
+                try:
+                    nt._getfinalpathname(name)
+                except Exception:
+                    # Failure is expected
+                    pass
+                try:
+                    os.stat(name)
+                except Exception:
+                    pass
+
+        ok = kernel.GetProcessHandleCount(hproc, ctypes.byref(handle_count))
+        self.assertEqual(1, ok)
+
+        handle_delta = handle_count.value - before_count
+
+        self.assertEqual(0, handle_delta)
+
+    @support.requires_subprocess()
+    def test_stat_unlink_race(self):
+        # bpo-46785: the implementation of os.stat() falls back to reading
+        # the parent directory if CreateFileW() fails with a permission
+        # error. If reading the parent directory fails because the file or
+        # directory are subsequently unlinked, or because the volume or
+        # share are no longer available, then the original permission error
+        # should not be restored.
+        filename =  os_helper.TESTFN
+        self.addCleanup(os_helper.unlink, filename)
+        deadline = time.time() + 5
+        command = textwrap.dedent("""\
+            import os
+            import sys
+            import time
+
+            filename = sys.argv[1]
+            deadline = float(sys.argv[2])
+
+            while time.time() < deadline:
+                try:
+                    with open(filename, "w") as f:
+                        pass
+                except OSError:
+                    pass
+                try:
+                    os.remove(filename)
+                except OSError:
+                    pass
+            """)
+
+        with subprocess.Popen([sys.executable, '-c', command, filename, str(deadline)]) as proc:
+            while time.time() < deadline:
+                try:
+                    os.stat(filename)
+                except FileNotFoundError as e:
+                    assert e.winerror == 2  # ERROR_FILE_NOT_FOUND
+            try:
+                proc.wait(1)
+            except subprocess.TimeoutExpired:
+                proc.terminate()
+
+    @support.requires_subprocess()
+    def test_stat_inaccessible_file(self):
+        filename = os_helper.TESTFN
+        ICACLS = os.path.expandvars(r"%SystemRoot%\System32\icacls.exe")
+
+        with open(filename, "wb") as f:
+            f.write(b'Test data')
+
+        stat1 = os.stat(filename)
+
+        try:
+            # Remove all permissions from the file
+            subprocess.check_output([ICACLS, filename, "/inheritance:r"],
+                                    stderr=subprocess.STDOUT)
+        except subprocess.CalledProcessError as ex:
+            if support.verbose:
+                print(ICACLS, filename, "/inheritance:r", "failed.")
+                print(ex.stdout.decode("oem", "replace").rstrip())
+            try:
+                os.unlink(filename)
+            except OSError:
+                pass
+            self.skipTest("Unable to create inaccessible file")
+
+        def cleanup():
+            # Give delete permission to the owner (us)
+            subprocess.check_output([ICACLS, filename, "/grant", "*WD:(D)"],
+                                    stderr=subprocess.STDOUT)
+            os.unlink(filename)
+
+        self.addCleanup(cleanup)
+
+        if support.verbose:
+            print("File:", filename)
+            print("stat with access:", stat1)
+
+        # First test - we shouldn't raise here, because we still have access to
+        # the directory and can extract enough information from its metadata.
+        stat2 = os.stat(filename)
+
+        if support.verbose:
+            print(" without access:", stat2)
+
+        # We may not get st_dev/st_ino, so ensure those are 0 or match
+        self.assertIn(stat2.st_dev, (0, stat1.st_dev))
+        self.assertIn(stat2.st_ino, (0, stat1.st_ino))
+
+        # st_mode and st_size should match (for a normal file, at least)
+        self.assertEqual(stat1.st_mode, stat2.st_mode)
+        self.assertEqual(stat1.st_size, stat2.st_size)
+
+        # st_ctime and st_mtime should be the same
+        self.assertEqual(stat1.st_ctime, stat2.st_ctime)
+        self.assertEqual(stat1.st_mtime, stat2.st_mtime)
+
+        # st_atime should be the same or later
+        self.assertGreaterEqual(stat1.st_atime, stat2.st_atime)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Lib/test/test_os/utils.py b/Lib/test/test_os/utils.py
new file mode 100644
index 00000000000..e0c39598c31
--- /dev/null
+++ b/Lib/test/test_os/utils.py
@@ -0,0 +1,3 @@
+def create_file(filename, content=b'content'):
+    with open(filename, "xb", 0) as fp:
+        fp.write(content)
diff --git a/crates/host_env/src/crt_fd.rs b/crates/host_env/src/crt_fd.rs
index 5134b1dcf67..5b2cf685be1 100644
--- a/crates/host_env/src/crt_fd.rs
+++ b/crates/host_env/src/crt_fd.rs
@@ -319,7 +319,10 @@ impl AsRawFd for Borrowed<'_> {
 
 #[inline]
 fn ebadf() -> io::Error {
-    io::Error::from_raw_os_error(c::EBADF)
+    cfg_select! {
+        windows => crate::os::io_error_from_errno(c::EBADF),
+        _ => io::Error::from_raw_os_error(c::EBADF),
+    }
 }
 
 pub fn open(path: &ffi::CStr, flags: i32, mode: i32) -> io::Result<Owned> {
@@ -397,7 +400,12 @@ fn close_range_slow(low: Raw, high: Raw) {
 }
 
 pub fn ftruncate(fd: Borrowed<'_>, len: Offset) -> io::Result<()> {
-    let ret = unsafe { suppress_iph!(c::ftruncate(fd.as_raw(), len)) };
+    ftruncate_raw(fd.as_raw(), len)
+}
+
+/// Truncate a caller-supplied descriptor, letting the CRT validate its value.
+pub fn ftruncate_raw(fd: Raw, len: Offset) -> io::Result<()> {
+    let ret = unsafe { suppress_iph!(c::ftruncate(fd, len)) };
     // On Windows, _chsize_s returns 0 on success, or a positive error code (errno value) on failure.
     // On other platforms, ftruncate returns 0 on success, or -1 on failure with errno set.
     cfg_select! {
diff --git a/crates/host_env/src/posix.rs b/crates/host_env/src/posix.rs
index 053148ca4a4..8308108cd34 100644
--- a/crates/host_env/src/posix.rs
+++ b/crates/host_env/src/posix.rs
@@ -15,6 +15,10 @@ pub use super::posix_unix_like::*;
 
 pub use libc::{c_char, pid_t};
 
+/// `<sys/param.h>` on glibc defines NODEV as `(dev_t)-1`.
+#[cfg(all(target_os = "linux", target_env = "gnu"))]
+pub const NODEV: i32 = -1;
+
 /// `<sysexits.h>`. The `libc` crate does not bind these.
 pub const EX_OK: i32 = 0;
 pub const EX_USAGE: i32 = 64;
@@ -1148,13 +1152,22 @@ pub fn sync() {
     unsafe { libc::sync() };
 }
 
-pub fn getlogin() -> Option<CString> {
+pub fn getlogin() -> std::io::Result<CString> {
+    let old_errno = crate::os::get_errno();
+    crate::os::clear_errno();
     let ptr = unsafe { libc::getlogin() };
-    if ptr.is_null() {
-        None
+    let result = if ptr.is_null() {
+        let errno = crate::os::get_errno();
+        Err(if errno == 0 {
+            std::io::Error::other("unable to determine login name")
+        } else {
+            std::io::Error::from_raw_os_error(errno)
+        })
     } else {
-        Some(unsafe { CStr::from_ptr(ptr) }.to_owned())
-    }
+        Ok(unsafe { CStr::from_ptr(ptr) }.to_owned())
+    };
+    crate::os::set_errno(old_errno);
+    result
 }
 
 pub fn restore_signals() {
@@ -1945,26 +1958,71 @@ pub fn posix_spawn(config: PosixSpawnConfig<'_>) -> std::io::Result<libc::pid_t>
 
 #[cfg(target_os = "linux")]
 pub fn sendfile(
-    out_fd: BorrowedFd<'_>,
-    in_fd: BorrowedFd<'_>,
-    offset: &mut crate::crt_fd::Offset,
+    out_fd: i32,
+    in_fd: i32,
+    offset: Option<&mut crate::crt_fd::Offset>,
     count: usize,
 ) -> std::io::Result<usize> {
-    nix::sys::sendfile::sendfile(out_fd, in_fd, Some(offset), count).map_err(std::io::Error::from)
+    let offset = offset.map_or(core::ptr::null_mut(), |offset| offset as *mut _);
+    // The syscall validates raw descriptor numbers; no Rust fd lifetime is claimed.
+    let result = unsafe { libc::sendfile(out_fd, in_fd, offset, count) };
+    if result < 0 {
+        Err(std::io::Error::last_os_error())
+    } else {
+        Ok(result as usize)
+    }
 }
 
 #[cfg(target_os = "macos")]
 pub fn sendfile(
-    in_fd: BorrowedFd<'_>,
-    out_fd: BorrowedFd<'_>,
+    in_fd: i32,
+    out_fd: i32,
     offset: crate::crt_fd::Offset,
-    count: i64,
+    written: &mut i64,
     headers: Option<&[&[u8]]>,
     trailers: Option<&[&[u8]]>,
-) -> (std::io::Result<()>, i64) {
-    let (res, written) =
-        nix::sys::sendfile::sendfile(in_fd, out_fd, offset, Some(count), headers, trailers);
-    (res.map_err(std::io::Error::from), written)
+    flags: i32,
+) -> std::io::Result<()> {
+    let header_count = headers.map_or(0, |buffers| buffers.len());
+    let trailer_count = trailers.map_or(0, |buffers| buffers.len());
+    let (Ok(header_count), Ok(trailer_count)) =
+        (i32::try_from(header_count), i32::try_from(trailer_count))
+    else {
+        return Err(std::io::Error::from(std::io::ErrorKind::InvalidInput));
+    };
+    let iovecs = |buffers: Option<&[&[u8]]>| {
+        buffers
+            .unwrap_or_default()
+            .iter()
+            .map(|buffer| libc::iovec {
+                iov_base: buffer.as_ptr().cast_mut().cast(),
+                iov_len: buffer.len(),
+            })
+            .collect::<Vec<_>>()
+    };
+    let mut headers = iovecs(headers);
+    let mut trailers = iovecs(trailers);
+    let mut hdtr = libc::sf_hdtr {
+        headers: if headers.is_empty() {
+            core::ptr::null_mut()
+        } else {
+            headers.as_mut_ptr()
+        },
+        hdr_cnt: header_count,
+        trailers: if trailers.is_empty() {
+            core::ptr::null_mut()
+        } else {
+            trailers.as_mut_ptr()
+        },
+        trl_cnt: trailer_count,
+    };
+    // The slices and both iovec arrays remain live for this synchronous call.
+    let result = unsafe { libc::sendfile(in_fd, out_fd, offset, written, &mut hdtr, flags) };
+    if result < 0 {
+        Err(std::io::Error::last_os_error())
+    } else {
+        Ok(())
+    }
 }
 
 #[cfg(any(
diff --git a/crates/vm/src/stdlib/os.rs b/crates/vm/src/stdlib/os.rs
index 1e846d9fa61..7a51f81ae0f 100644
--- a/crates/vm/src/stdlib/os.rs
+++ b/crates/vm/src/stdlib/os.rs
@@ -209,16 +209,17 @@ pub(super) mod _os {
     #[cfg(any(unix, windows))]
     use crate::utils::ToCString;
     use crate::{
-        AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject,
+        AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult,
         builtins::{
-            PyBytesRef, PyFloat, PyGenericAlias, PyIntRef, PyStrRef, PyTuple, PyTupleRef, PyTypeRef,
+            PyBytes, PyBytesRef, PyFloat, PyGenericAlias, PyIntRef, PyStr, PyStrRef, PyTuple,
+            PyTupleRef, PyTypeRef,
         },
         class::PyClassDef,
         common::lock::{OnceCell, PyRwLock},
         convert::{IntoPyException, ToPyException, ToPyObject},
         exceptions::{OSErrorBuilder, ToOSErrorBuilder},
         function::{
-            ArgBytesLike, ArgMemoryBuffer, ArgumentError, FromArgs, FsPath, FuncArgs, Param,
+            ArgBytesLike, ArgMemoryBuffer, ArgumentError, Either, FromArgs, FsPath, FuncArgs, Param,
         },
         host_env::crt_fd,
         ospath::{OsPath, OsPathOrFd, OutputMode, PathConverter},
@@ -1911,41 +1912,35 @@ pub(super) mod _os {
     }
 
     fn file_time(obj: &PyObject, vm: &VirtualMachine) -> PyResult<FileTime> {
-        if let Some(float) = obj.downcast_ref::<PyFloat>() {
-            let value = float.to_f64();
-            if value.is_nan() {
-                return Err(vm.new_value_error("Invalid value NaN (not a number)"));
-            }
-            let mut seconds = value.trunc();
-            let mut nanoseconds = (value.fract() * 1_000_000_000.0).floor();
-            if nanoseconds < 0.0 {
-                nanoseconds += 1_000_000_000.0;
-                seconds -= 1.0;
-            }
-            let minimum = FileTimeSeconds::MIN as f64;
-            // The signed maximum rounds up when represented by f64.
-            if !(minimum..-minimum).contains(&seconds) {
-                return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
-            }
-            Ok(FileTime {
-                seconds: seconds as FileTimeSeconds,
-                nanoseconds: nanoseconds as libc::c_long,
-            })
-        } else {
-            let seconds = match file_time_seconds(obj, vm) {
-                Err(err) if err.fast_isinstance(vm.ctx.exceptions.type_error) => {
-                    return Err(vm.new_type_error(format!(
-                        "argument must be int or float, not {}",
-                        obj.class().fully_qualified_name(vm)?
-                    )));
-                }
-                result => result?,
-            };
-            Ok(FileTime {
-                seconds,
+        if obj.number().is_index() {
+            return Ok(FileTime {
+                seconds: file_time_seconds(obj, vm)?,
                 nanoseconds: 0,
-            })
+            });
         }
+        let value = if let Some(float) = obj.downcast_ref::<PyFloat>() {
+            float.to_f64()
+        } else {
+            obj.try_float(vm)?.to_f64()
+        };
+        if value.is_nan() {
+            return Err(vm.new_value_error("Invalid value NaN (not a number)"));
+        }
+        let mut seconds = value.trunc();
+        let mut nanoseconds = (value.fract() * 1_000_000_000.0).floor();
+        if nanoseconds < 0.0 {
+            nanoseconds += 1_000_000_000.0;
+            seconds -= 1.0;
+        }
+        let minimum = FileTimeSeconds::MIN as f64;
+        // The signed maximum rounds up when represented by f64.
+        if !(minimum..-minimum).contains(&seconds) {
+            return Err(vm.new_overflow_error("timestamp out of range for platform time_t"));
+        }
+        Ok(FileTime {
+            seconds: seconds as FileTimeSeconds,
+            nanoseconds: nanoseconds as libc::c_long,
+        })
     }
 
     #[pyfunction]
@@ -2183,27 +2178,66 @@ pub(super) mod _os {
         crate::host_env::time::strerror(code)
     }
 
-    #[pyfunction]
     pub(crate) fn ftruncate(fd: crt_fd::Borrowed<'_>, length: crt_fd::Offset) -> io::Result<()> {
         crt_fd::ftruncate(fd, length)
     }
 
+    #[pyfunction(name = "ftruncate")]
+    fn py_ftruncate(fd: i32, length: crt_fd::Offset, vm: &VirtualMachine) -> PyResult<()> {
+        ftruncate_impl(fd, length, vm)
+    }
+
+    fn ftruncate_impl(
+        fd: crt_fd::Raw,
+        length: crt_fd::Offset,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
+        use crate::host_env::os::ErrorExt;
+
+        vm.audit("os.truncate", || (fd, length))?;
+        loop {
+            match vm.allow_threads(|| crt_fd::ftruncate_raw(fd, length)) {
+                Ok(()) => return Ok(()),
+                Err(err) if err.posix_errno() == libc::EINTR => vm.check_signals()?,
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
+        }
+    }
+
     #[derive(FromArgs)]
     struct TruncateArgs {
         #[pyarg(any)]
         path: PyObjectRef,
         #[pyarg(any)]
-        length: crt_fd::Offset,
+        length: PyObjectRef,
     }
 
     #[pyfunction]
     fn truncate(args: TruncateArgs, vm: &VirtualMachine) -> PyResult<()> {
         let TruncateArgs { path, length } = args;
-        match path.clone().try_into_value::<crt_fd::Borrowed<'_>>(vm) {
-            Ok(fd) => return ftruncate(fd, length).map_err(|e| e.into_pyexception(vm)),
-            Err(e) if e.fast_isinstance(vm.ctx.exceptions.warning) => return Err(e),
-            Err(_) => {}
-        }
+        let index =
+            if path.downcast_ref::<PyStr>().is_some() || path.downcast_ref::<PyBytes>().is_some() {
+                None
+            } else {
+                path.try_index_opt(vm)
+            };
+        let path = if let Some(index) = index {
+            super::warn_if_bool_fd(&path, vm)?;
+            Either::A(index?.try_to_primitive::<crt_fd::Raw>(vm)?)
+        } else {
+            Either::B(
+                PathConverter::new()
+                    .function("truncate")
+                    .try_path_inner(path, true, vm)?,
+            )
+        };
+        let length = length.try_into_value::<crt_fd::Offset>(vm)?;
+        let path = match path {
+            Either::A(fd) => {
+                return ftruncate_impl(fd, length, vm);
+            }
+            Either::B(path) => path,
+        };
 
         #[cold]
         fn error(
@@ -2214,7 +2248,8 @@ pub(super) mod _os {
             OSErrorBuilder::with_filename(&error, path, vm)
         }
 
-        let path = OsPath::try_from_object(vm, path)?;
+        vm.audit("os.truncate", || (path.filename(vm), length))?;
+
         // TODO: just call libc::truncate() on POSIX
         let f = match crate::host_env::fs::open_write(&path) {
             Ok(f) => f,
@@ -2439,6 +2474,7 @@ pub(super) mod _os {
             SupportFunc::new("rmdir", Some(false), Some(RMDIR_DIR_FD), Some(false)),
             SupportFunc::new("scandir", Some(SCANDIR_FD), Some(false), Some(false)),
             SupportFunc::new("stat", Some(true), Some(STAT_DIR_FD), Some(true)),
+            SupportFunc::new("lstat", Some(false), Some(STAT_DIR_FD), Some(false)),
             SupportFunc::new("fstat", Some(false), Some(false), Some(false)),
             SupportFunc::new("symlink", Some(false), Some(SYMLINK_DIR_FD), Some(false)),
             SupportFunc::new("truncate", Some(true), Some(false), Some(false)),
diff --git a/crates/vm/src/stdlib/posix.rs b/crates/vm/src/stdlib/posix.rs
index 26371c87e0a..ae70b7001bb 100644
--- a/crates/vm/src/stdlib/posix.rs
+++ b/crates/vm/src/stdlib/posix.rs
@@ -49,6 +49,10 @@ pub mod module {
     #[pyattr]
     use rustpython_host_env::posix::PIDFD_NONBLOCK;
 
+    #[cfg(all(target_os = "linux", target_env = "gnu"))]
+    #[pyattr]
+    use rustpython_host_env::posix::NODEV;
+
     #[cfg(target_os = "macos")]
     #[pyattr]
     use rustpython_host_env::posix::{
@@ -881,7 +885,11 @@ pub mod module {
 
     /// Warn if forking from a multi-threaded process.
     /// `num_os_threads` should be captured before parent after-fork hooks run.
-    fn warn_if_multi_threaded(name: &str, num_os_threads: isize, vm: &VirtualMachine) {
+    fn warn_if_multi_threaded(
+        name: &str,
+        num_os_threads: isize,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
         let num_threads = if num_os_threads > 0 {
             num_os_threads as usize
         } else {
@@ -895,7 +903,7 @@ pub mod module {
                 .and_then(|m| m.get_item("threading", vm))
             {
                 Ok(m) => m,
-                Err(_) => return,
+                Err(_) => return Ok(()),
             };
             let active = threading.get_attr("_active", vm).ok();
             let limbo = threading.get_attr("_limbo", vm).ok();
@@ -920,10 +928,9 @@ pub mod module {
             );
 
             // Match PyErr_WarnFormat(..., stacklevel=1) in CPython.
-            // Best effort: ignore failures like CPython does in this path.
-            let _ =
-                crate::stdlib::_warnings::warn(vm.ctx.exceptions.deprecation_warning, msg, 1, vm);
+            crate::stdlib::_warnings::warn(vm.ctx.exceptions.deprecation_warning, msg, 1, vm)?;
         }
+        Ok(())
     }
 
     #[pyfunction]
@@ -960,7 +967,7 @@ pub mod module {
                 let num_os_threads = get_number_of_os_threads();
                 py_os_after_fork_parent(vm);
                 // Warn only after parent callback path resumes the world.
-                warn_if_multi_threaded("fork", num_os_threads, vm);
+                warn_if_multi_threaded("fork", num_os_threads, vm)?;
                 Ok(pid)
             }
             Err(err) => {
@@ -1002,7 +1009,7 @@ pub mod module {
             Ok((pid, master)) => {
                 let num_os_threads = get_number_of_os_threads();
                 py_os_after_fork_parent(vm);
-                warn_if_multi_threaded("forkpty", num_os_threads, vm);
+                warn_if_multi_threaded("forkpty", num_os_threads, vm)?;
                 Ok((pid, master))
             }
             Err(err) => {
@@ -2347,23 +2354,10 @@ pub mod module {
     }
 
     #[pyfunction]
-    fn getlogin(vm: &VirtualMachine) -> PyResult<String> {
-        // Get a pointer to the login name string. The string is statically
-        // allocated and might be overwritten on subsequent calls to this
-        // function or to `cuserid()`. See man getlogin(3) for more information.
-        let Some(login) = rustpython_host_env::posix::getlogin() else {
-            return Err(vm.new_os_error("unable to determine login name"));
-        };
-        login.to_str().map(|s| s.to_owned()).map_err(|e| {
-            vm.new_unicode_decode_error(
-                vm.ctx.new_str("utf-8"),
-                vm.ctx.new_bytes(login.as_bytes().to_vec()),
-                e.valid_up_to(),
-                e.error_len()
-                    .map_or(login.as_bytes().len(), |n| e.valid_up_to() + n),
-                vm.ctx.new_str("unable to decode login name"),
-            )
-        })
+    fn getlogin(vm: &VirtualMachine) -> PyResult<crate::builtins::PyStrRef> {
+        let login =
+            rustpython_host_env::posix::getlogin().map_err(|err| err.into_pyexception(vm))?;
+        Ok(vm.fsdecode(std::ffi::OsString::from_vec(login.into_bytes())))
     }
 
     // cfg from nix
@@ -2868,10 +2862,16 @@ pub mod module {
 
     #[cfg(any(target_os = "linux", target_os = "macos"))]
     #[derive(FromArgs)]
-    struct SendFileArgs<'fd> {
-        out_fd: BorrowedFd<'fd>,
-        in_fd: BorrowedFd<'fd>,
+    struct SendFileArgs {
+        out_fd: i32,
+        in_fd: i32,
+        #[cfg(target_os = "linux")]
+        offset: PyObjectRef,
+        #[cfg(target_os = "macos")]
         offset: rustpython_host_env::crt_fd::Offset,
+        #[cfg(target_os = "linux")]
+        count: isize,
+        #[cfg(target_os = "macos")]
         count: i64,
         #[cfg(target_os = "macos")]
         // Missing means an empty header list.
@@ -2882,35 +2882,58 @@ pub mod module {
         #[pyarg(any, optional, py_default = "()")]
         trailers: OptionalArg<PyObjectRef>,
         #[cfg(target_os = "macos")]
-        #[allow(dead_code)]
         #[pyarg(any, default)]
-        // TODO: not implemented
         flags: i32,
     }
 
     #[cfg(target_os = "linux")]
     #[pyfunction]
-    fn sendfile(args: SendFileArgs<'_>, vm: &VirtualMachine) -> PyResult {
-        let mut file_offset = args.offset;
+    fn sendfile(args: SendFileArgs, vm: &VirtualMachine) -> PyResult {
+        if args.count < 0 {
+            return Err(vm.new_value_error("count cannot be negative"));
+        }
+        let mut file_offset = if vm.is_none(&args.offset) {
+            None
+        } else {
+            Some(args.offset.try_into_value(vm)?)
+        };
 
-        let res = rustpython_host_env::posix::sendfile(
-            args.out_fd,
-            args.in_fd,
-            &mut file_offset,
-            args.count as usize,
-        )
-        .map_err(|err| err.into_pyexception(vm))?;
-        Ok(vm.ctx.new_int(res as u64).into())
+        loop {
+            let result = vm.allow_threads(|| {
+                rustpython_host_env::posix::sendfile(
+                    args.out_fd,
+                    args.in_fd,
+                    file_offset.as_mut(),
+                    args.count as usize,
+                )
+            });
+            match result {
+                Ok(written) => return Ok(vm.ctx.new_int(written).into()),
+                Err(err) if err.raw_os_error() == Some(libc::EINTR) => vm.check_signals()?,
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
+        }
     }
 
     #[cfg(target_os = "macos")]
     fn _extract_vec_bytes(
         x: OptionalArg,
+        kind: &str,
         vm: &VirtualMachine,
     ) -> PyResult<Option<Vec<crate::function::ArgBytesLike>>> {
         x.into_option()
             .map(|x| {
-                let v: Vec<crate::function::ArgBytesLike> = x.try_to_value(vm)?;
+                let sequence = x.try_sequence(vm).map_err(|_| {
+                    vm.new_type_error(format!("sendfile() {kind}s must be a sequence"))
+                })?;
+                let count = sequence.length(vm)?;
+                if count > i32::MAX as usize {
+                    return Err(vm.new_overflow_error(format!("sendfile() {kind} is too large")));
+                }
+                let mut v = Vec::with_capacity(count);
+                for index in 0..count {
+                    v.push(sequence.get_item(index as isize, vm)?.try_into_value(vm)?);
+                }
                 Ok(if v.is_empty() { None } else { Some(v) })
             })
             .transpose()
@@ -2919,47 +2942,78 @@ pub mod module {
 
     #[cfg(target_os = "macos")]
     #[pyfunction]
-    fn sendfile(args: SendFileArgs<'_>, vm: &VirtualMachine) -> PyResult {
-        let headers = _extract_vec_bytes(args.headers, vm)?;
-        let count = headers
-            .as_ref()
-            .map_or(0, |v| v.iter().map(|s| s.len()).sum()) as i64
-            + args.count;
+    fn sendfile(args: SendFileArgs, vm: &VirtualMachine) -> PyResult {
+        if args.count < 0 {
+            return Err(vm.new_value_error("count cannot be negative"));
+        }
+        let headers = _extract_vec_bytes(args.headers, "header", vm)?;
+        let mut count = args.count;
+        if let Some(headers) = &headers {
+            for header in headers {
+                let length = i64::try_from(header.len())
+                    .map_err(|_| vm.new_overflow_error("sendfile() header is too large"))?;
+                if count >= i64::MAX - length {
+                    return Err(vm.new_overflow_error("sendfile() header is too large"));
+                }
+                count += length;
+            }
+        }
 
+        let trailers = _extract_vec_bytes(args.trailers, "trailer", vm)?;
+        // Complete all Python argument callbacks before snapshotting mutable
+        // exporters. No buffer lock may survive another callback or blocking I/O.
         let headers = headers
             .as_ref()
-            .map(|v| v.iter().map(|b| b.borrow_buf()).collect::<Vec<_>>());
+            .map(|v| {
+                v.iter()
+                    .map(|b| b.borrow_buf_unlocked(vm))
+                    .collect::<PyResult<Vec<_>>>()
+            })
+            .transpose()?;
         let headers = headers
             .as_ref()
             .map(|v| v.iter().map(|borrowed| &**borrowed).collect::<Vec<_>>());
         let headers = headers.as_deref();
 
-        let trailers = _extract_vec_bytes(args.trailers, vm)?;
         let trailers = trailers
             .as_ref()
-            .map(|v| v.iter().map(|b| b.borrow_buf()).collect::<Vec<_>>());
+            .map(|v| {
+                v.iter()
+                    .map(|b| b.borrow_buf_unlocked(vm))
+                    .collect::<PyResult<Vec<_>>>()
+            })
+            .transpose()?;
         let trailers = trailers
             .as_ref()
             .map(|v| v.iter().map(|borrowed| &**borrowed).collect::<Vec<_>>());
         let trailers = trailers.as_deref();
 
-        let (res, written) = rustpython_host_env::posix::sendfile(
-            args.in_fd,
-            args.out_fd,
-            args.offset,
-            count,
-            headers,
-            trailers,
-        );
-        // On macOS, sendfile can return EAGAIN even when some bytes were written.
-        // In that case, we should return the number of bytes written rather than
-        // raising an exception. Only raise an error if no bytes were written.
-        if let Err(err) = res
-            && written == 0
-        {
-            return Err(err.into_pyexception(vm));
+        let mut written = count;
+        loop {
+            let result = vm.allow_threads(|| {
+                rustpython_host_env::posix::sendfile(
+                    args.in_fd,
+                    args.out_fd,
+                    args.offset,
+                    &mut written,
+                    headers,
+                    trailers,
+                    args.flags,
+                )
+            });
+            match result {
+                Ok(()) => return Ok(vm.ctx.new_int(written).into()),
+                Err(err) if err.raw_os_error() == Some(libc::EINTR) => vm.check_signals()?,
+                // CPython permits partial success only for EAGAIN or EBUSY.
+                Err(err)
+                    if written != 0
+                        && matches!(err.raw_os_error(), Some(libc::EAGAIN | libc::EBUSY)) =>
+                {
+                    return Ok(vm.ctx.new_int(written).into());
+                }
+                Err(err) => return Err(err.into_pyexception(vm)),
+            }
         }
-        Ok(vm.ctx.new_int(written as u64).into())
     }
 
     #[cfg(target_os = "linux")]
diff --git a/crates/vm/src/stdlib/posix_compat.rs b/crates/vm/src/stdlib/posix_compat.rs
index 4ac80a03d37..ad1709c99c0 100644
--- a/crates/vm/src/stdlib/posix_compat.rs
+++ b/crates/vm/src/stdlib/posix_compat.rs
@@ -10,6 +10,8 @@ pub(crate) use module::module_def;
     super::posix_unix_like::_posix_unix_like,
 ))]
 pub(crate) mod module {
+    #[cfg(target_os = "wasi")]
+    use crate::{AsObject, function::OptionalArg, ospath::PathConverter};
     use crate::{
         Py, PyObjectRef, PyResult, VirtualMachine,
         builtins::PyStrRef,
@@ -17,6 +19,64 @@ pub(crate) mod module {
         stdlib::os::{_os, DirFd, SupportFunc, SymlinkArgs, TargetIsDirectory},
     };
 
+    #[cfg(target_os = "wasi")]
+    #[derive(FromArgs)]
+    struct ChmodArgs {
+        #[pyarg(any)]
+        path: PyObjectRef,
+        #[pyarg(any)]
+        mode: PyObjectRef,
+        #[pyarg(named, default, py_default = "None")]
+        dir_fd: Option<PyObjectRef>,
+        #[pyarg(named, optional, py_default = "(os.name != 'nt')")]
+        follow_symlinks: OptionalArg<PyObjectRef>,
+    }
+
+    #[cfg(target_os = "wasi")]
+    #[pyfunction]
+    fn chmod(args: ChmodArgs, vm: &VirtualMachine) -> PyResult<()> {
+        // Keep the clinic conversion order even though WASI has no chmod syscall.
+        let path = PathConverter::new()
+            .function("chmod")
+            .argument("path")
+            .try_path(args.path, vm)?;
+        let mode = args.mode.try_index(vm)?;
+        let mode = i32::try_from(mode.as_bigint())
+            .map_err(|_| vm.new_overflow_error("Python int too large to convert to C int"))?;
+        if let Some(dir_fd) = args.dir_fd {
+            super::super::os::warn_if_bool_fd(&dir_fd, vm)?;
+            let fd = dir_fd.try_index_opt(vm).unwrap_or_else(|| {
+                Err(vm.new_type_error(format!(
+                    "argument should be integer or None, not {}",
+                    dir_fd.class().name()
+                )))
+            })?;
+            let fd = i32::try_from(fd.as_bigint()).map_err(|_| {
+                vm.new_overflow_error(if fd.as_bigint().sign() == malachite_bigint::Sign::Minus {
+                    "fd is less than minimum"
+                } else {
+                    "fd is greater than maximum"
+                })
+            })?;
+            if fd != rustpython_host_env::os::AT_FDCWD {
+                return Err(vm.new_not_implemented_error("dir_fd unavailable on this platform"));
+            }
+        }
+        let follow_symlinks = match args.follow_symlinks {
+            OptionalArg::Present(value) => value.try_to_bool(vm)?,
+            OptionalArg::Missing => true,
+        };
+        if !follow_symlinks {
+            return Err(
+                vm.new_not_implemented_error("chmod: follow_symlinks unavailable on this platform")
+            );
+        }
+        vm.audit("os.chmod", || (path.filename(vm), mode, -1))?;
+        // CPython 3.15 posixmodule.c explicitly ignores the absent WASI syscall
+        // after argument conversion and auditing. No supports_* capability is set.
+        Ok(())
+    }
+
     #[derive(FromArgs)]
     struct AccessArgs<'a> {
         #[pyarg(any)]
diff --git a/scripts/update_lib/cmd_migrate.py b/scripts/update_lib/cmd_migrate.py
index 5e4c2656a9c..6ec271851e8 100644
--- a/scripts/update_lib/cmd_migrate.py
+++ b/scripts/update_lib/cmd_migrate.py
@@ -20,7 +20,9 @@
 
 import argparse
 import ast
+import builtins
 import collections
+import hashlib
 import json
 import pathlib
 import shutil
@@ -186,6 +188,244 @@ def _verify_call(contents: str, expected: str) -> None:
         raise ValueError("Ambiguous environment-reset call text")
 
 
+def copy_support_assignments(source, destination, names, exports):
+    """Copy contiguous canonical assignments and export lines, failing closed.
+
+    Only additions are supported. Dependencies must already be imported with the
+    same bindings, or be builtins/earlier selected assignments. The next unchanged
+    top-level statement is the insertion anchor; unrelated bytes stay untouched.
+    """
+    if (
+        not names
+        or len(set(names)) != len(names)
+        or not all(name.isidentifier() for name in names)
+    ):
+        raise ValueError("Assignment names must be unique identifiers")
+    if len(set(exports)) != len(exports) or not set(exports) <= set(names):
+        raise ValueError("Exports must be unique selected assignment names")
+    src, dst = ast.parse(source), ast.parse(destination)
+
+    def assignment(tree, name):
+        matches = [
+            node
+            for node in ast.walk(tree)
+            if isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Store)
+            and node.id == name
+        ]
+        nodes = [
+            node
+            for node in tree.body
+            if isinstance(node, ast.Assign)
+            and len(node.targets) == 1
+            and isinstance(node.targets[0], ast.Name)
+            and node.targets[0].id == name
+        ]
+        rebound = any(
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
+            and node.name == name
+            or isinstance(node, ast.alias)
+            and (node.asname or node.name.split(".")[0]) == name
+            for node in ast.walk(tree)
+        )
+        if len(nodes) != 1 or matches != [nodes[0].targets[0]] or rebound:
+            raise ValueError(f"Expected one top-level assignment to {name}")
+        return nodes[0]
+
+    selected = [assignment(src, name) for name in names]
+    start = src.body.index(selected[0])
+    if src.body[start : start + len(names)] != selected or start + len(names) == len(
+        src.body
+    ):
+        raise ValueError("Assignments must be contiguous and followed by an anchor")
+    if (start and src.body[start - 1].end_lineno >= selected[0].lineno) or (
+        selected[-1].end_lineno >= src.body[start + len(names)].lineno
+    ):
+        raise ValueError("Selected assignments must occupy complete source lines")
+    for node in ast.walk(dst):
+        if (
+            isinstance(node, ast.Name)
+            and isinstance(node.ctx, ast.Store)
+            and node.id in names
+            or isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))
+            and node.name in names
+            or isinstance(node, ast.alias)
+            and (node.asname or node.name.split(".")[0]) in names
+        ):
+            raise ValueError("A selected assignment already exists in the destination")
+
+    anchor = src.body[start + len(names)]
+    matches = [node for node in dst.body if ast.dump(node) == ast.dump(anchor)]
+    if len(matches) != 1:
+        raise ValueError("Expected one unchanged destination anchor")
+    for tree, node in ((src, anchor), (dst, matches[0])):
+        index = tree.body.index(node)
+        if (
+            getattr(node, "decorator_list", ())
+            or node.col_offset
+            or index
+            and tree.body[index - 1].end_lineno >= node.lineno
+            or index + 1 < len(tree.body)
+            and tree.body[index + 1].lineno <= node.end_lineno
+        ):
+            raise ValueError(
+                "Anchor must be undecorated and occupy complete source lines"
+            )
+    available = set(vars(builtins))
+    imports = (set(), set())
+    for node in src.body:
+        matching = [
+            other
+            for other in dst.body
+            if ast.dump(node) == ast.dump(other) and other.lineno < matches[0].lineno
+        ]
+        if (
+            isinstance(node, (ast.Import, ast.ImportFrom))
+            and node.lineno < selected[0].lineno
+            and len(matching) == 1
+        ):
+            imports[0].update(node.names)
+            imports[1].update(matching[0].names)
+            available.update(
+                alias.asname or alias.name.split(".")[0] for alias in node.names
+            )
+    dependencies = set()
+    for name, node in zip(names, selected):
+        used = {
+            child.id
+            for child in ast.walk(node.value)
+            if isinstance(child, ast.Name) and isinstance(child.ctx, ast.Load)
+        }
+        missing = used - available
+        if missing:
+            raise ValueError(f"Missing dependencies for {name}: {sorted(missing)}")
+        dependencies.update(used - set(names))
+        available.add(name)
+    for tree, boundary, allowed in (
+        (src, selected[0].lineno, imports[0]),
+        (dst, matches[0].lineno, imports[1]),
+    ):
+        if dependencies and any(
+            isinstance(node, ast.ImportFrom)
+            and node.lineno < boundary
+            and any(alias.name == "*" for alias in node.names)
+            for node in ast.walk(tree)
+        ):
+            raise ValueError("Wildcard import makes dependency bindings ambiguous")
+        for dependency in dependencies:
+            bindings = []
+            for node in ast.walk(tree):
+                if getattr(node, "lineno", boundary) >= boundary:
+                    continue
+                if isinstance(node, ast.Name) and isinstance(
+                    node.ctx, (ast.Store, ast.Del)
+                ):
+                    name = node.id
+                elif isinstance(node, ast.alias):
+                    name = node.asname or node.name.split(".")[0]
+                elif isinstance(
+                    node,
+                    (
+                        ast.FunctionDef,
+                        ast.AsyncFunctionDef,
+                        ast.ClassDef,
+                        ast.ExceptHandler,
+                        ast.MatchAs,
+                        ast.MatchStar,
+                    ),
+                ):
+                    name = node.name
+                elif isinstance(node, ast.arg):
+                    name = node.arg
+                elif isinstance(node, ast.MatchMapping):
+                    name = node.rest
+                else:
+                    continue
+                if name == dependency:
+                    bindings.append(node)
+            if len(bindings) > 1 or any(node not in allowed for node in bindings):
+                raise ValueError(f"Ambiguous dependency binding: {dependency}")
+
+    source_lines, lines = source.splitlines(True), destination.splitlines(True)
+    additions = [
+        (
+            matches[0].lineno - 1,
+            source_lines[selected[0].lineno - 1 : anchor.lineno - 1],
+        )
+    ]
+    if exports:
+        src_all, dst_all = assignment(src, "__all__"), assignment(dst, "__all__")
+        if not isinstance(src_all.value, ast.List) or not isinstance(
+            dst_all.value, ast.List
+        ):
+            raise ValueError("__all__ must be a literal list")
+        if any(
+            not isinstance(node, ast.Constant) or not isinstance(node.value, str)
+            for node in src_all.value.elts + dst_all.value.elts
+        ):
+            raise ValueError("__all__ entries must be literal strings")
+        src_values, dst_values = (
+            ast.literal_eval(src_all.value),
+            ast.literal_eval(dst_all.value),
+        )
+        if any(src_values.count(name) != 1 or name in dst_values for name in exports):
+            raise ValueError(
+                "Exports must occur once in canonical __all__ and be absent locally"
+            )
+        export_nodes = [node for node in src_all.value.elts if node.value in exports]
+        export_lines = sorted({node.lineno - 1 for node in export_nodes})
+        fragment = "".join(source_lines[line] for line in export_lines)
+        if ast.literal_eval("[\n" + fragment + "]") != exports:
+            raise ValueError(
+                "Canonical export lines include unselected entries or differ in order"
+            )
+        if lines[dst_all.lineno - 1].strip() != "__all__ = [":
+            raise ValueError("Expected a multiline destination __all__ list")
+        additions.append((dst_all.lineno, [fragment]))
+    for offset, fragment in sorted(additions, reverse=True):
+        lines[offset:offset] = fragment
+    result = "".join(lines)
+    ast.parse(result)
+    return result
+
+
+def migrate_support_assignments(path, names, exports, source_hash, destination_hash):
+    target = parse_lib_path(path)
+    if (
+        target.parts[:3] != ("Lib", "test", "support")
+        or ".." in target.parts
+        or target.suffix != ".py"
+    ):
+        raise ValueError(
+            "Selective assignments are restricted to Lib/test/support Python files"
+        )
+    source, destination = path.read_bytes(), target.read_bytes()
+    for label, contents, expected in (
+        ("source", source, source_hash),
+        ("destination", destination, destination_hash),
+    ):
+        if hashlib.sha256(contents).hexdigest() != expected:
+            raise ValueError(f"{label} SHA256 does not match the pinned input")
+    result = copy_support_assignments(
+        source.decode("utf-8"), destination.decode("utf-8"), names, exports
+    ).encode("utf-8")
+    if target.read_bytes() != destination:
+        raise ValueError("Destination changed during assignment preparation")
+    target.write_bytes(result)
+    print(
+        json.dumps(
+            {
+                "assignments": names,
+                "exports": exports,
+                "source_sha256": source_hash,
+                "destination_before_sha256": destination_hash,
+                "destination_after_sha256": hashlib.sha256(result).hexdigest(),
+            },
+            sort_keys=True,
+        )
+    )
+
+
 def patch_single_content(
     src_path: pathlib.Path,
     lib_path: pathlib.Path,
@@ -485,9 +725,35 @@ def main(argv: list[str] | None = None) -> int:
         type=pathlib.Path,
         help="Single-file target, remaining decorator patches, and optional doctests JSON",
     )
+    parser.add_argument(
+        "--support-assignments",
+        nargs="+",
+        help="Add selected canonical test.support assignments",
+    )
+    parser.add_argument("--support-exports", nargs="+", default=[])
+    parser.add_argument("--source-sha256")
+    parser.add_argument("--destination-sha256")
     args = parser.parse_args(argv)
 
     try:
+        if args.support_assignments:
+            if args.legacy_map is not None or args.patches is not None:
+                raise ValueError(
+                    "Selective support imports cannot combine with --legacy-map or --patches"
+                )
+            try:
+                migrate_support_assignments(
+                    args.path,
+                    args.support_assignments,
+                    args.support_exports,
+                    args.source_sha256,
+                    args.destination_sha256,
+                )
+            except (SyntaxError, UnicodeError) as exc:
+                raise ValueError(str(exc)) from exc
+            return 0
+        if args.support_exports or args.source_sha256 or args.destination_sha256:
+            raise ValueError("Selective options require --support-assignments")
         if args.patches is not None and (
             args.path.is_dir() or args.legacy_map is not None
         ):
diff --git a/scripts/update_lib/tests/test_migrate.py b/scripts/update_lib/tests/test_migrate.py
index 2c18a9e1a3c..033dab41c34 100644
--- a/scripts/update_lib/tests/test_migrate.py
+++ b/scripts/update_lib/tests/test_migrate.py
@@ -1,6 +1,7 @@
 """Tests for migrate.py - file migration operations."""
 
 import contextlib
+import hashlib
 import io
 import json
 import pathlib
@@ -16,6 +17,7 @@
     LOCAL_CALL,
     SYSCONFIG_LOCAL,
     SYSCONFIG_UPSTREAM,
+    copy_support_assignments,
     main,
     migrate_package,
     patch_directory,
@@ -35,6 +37,235 @@
 )
 
 
+class TestSupportAssignments(unittest.TestCase):
+    names = ["_available", "requires_available", "requires_unavailable"]
+    exports = names[1:]
+    imports = "import os\nimport unittest\n\n"
+    export_line = '    "requires_available", "requires_unavailable",\n'
+    block = (
+        "_available = hasattr(os, 'geteuid') and os.geteuid() == 0\n"
+        'requires_available = unittest.skipUnless(_available, "needs privilege")\n'
+        'requires_unavailable = unittest.skipIf(_available, "needs no privilege")\n\n\n'
+    )
+    source = (
+        imports
+        + '__all__ = [\n    "existing",\n'
+        + export_line
+        + "]\n\n\n"
+        + block
+        + "STATUS = 1\n"
+    )
+    destination = (
+        imports
+        + '__all__ = [\n    "existing",\n]\n\n\ndef legacy():\n    return "keep π"\n\n\nSTATUS = 1\n'
+    )
+
+    def test_canonical_bytes_and_unrelated_helpers_are_preserved(self):
+        result = copy_support_assignments(
+            self.source, self.destination, self.names, self.exports
+        )
+        self.assertEqual(
+            result,
+            self.destination.replace(
+                "__all__ = [\n", "__all__ = [\n" + self.export_line
+            ).replace("STATUS = 1\n", self.block + "STATUS = 1\n"),
+        )
+
+    def test_missing_ambiguous_and_noncontiguous_assignments_are_rejected(self):
+        for source in (
+            self.source.replace("_available =", "_other ="),
+            self.source + "_available = False\n",
+            self.source + "def requires_available(): pass\n",
+            self.source + "import other as requires_unavailable\n",
+            self.source.replace(
+                "requires_available =", "unselected = 0\nrequires_available ="
+            ),
+            self.source.replace("_available =", "unselected = 0; _available =", 1),
+            self.source.replace("STATUS = 1", ""),
+        ):
+            with self.subTest(source=source), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    source, self.destination, self.names, self.exports
+                )
+
+    def test_existing_bindings_dependencies_and_anchor_fail_closed(self):
+        for destination in (
+            self.destination + "_available = False\n",
+            self.destination + "def requires_available(): pass\n",
+            self.destination + "import other as requires_unavailable\n",
+            self.destination.replace("import os\n", ""),
+            self.destination.replace("import os\n", "import os as other\n"),
+            self.destination.replace("import os\n", "") + "import os\n",
+            self.destination.replace("STATUS = 1", "STATUS = 2"),
+            self.destination + "STATUS = 1\n",
+        ):
+            with self.subTest(destination=destination), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    self.source, destination, self.names, self.exports
+                )
+
+    def test_invalid_exports_are_rejected(self):
+        for source, destination, exports in (
+            (self.source, self.destination, self.exports + self.exports),
+            (self.source, self.destination, ["unselected"]),
+            (self.source, self.destination, list(reversed(self.exports))),
+            (self.source, self.destination, self.exports[:1]),
+            (
+                self.source,
+                self.destination.replace('"existing"', '"requires_available"'),
+                self.exports,
+            ),
+            (self.source.replace(self.export_line, ""), self.destination, self.exports),
+            (
+                self.source,
+                self.destination.replace('[\n    "existing",\n]', '("existing",)'),
+                self.exports,
+            ),
+            (self.source.replace('"existing"', "3"), self.destination, self.exports),
+            (self.source, self.destination.replace('"existing"', "3"), self.exports),
+        ):
+            with self.subTest(exports=exports), self.assertRaises(ValueError):
+                copy_support_assignments(source, destination, self.names, exports)
+
+    def test_dependency_rebinding_deletion_and_builtin_shadowing_are_rejected(self):
+        for binding in (
+            "os = replacement",
+            "del os",
+            "import other as os",
+            "hasattr = replacement",
+            "del hasattr",
+            "if condition:\n    unittest = replacement",
+        ):
+            for source_side in (False, True):
+                source, destination = self.source, self.destination
+                if source_side:
+                    source = source.replace("__all__ = [", binding + "\n__all__ = [")
+                else:
+                    destination = destination.replace(
+                        "__all__ = [", binding + "\n__all__ = ["
+                    )
+                with (
+                    self.subTest(binding=binding, source_side=source_side),
+                    self.assertRaisesRegex(ValueError, "Ambiguous dependency binding"),
+                ):
+                    copy_support_assignments(
+                        source, destination, self.names, self.exports
+                    )
+
+    def test_decorated_and_partial_line_anchors_are_rejected(self):
+        for replacement in (
+            "@decorator\ndef anchor(): pass",
+            "STATUS = 1; other = 2",
+            "other = 2; STATUS = 1",
+        ):
+            with self.subTest(replacement=replacement), self.assertRaises(ValueError):
+                copy_support_assignments(
+                    self.source.replace("STATUS = 1", replacement),
+                    self.destination.replace("STATUS = 1", replacement),
+                    self.names,
+                    self.exports,
+                )
+
+    def test_wildcard_import_dependencies_are_rejected(self):
+        for source, destination in (
+            ("from other import *\n" + self.source, self.destination),
+            (self.source, "from other import *\n" + self.destination),
+        ):
+            with self.assertRaisesRegex(ValueError, "Wildcard import"):
+                copy_support_assignments(source, destination, self.names, self.exports)
+
+    def test_cli_hash_pins_scope_and_receipt(self):
+        with tempfile.TemporaryDirectory() as temp, contextlib.chdir(temp):
+            source = pathlib.Path("canonical/Lib/test/support/__init__.py")
+            target = pathlib.Path("Lib/test/support/__init__.py")
+            source.parent.mkdir(parents=True)
+            target.parent.mkdir(parents=True)
+            source.write_text(self.source)
+            target.write_text(self.destination)
+            args = [
+                str(source),
+                "--support-assignments",
+                *self.names,
+                "--support-exports",
+                *self.exports,
+                "--source-sha256",
+                hashlib.sha256(source.read_bytes()).hexdigest(),
+                "--destination-sha256",
+                "0" * 64,
+            ]
+            with contextlib.redirect_stderr(io.StringIO()):
+                self.assertEqual(main(args), 1)
+                self.assertEqual(
+                    main([str(source), "--support-assignments", *self.names]), 1
+                )
+            self.assertEqual(target.read_text(), self.destination)
+            args[-1] = hashlib.sha256(target.read_bytes()).hexdigest()
+            with contextlib.redirect_stdout(io.StringIO()) as output:
+                self.assertEqual(main(args), 0)
+            self.assertIn('"destination_after_sha256"', output.getvalue())
+            self.assertIn(self.block, target.read_text())
+            for path in (
+                "canonical/Lib/os.py",
+                "canonical/Lib/test/support/../../os.py",
+            ):
+                with contextlib.redirect_stderr(io.StringIO()):
+                    self.assertEqual(main([path, *args[1:]]), 1)
+
+    def test_selective_mode_rejects_existing_modes_before_reading_metadata(self):
+        for mode in ("--patches", "--legacy-map"):
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-assignments",
+                            *self.names,
+                            mode,
+                            "missing-metadata.json",
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("cannot combine", errors.getvalue())
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-exports",
+                            *self.exports,
+                            mode,
+                            "missing-metadata.json",
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("require --support-assignments", errors.getvalue())
+
+    def test_default_syntax_error_behavior_is_unchanged(self):
+        with patch(
+            "update_lib.cmd_migrate.patch_file", side_effect=SyntaxError("invalid")
+        ):
+            with self.assertRaises(SyntaxError):
+                main(["missing/Lib/test/support/example.py"])
+        with patch(
+            "update_lib.cmd_migrate.migrate_support_assignments",
+            side_effect=SyntaxError("invalid"),
+        ):
+            with contextlib.redirect_stderr(io.StringIO()) as errors:
+                self.assertEqual(
+                    main(
+                        [
+                            "missing/Lib/test/support/example.py",
+                            "--support-assignments",
+                            *self.names,
+                        ]
+                    ),
+                    1,
+                )
+            self.assertIn("invalid", errors.getvalue())
+
+
 class TestPatchSingleContent(unittest.TestCase):
     """Tests for patch_single_content function."""