"""

file = open(TESTFN, "wb")

try:

return file.fileno()

finally:

file.close()

global _can_symlink

if _can_symlink is not None:

return _can_symlink

symlink_path = TESTFN + "can_symlink"

try:

os.symlink(TESTFN, symlink_path)

can = True

except (OSError, NotImplementedError, AttributeError):

can = False

else:

os.remove(symlink_path)

_can_symlink = can

"""Skip decorator for tests that require functional symlink"""

ok = can_symlink()

msg = "Requires functional symlink implementation"

import tempfile

global _can_xattr

if _can_xattr is not None:

return _can_xattr

if not hasattr(os, "setxattr"):

can = False

else:

import platform

tmp_dir = tempfile.mkdtemp()

tmp_fp, tmp_name = tempfile.mkstemp(dir=tmp_dir)

try:

with open(TESTFN, "wb") as fp:

try:

# TESTFN & tempfile may use different file systems with

# different capabilities

os.setxattr(tmp_fp, b"user.test", b"")

os.setxattr(tmp_name, b"trusted.foo", b"42")

os.setxattr(fp.fileno(), b"user.test", b"")

# Kernels < 2.6.39 don't respect setxattr flags.

kernel_version = platform.release()

m = re.match(r"2.6.(\d{1,2})", kernel_version)

can = m is None or int(m.group(1)) >= 39

except OSError:

can = False

finally:

unlink(TESTFN)

unlink(tmp_name)

rmdir(tmp_dir)

_can_xattr = can

"""Skip decorator for tests that require functional extended attributes"""

ok = can_xattr()

msg = "no non-broken extended attribute support"

try:

_unlink(filename)

except (FileNotFoundError, NotADirectoryError):

# 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,

from test.support import _force_run

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("support.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)

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]

import shutil

try:

shutil.rmtree(path)

return

except OSError:

pass

def _rmtree_inner(path):

from test.support import _force_run

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)

try:

_rmdir(dirname)

except FileNotFoundError:

try:

_rmtree(path)

except FileNotFoundError:

"""Return a context manager that creates a temporary directory.

import tempfile

dir_created = False

if path is None:

path = tempfile.mkdtemp()

dir_created = True

path = os.path.realpath(path)

else:

try:

os.mkdir(path)

dir_created = True

except OSError as exc:

if not quiet:

raise

warnings.warn(f'tests may fail, unable to create '

f'temporary directory {path!r}: {exc}',

RuntimeWarning, stacklevel=3)

if dir_created:

pid = os.getpid()

try:

yield path

finally:

# In case the process forks, let only the parent remove the

# directory. The child has a different process id. (bpo-30028)

if dir_created and pid == os.getpid():

"""Return a context manager that changes the current working directory.

saved_dir = os.getcwd()

try:

os.chdir(os.path.realpath(path))

except OSError as exc:

if not quiet:

raise

warnings.warn(f'tests may fail, unable to change the current working '

f'directory to {path!r}: {exc}',

RuntimeWarning, stacklevel=3)

try:

yield os.getcwd()

finally:

"""

with temp_dir(path=name, quiet=quiet) as temp_path:

with change_cwd(temp_path, quiet=quiet) as cwd_dir:

"""Open a file descriptor to a directory."""

assert os.path.isdir(path)

dir_fd = os.open(path, os.O_RDONLY)

try:

yield dir_fd

finally:

"""Detects if the file system for the specified directory

import tempfile

with tempfile.NamedTemporaryFile(dir=directory) as base:

base_path = base.name

case_path = base_path.upper()

if case_path == base_path:

case_path = base_path.lower()

try:

return os.path.samefile(base_path, case_path)

except FileNotFoundError:

"""Simple implementing of the path protocol.

def __init__(self, path):

self.path = path

def __repr__(self):

return f'<FakePath {self.path!r}>'

def __fspath__(self):

if (isinstance(self.path, BaseException) or

isinstance(self.path, type) and

issubclass(self.path, BaseException)):

raise self.path

else:

"""Count the number of open file descriptors.

if sys.platform.startswith(('linux', 'freebsd', 'emscripten')):

try:

names = os.listdir("/proc/self/fd")

# Subtract one because listdir() internally opens a file

# descriptor to list the content of the /proc/self/fd/ directory.

return len(names) - 1

except FileNotFoundError:

pass

MAXFD = 256

if hasattr(os, 'sysconf'):

try:

MAXFD = os.sysconf("SC_OPEN_MAX")

except OSError:

pass

old_modes = None

if sys.platform == 'win32':

# bpo-25306, bpo-31009: Call CrtSetReportMode() to not kill the process

# on invalid file descriptor if Python is compiled in debug mode

try:

import msvcrt

msvcrt.CrtSetReportMode

except (AttributeError, ImportError):

# no msvcrt or a release build

pass

else:

old_modes = {}

for report_type in (msvcrt.CRT_WARN,

msvcrt.CRT_ERROR,

msvcrt.CRT_ASSERT):

old_modes[report_type] = msvcrt.CrtSetReportMode(report_type,

0)

try:

count = 0

for fd in range(MAXFD):

try:

# Prefer dup() over fstat(). fstat() can require input/output

# whereas dup() doesn't.

fd2 = os.dup(fd)

except OSError as e:

if e.errno != errno.EBADF:

raise

else:

os.close(fd2)

count += 1

finally:

if old_modes is not None:

for report_type in (msvcrt.CRT_WARN,

msvcrt.CRT_ERROR,

msvcrt.CRT_ASSERT):

msvcrt.CrtSetReportMode(report_type, old_modes[report_type])

"""Context manager that temporarily sets the process umask."""

oldmask = os.umask(umask)

try:

yield

finally:

"""Class to help protect the environment variable properly. Can be used as

def __init__(self):

self._environ = os.environ

self._changed = {}

def __getitem__(self, envvar):

return self._environ[envvar]

def __setitem__(self, envvar, value):

# Remember the initial value on the first access

if envvar not in self._changed:

self._changed[envvar] = self._environ.get(envvar)

self._environ[envvar] = value

def __delitem__(self, envvar):

# Remember the initial value on the first access

if envvar not in self._changed:

self._changed[envvar] = self._environ.get(envvar)

if envvar in self._environ:

del self._environ[envvar]

def keys(self):

return self._environ.keys()

def __iter__(self):

return iter(self._environ)

def __len__(self):

return len(self._environ)

def set(self, envvar, value):

self[envvar] = value

def unset(self, envvar):

del self[envvar]

def copy(self):

# We do what os.environ.copy() does.

return dict(self)

def __enter__(self):

return self

def __exit__(self, *ignore_exc):

for (k, v) in self._changed.items():

if v is None:

if k in self._environ:

del self._environ[k]

else:

self._environ[k] = v