Migrate constants and functions to helper

Author: fanninpm

Lib/test/support/__init__.py

@@ -83,9 +83,6 @@
     # io
     "record_original_stdout", "get_original_stdout", "captured_stdout",
     "captured_stdin", "captured_stderr",
-    # filesystem
-    "TESTFN", "SAVEDCWD", "unlink", "rmtree", "temp_cwd", "findfile",
-    "create_empty_file", "can_symlink", "fs_is_case_insensitive",
     # unittest
     "is_resource_enabled", "requires", "requires_freebsd_version",
     "requires_linux_version", "requires_mac_ver",
@@ -279,135 +276,6 @@ def _force_run(path, func, *args):
         os.chmod(path, stat.S_IRWXU)
         return func(*args)
 
-if sys.platform.startswith("win"):
-    def _waitfor(func, pathname, waitall=False):
-        # Perform the operation
-        func(pathname)
-        # Now setup the wait loop
-        if waitall:
-            dirname = pathname
-        else:
-            dirname, name = os.path.split(pathname)
-            dirname = dirname or '.'
-        # Check for `pathname` to be removed from the filesystem.
-        # The exponential backoff of the timeout amounts to a total
-        # of ~1 second after which the deletion is probably an error
-        # anyway.
-        # Testing on an i7@4.3GHz shows that usually only 1 iteration is
-        # required when contention occurs.
-        timeout = 0.001
-        while timeout < 1.0:
-            # Note we are only testing for the existence of the file(s) in
-            # the contents of the directory regardless of any security or
-            # access rights.  If we have made it this far, we have sufficient
-            # permissions to do that much using Python's equivalent of the
-            # Windows API FindFirstFile.
-            # Other Windows APIs can fail or give incorrect results when
-            # dealing with files that are pending deletion.
-            L = os.listdir(dirname)
-            if not (L if waitall else name in L):
-                return
-            # Increase the timeout and try again
-            time.sleep(timeout)
-            timeout *= 2
-        warnings.warn('tests may fail, delete still pending for ' + pathname,
-                      RuntimeWarning, stacklevel=4)
-
-    def _unlink(filename):
-        # XXX RUSTPYTHON: on ci, unlink() raises PermissionError when target doesn't exist.
-        # Might also happen locally, but not sure
-        if not os.path.exists(filename):
-            return
-        _waitfor(os.unlink, filename)
-
-    def _rmdir(dirname):
-        # XXX RUSTPYTHON: on ci, unlink() raises PermissionError when target doesn't exist.
-        # Might also happen locally, but not sure
-        if not os.path.exists(dirname):
-            return
-        _waitfor(os.rmdir, dirname)
-
-    def _rmtree(path):
-        # XXX RUSTPYTHON: on ci, unlink() raises PermissionError when target doesn't exist.
-        # Might also happen locally, but not sure
-        if not os.path.exists(path):
-            return
-        def _rmtree_inner(path):
-            for name in _force_run(path, os.listdir, path):
-                fullname = os.path.join(path, name)
-                try:
-                    mode = os.lstat(fullname).st_mode
-                except OSError as exc:
-                    print("os_helper.rmtree(): os.lstat(%r) failed with %s" % (fullname, exc),
-                          file=sys.__stderr__)
-                    mode = 0
-                if stat.S_ISDIR(mode):
-                    _waitfor(_rmtree_inner, fullname, waitall=True)
-                    _force_run(fullname, os.rmdir, fullname)
-                else:
-                    _force_run(fullname, os.unlink, fullname)
-        _waitfor(_rmtree_inner, path, waitall=True)
-        _waitfor(lambda p: _force_run(p, os.rmdir, p), path)
-
-    def _longpath(path):
-        try:
-            import ctypes
-        except ImportError:
-            # No ctypes means we can't expands paths.
-            pass
-        else:
-            buffer = ctypes.create_unicode_buffer(len(path) * 2)
-            length = ctypes.windll.kernel32.GetLongPathNameW(path, buffer,
-                                                             len(buffer))
-            if length:
-                return buffer[:length]
-        return path
-else:
-    _unlink = os.unlink
-    _rmdir = os.rmdir
-
-    def _rmtree(path):
-        try:
-            shutil.rmtree(path)
-            return
-        except OSError:
-            pass
-
-        def _rmtree_inner(path):
-            for name in _force_run(path, os.listdir, path):
-                fullname = os.path.join(path, name)
-                try:
-                    mode = os.lstat(fullname).st_mode
-                except OSError:
-                    mode = 0
-                if stat.S_ISDIR(mode):
-                    _rmtree_inner(fullname)
-                    _force_run(path, os.rmdir, fullname)
-                else:
-                    _force_run(path, os.unlink, fullname)
-        _rmtree_inner(path)
-        os.rmdir(path)
-
-    def _longpath(path):
-        return path
-
-def unlink(filename):
-    try:
-        _unlink(filename)
-    except (FileNotFoundError, NotADirectoryError):
-        pass
-
-def rmdir(dirname):
-    try:
-        _rmdir(dirname)
-    except FileNotFoundError:
-        pass
-
-def rmtree(path):
-    try:
-        _rmtree(path)
-    except FileNotFoundError:
-        pass
 
 # Check whether a gui is actually available
 def _is_gui_available():
@@ -2245,6 +2113,7 @@ def skip_unless_bind_unix_socket(test):
             except OSError as e:
                 _bind_nix_socket_error = e
             finally:
+                from .os_helper import unlink
                 unlink(path)
     if _bind_nix_socket_error:
         msg = 'Requires a functional unix bind(): %s' % _bind_nix_socket_error