@functools.wraps(func)

def wrap(*args, **kwargs):

try:

builtin_rename = os.rename

os.rename = _fake_rename

return func(*args, **kwargs)

finally:

os.rename = builtin_rename

"""Write *content* to a file located at *path*.

if isinstance(path, tuple):

path = os.path.join(*path)

if isinstance(content, str):

content = content.encode()

with open(path, 'xb') as fp:

"""Create a test file with an arbitrary size and random text content."""

def chunks(total, step):

assert total >= step

while total > step:

yield step

total -= step

if total:

yield total

bufsize = min(size, 8192)

chunk = b"".join([random.choice(string.ascii_letters).encode()

for i in range(bufsize)])

with open(path, 'wb') as f:

for csize in chunks(size, bufsize):

f.write(chunk)

"""Return contents from a file located at *path*.

if isinstance(path, tuple):

path = os.path.join(*path)

mode = 'rb' if binary else 'r'

encoding = None if binary else "utf-8"

with open(path, mode, encoding=encoding) as fp:

res = []

for name in sorted(os.listdir(path)):

p = os.path.join(path, name)

if os.path.isdir(p) and not os.path.islink(p):

res.append(name + '/')

for n in rlistdir(p):

res.append(name + '/' + n)

else:

res.append(name)

# ...apparently Linux and Solaris are the only ones

if not hasattr(os, "sendfile"):

return False

srcname = None

dstname = None

try:

with tempfile.NamedTemporaryFile("wb", dir=os.getcwd(), delete=False) as f:

srcname = f.name

f.write(b"0123456789")

with open(srcname, "rb") as src:

with tempfile.NamedTemporaryFile("wb", dir=os.getcwd(), delete=False) as dst:

dstname = dst.name

infd = src.fileno()

outfd = dst.fileno()

try:

os.sendfile(outfd, infd, 0, 2)

except OSError:

return False

else:

return True

finally:

if srcname is not None:

os_helper.unlink(srcname)

if dstname is not None:

if AIX and sys.maxsize == 2147483647:

hdrs=subprocess.getoutput("/usr/bin/dump -o %s" % sys.executable)

maxdata=hdrs.split("\n")[-1].split()[1]

return int(maxdata,16) >= 0x20000000

else:

def mkdtemp(self, prefix=None):

"""Create a temporary directory that will be cleaned up.

d = tempfile.mkdtemp(prefix=prefix, dir=os.getcwd())

self.addCleanup(os_helper.rmtree, d)

def test_rmtree_works_on_bytes(self):

tmp = self.mkdtemp()

victim = os.path.join(tmp, 'killme')

os.mkdir(victim)

create_file(os.path.join(victim, 'somefile'), 'foo')

victim = os.fsencode(victim)

self.assertIsInstance(victim, bytes)

shutil.rmtree(victim)

@os_helper.skip_unless_symlink

def test_rmtree_fails_on_symlink_onerror(self):

tmp = self.mkdtemp()

dir_ = os.path.join(tmp, 'dir')

os.mkdir(dir_)

link = os.path.join(tmp, 'link')

os.symlink(dir_, link)

self.assertRaises(OSError, shutil.rmtree, link)

self.assertTrue(os.path.exists(dir_))

self.assertTrue(os.path.lexists(link))

errors = []

def onerror(*args):

errors.append(args)

shutil.rmtree(link, onerror=onerror)

self.assertEqual(len(errors), 1)

self.assertIs(errors[0][0], os.path.islink)

self.assertEqual(errors[0][1], link)

self.assertIsInstance(errors[0][2][1], OSError)

@os_helper.skip_unless_symlink

def test_rmtree_fails_on_symlink_onexc(self):

tmp = self.mkdtemp()

dir_ = os.path.join(tmp, 'dir')

os.mkdir(dir_)

link = os.path.join(tmp, 'link')

os.symlink(dir_, link)

self.assertRaises(OSError, shutil.rmtree, link)

self.assertTrue(os.path.exists(dir_))

self.assertTrue(os.path.lexists(link))

errors = []

def onexc(*args):

errors.append(args)

shutil.rmtree(link, onexc=onexc)

self.assertEqual(len(errors), 1)

self.assertIs(errors[0][0], os.path.islink)

self.assertEqual(errors[0][1], link)

self.assertIsInstance(errors[0][2], OSError)

@os_helper.skip_unless_symlink

def test_rmtree_works_on_symlinks(self):

tmp = self.mkdtemp()

dir1 = os.path.join(tmp, 'dir1')

dir2 = os.path.join(dir1, 'dir2')

dir3 = os.path.join(tmp, 'dir3')

for d in dir1, dir2, dir3:

os.mkdir(d)

file1 = os.path.join(tmp, 'file1')

create_file(file1, 'foo')

link1 = os.path.join(dir1, 'link1')

os.symlink(dir2, link1)

link2 = os.path.join(dir1, 'link2')

os.symlink(dir3, link2)

link3 = os.path.join(dir1, 'link3')

os.symlink(file1, link3)

# make sure symlinks are removed but not followed

shutil.rmtree(dir1)

self.assertFalse(os.path.exists(dir1))

self.assertTrue(os.path.exists(dir3))

self.assertTrue(os.path.exists(file1))

@unittest.skipUnless(_winapi, 'only relevant on Windows')

def test_rmtree_fails_on_junctions_onerror(self):

tmp = self.mkdtemp()

dir_ = os.path.join(tmp, 'dir')

os.mkdir(dir_)

link = os.path.join(tmp, 'link')

_winapi.CreateJunction(dir_, link)

self.addCleanup(os_helper.unlink, link)

self.assertRaises(OSError, shutil.rmtree, link)

self.assertTrue(os.path.exists(dir_))

self.assertTrue(os.path.lexists(link))

errors = []

def onerror(*args):

errors.append(args)

shutil.rmtree(link, onerror=onerror)

self.assertEqual(len(errors), 1)

self.assertIs(errors[0][0], os.path.islink)

self.assertEqual(errors[0][1], link)

self.assertIsInstance(errors[0][2][1], OSError)

@unittest.skipUnless(_winapi, 'only relevant on Windows')

def test_rmtree_fails_on_junctions_onexc(self):

tmp = self.mkdtemp()

dir_ = os.path.join(tmp, 'dir')

os.mkdir(dir_)

link = os.path.join(tmp, 'link')

_winapi.CreateJunction(dir_, link)

self.addCleanup(os_helper.unlink, link)

self.assertRaises(OSError, shutil.rmtree, link)

self.assertTrue(os.path.exists(dir_))

self.assertTrue(os.path.lexists(link))

errors = []

def onexc(*args):

errors.append(args)

shutil.rmtree(link, onexc=onexc)

self.assertEqual(len(errors), 1)

self.assertIs(errors[0][0], os.path.islink)

self.assertEqual(errors[0][1], link)

self.assertIsInstance(errors[0][2], OSError)

@unittest.skipUnless(_winapi, 'only relevant on Windows')

def test_rmtree_works_on_junctions(self):

tmp = self.mkdtemp()

dir1 = os.path.join(tmp, 'dir1')

dir2 = os.path.join(dir1, 'dir2')

dir3 = os.path.join(tmp, 'dir3')

for d in dir1, dir2, dir3:

os.mkdir(d)

file1 = os.path.join(tmp, 'file1')

create_file(file1, 'foo')

link1 = os.path.join(dir1, 'link1')

_winapi.CreateJunction(dir2, link1)

link2 = os.path.join(dir1, 'link2')

_winapi.CreateJunction(dir3, link2)

link3 = os.path.join(dir1, 'link3')

_winapi.CreateJunction(file1, link3)

# make sure junctions are removed but not followed

shutil.rmtree(dir1)

self.assertFalse(os.path.exists(dir1))

self.assertTrue(os.path.exists(dir3))

self.assertTrue(os.path.exists(file1))

def test_rmtree_errors(self):

# filename is guaranteed not to exist

filename = tempfile.mktemp(dir=self.mkdtemp())

self.assertRaises(FileNotFoundError, shutil.rmtree, filename)

# test that ignore_errors option is honored

shutil.rmtree(filename, ignore_errors=True)

# existing file

tmpdir = self.mkdtemp()

filename = os.path.join(tmpdir, "tstfile")

create_file(filename)

with self.assertRaises(NotADirectoryError) as cm:

shutil.rmtree(filename)

self.assertEqual(cm.exception.filename, filename)

self.assertTrue(os.path.exists(filename))

# test that ignore_errors option is honored

shutil.rmtree(filename, ignore_errors=True)

self.assertTrue(os.path.exists(filename))

self.assertRaises(TypeError, shutil.rmtree, None)

self.assertRaises(TypeError, shutil.rmtree, None, ignore_errors=True)

exc = TypeError if shutil.rmtree.avoids_symlink_attacks else NotImplementedError

with self.assertRaises(exc):

shutil.rmtree(filename, dir_fd='invalid')

with self.assertRaises(exc):

shutil.rmtree(filename, dir_fd='invalid', ignore_errors=True)

def test_rmtree_errors_onerror(self):

tmpdir = self.mkdtemp()

filename = os.path.join(tmpdir, "tstfile")

create_file(filename)

errors = []

def onerror(*args):

errors.append(args)

shutil.rmtree(filename, onerror=onerror)

self.assertEqual(len(errors), 2)

self.assertIs(errors[0][0], os.scandir)

self.assertEqual(errors[0][1], filename)

self.assertIsInstance(errors[0][2][1], NotADirectoryError)

self.assertEqual(errors[0][2][1].filename, filename)

self.assertIs(errors[1][0], os.rmdir)

self.assertEqual(errors[1][1], filename)

self.assertIsInstance(errors[1][2][1], NotADirectoryError)

self.assertEqual(errors[1][2][1].filename, filename)

def test_rmtree_errors_onexc(self):

tmpdir = self.mkdtemp()

filename = os.path.join(tmpdir, "tstfile")

create_file(filename)

errors = []

def onexc(*args):

errors.append(args)

shutil.rmtree(filename, onexc=onexc)

self.assertEqual(len(errors), 2)

self.assertIs(errors[0][0], os.scandir)

self.assertEqual(errors[0][1], filename)

self.assertIsInstance(errors[0][2], NotADirectoryError)

self.assertEqual(errors[0][2].filename, filename)

self.assertIs(errors[1][0], os.rmdir)

self.assertEqual(errors[1][1], filename)

self.assertIsInstance(errors[1][2], NotADirectoryError)

self.assertEqual(errors[1][2].filename, filename)

@unittest.skipIf(sys.platform[:6] == 'cygwin',

@os_helper.skip_if_dac_override

@os_helper.skip_unless_working_chmod

def test_on_error(self):

self.errorState = 0

os.mkdir(TESTFN)

self.addCleanup(shutil.rmtree, TESTFN)

self.child_file_path = os.path.join(TESTFN, 'a')

self.child_dir_path = os.path.join(TESTFN, 'b')

os_helper.create_empty_file(self.child_file_path)

os.mkdir(self.child_dir_path)

old_dir_mode = os.stat(TESTFN).st_mode

old_child_file_mode = os.stat(self.child_file_path).st_mode

old_child_dir_mode = os.stat(self.child_dir_path).st_mode

# Make unwritable.

new_mode = stat.S_IREAD|stat.S_IEXEC

os.chmod(self.child_file_path, new_mode)

os.chmod(self.child_dir_path, new_mode)

os.chmod(TESTFN, new_mode)

self.addCleanup(os.chmod, TESTFN, old_dir_mode)

self.addCleanup(os.chmod, self.child_file_path, old_child_file_mode)

self.addCleanup(os.chmod, self.child_dir_path, old_child_dir_mode)

shutil.rmtree(TESTFN, onerror=self.check_args_to_onerror)

# Test whether onerror has actually been called.

self.assertEqual(self.errorState, 3,

"Expected call to onerror function did not happen.")

def check_args_to_onerror(self, func, arg, exc):

# test_rmtree_errors deliberately runs rmtree

# on a directory that is chmod 500, which will fail.

# This function is run when shutil.rmtree fails.

# 99.9% of the time it initially fails to remove

# a file in the directory, so the first time through

# func is os.remove.

# However, some Linux machines running ZFS on

# FUSE experienced a failure earlier in the process

# at os.listdir. The first failure may legally

# be either.

if self.errorState < 2:

if func is os.unlink:

self.assertEqual(arg, self.child_file_path)

elif func is os.rmdir:

self.assertEqual(arg, self.child_dir_path)

else:

self.assertIs(func, os.listdir)

self.assertIn(arg, [TESTFN, self.child_dir_path])

self.assertIsSubclass(exc[0], OSError)

self.errorState += 1

else:

self.assertEqual(func, os.rmdir)

self.assertEqual(arg, TESTFN)

self.assertIsSubclass(exc[0], OSError)

self.errorState = 3

@unittest.skipIf(sys.platform[:6] == 'cygwin',

@os_helper.skip_if_dac_override

@os_helper.skip_unless_working_chmod

def test_on_exc(self):

self.errorState = 0

os.mkdir(TESTFN)

self.addCleanup(shutil.rmtree, TESTFN)

self.child_file_path = os.path.join(TESTFN, 'a')

self.child_dir_path = os.path.join(TESTFN, 'b')

os_helper.create_empty_file(self.child_file_path)

os.mkdir(self.child_dir_path)

old_dir_mode = os.stat(TESTFN).st_mode

old_child_file_mode = os.stat(self.child_file_path).st_mode

old_child_dir_mode = os.stat(self.child_dir_path).st_mode

# Make unwritable.

new_mode = stat.S_IREAD|stat.S_IEXEC

os.chmod(self.child_file_path, new_mode)

os.chmod(self.child_dir_path, new_mode)

os.chmod(TESTFN, new_mode)

self.addCleanup(os.chmod, TESTFN, old_dir_mode)

self.addCleanup(os.chmod, self.child_file_path, old_child_file_mode)

self.addCleanup(os.chmod, self.child_dir_path, old_child_dir_mode)

shutil.rmtree(TESTFN, onexc=self.check_args_to_onexc)

# Test whether onexc has actually been called.

self.assertEqual(self.errorState, 3,

"Expected call to onexc function did not happen.")

def check_args_to_onexc(self, func, arg, exc):

# test_rmtree_errors deliberately runs rmtree

# on a directory that is chmod 500, which will fail.

# This function is run when shutil.rmtree fails.

# 99.9% of the time it initially fails to remove

# a file in the directory, so the first time through

# func is os.remove.

# However, some Linux machines running ZFS on

# FUSE experienced a failure earlier in the process

# at os.listdir. The first failure may legally

# be either.

if self.errorState < 2:

if func is os.unlink:

self.assertEqual(arg, self.child_file_path)

elif func is os.rmdir:

self.assertEqual(arg, self.child_dir_path)

else:

self.assertIs(func, os.listdir)

self.assertIn(arg, [TESTFN, self.child_dir_path])

self.assertTrue(isinstance(exc, OSError))

self.errorState += 1

else:

self.assertEqual(func, os.rmdir)

self.assertEqual(arg, TESTFN)

self.assertTrue(isinstance(exc, OSError))

self.errorState = 3

@os_helper.skip_if_dac_override

@os_helper.skip_unless_working_chmod

@unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks,

def test_on_exc_kwargs(self):

os.mkdir(TESTFN)

self.addCleanup(shutil.rmtree, TESTFN)

child_dir_path = os.path.join(TESTFN, 'b')

child_file_path = os.path.join(child_dir_path, 'a')

os.mkdir(child_dir_path)

os_helper.create_empty_file(child_file_path)

old_child_dir_mode = os.stat(child_dir_path).st_mode

# Make unwritable.

new_mode = stat.S_IREAD|stat.S_IEXEC

os.chmod(child_dir_path, new_mode)

self.addCleanup(os.chmod, child_dir_path, old_child_dir_mode)

calls = []

def onexc(func, path, err, direntry=None, dir_fd=None):

calls.append((func, path, err))

if func is os.unlink:

self.assertEqual(direntry.name, os.path.basename(path))

self.assertTrue(os.path.samestat(

os.stat(path), os.stat(direntry.name, dir_fd=dir_fd)))

shutil.rmtree(TESTFN, onexc=onexc, _onexc_kwargs=True)

self.assertIn((os.unlink, child_file_path),

[(func, path) for func, path, err in calls])

@unittest.skipIf(sys.platform[:6] == 'cygwin',

@os_helper.skip_if_dac_override

@os_helper.skip_unless_working_chmod

def test_both_onerror_and_onexc(self):

onerror_called = False

onexc_called = False

def onerror(*args):

nonlocal onerror_called

onerror_called = True

def onexc(*args):

nonlocal onexc_called

onexc_called = True

os.mkdir(TESTFN)

self.addCleanup(shutil.rmtree, TESTFN)

self.child_file_path = os.path.join(TESTFN, 'a')

self.child_dir_path = os.path.join(TESTFN, 'b')

os_helper.create_empty_file(self.child_file_path)

os.mkdir(self.child_dir_path)

old_dir_mode = os.stat(TESTFN).st_mode

old_child_file_mode = os.stat(self.child_file_path).st_mode

old_child_dir_mode = os.stat(self.child_dir_path).st_mode

# Make unwritable.

new_mode = stat.S_IREAD|stat.S_IEXEC

os.chmod(self.child_file_path, new_mode)

os.chmod(self.child_dir_path, new_mode)

os.chmod(TESTFN, new_mode)

self.addCleanup(os.chmod, TESTFN, old_dir_mode)

self.addCleanup(os.chmod, self.child_file_path, old_child_file_mode)

self.addCleanup(os.chmod, self.child_dir_path, old_child_dir_mode)

shutil.rmtree(TESTFN, onerror=onerror, onexc=onexc)

self.assertTrue(onexc_called)

self.assertFalse(onerror_called)

def test_rmtree_does_not_choke_on_failing_lstat(self):

try:

orig_lstat = os.lstat

def raiser(fn, *args, **kwargs):

if fn != TESTFN:

raise OSError()

else:

return orig_lstat(fn)

os.lstat = raiser

os.mkdir(TESTFN)

create_file((TESTFN, 'foo'), 'foo')

shutil.rmtree(TESTFN)

finally:

os.lstat = orig_lstat

def test_rmtree_uses_safe_fd_version_if_available(self):

if shutil._use_fd_functions:

self.assertTrue(shutil.rmtree.avoids_symlink_attacks)

tmp_dir = self.mkdtemp()

d = os.path.join(tmp_dir, 'a')

os.mkdir(d)

try:

real_open = os.open

class Called(Exception): pass

def _raiser(*args, **kwargs):

raise Called

os.open = _raiser

self.assertRaises(Called, shutil.rmtree, d)

finally:

os.open = real_open

else:

self.assertFalse(shutil.rmtree.avoids_symlink_attacks)

@unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks, "requires safe rmtree")

def test_rmtree_fails_on_close(self):

# Test that the error handler is called for failed os.close() and that

# os.close() is only called once for a file descriptor.

tmp = self.mkdtemp()

dir1 = os.path.join(tmp, 'dir1')

os.mkdir(dir1)

dir2 = os.path.join(dir1, 'dir2')

os.mkdir(dir2)

def close(fd):

orig_close(fd)

nonlocal close_count

close_count += 1

raise OSError

close_count = 0

with support.swap_attr(os, 'close', close) as orig_close:

with self.assertRaises(OSError):

shutil.rmtree(dir1)

self.assertTrue(os.path.isdir(dir2))

self.assertEqual(close_count, 2)

close_count = 0

errors = []

def onexc(*args):

errors.append(args)

with support.swap_attr(os, 'close', close) as orig_close:

shutil.rmtree(dir1, onexc=onexc)

self.assertEqual(len(errors), 2)

self.assertIs(errors[0][0], close)

self.assertEqual(errors[0][1], dir2)

self.assertIs(errors[1][0], close)

self.assertEqual(errors[1][1], dir1)

self.assertEqual(close_count, 2)

@unittest.skipUnless(shutil.rmtree.avoids_symlink_attacks, "dir_fd is not supported")

def test_rmtree_with_dir_fd(self):

tmp_dir = self.mkdtemp()

victim = 'killme'

fullname = os.path.join(tmp_dir, victim)

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

self.addCleanup(os.close, dir_fd)

os.mkdir(fullname)

os.mkdir(os.path.join(fullname, 'subdir'))

create_file(os.path.join(fullname, 'subdir', 'somefile'), 'foo')

self.assertTrue(os.path.exists(fullname))

shutil.rmtree(victim, dir_fd=dir_fd)

self.assertFalse(os.path.exists(fullname))

@unittest.skipIf(shutil.rmtree.avoids_symlink_attacks, "dir_fd is supported")

def test_rmtree_with_dir_fd_unsupported(self):

tmp_dir = self.mkdtemp()

with self.assertRaises(NotImplementedError):

shutil.rmtree(tmp_dir, dir_fd=0)

self.assertTrue(os.path.exists(tmp_dir))

def test_rmtree_dont_delete_file(self):

# When called on a file instead of a directory, don't delete it.

handle, path = tempfile.mkstemp(dir=self.mkdtemp())

os.close(handle)

self.assertRaises(NotADirectoryError, shutil.rmtree, path)

os.remove(path)

@os_helper.skip_unless_symlink

def test_rmtree_on_symlink(self):

# bug 1669.

os.mkdir(TESTFN)

try:

src = os.path.join(TESTFN, 'cheese')

dst = os.path.join(TESTFN, 'shop')

os.mkdir(src)

os.symlink(src, dst)

self.assertRaises(OSError, shutil.rmtree, dst)

shutil.rmtree(dst, ignore_errors=True)

finally:

shutil.rmtree(TESTFN, ignore_errors=True)

@unittest.skipUnless(_winapi, 'only relevant on Windows')

def test_rmtree_on_junction(self):

os.mkdir(TESTFN)

try:

src = os.path.join(TESTFN, 'cheese')

dst = os.path.join(TESTFN, 'shop')

os.mkdir(src)

create_file(os.path.join(src, 'spam'))

_winapi.CreateJunction(src, dst)

self.assertRaises(OSError, shutil.rmtree, dst)

shutil.rmtree(dst, ignore_errors=True)

finally:

shutil.rmtree(TESTFN, ignore_errors=True)

@unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')

@unittest.skipIf(sys.platform == "vxworks",

def test_rmtree_on_named_pipe(self):

os.mkfifo(TESTFN)

try:

with self.assertRaises(NotADirectoryError):

shutil.rmtree(TESTFN)

self.assertTrue(os.path.exists(TESTFN))

finally:

os.unlink(TESTFN)

os.mkdir(TESTFN)

os.mkfifo(os.path.join(TESTFN, 'mypipe'))

shutil.rmtree(TESTFN)

self.assertFalse(os.path.exists(TESTFN))

@unittest.skipIf(sys.platform[:6] == 'cygwin',

@os_helper.skip_if_dac_override

@os_helper.skip_unless_working_chmod

def test_rmtree_deleted_race_condition(self):

# bpo-37260

#

# Test that a file or a directory deleted after it is enumerated

# by scandir() but before unlink() or rmdr() is called doesn't

# generate any errors.

def _onexc(fn, path, exc):

assert fn in (os.rmdir, os.unlink)

if not isinstance(exc, PermissionError):

raise

# Make the parent and the children writeable.

for p, mode in zip(paths, old_modes):

os.chmod(p, mode)

# Remove other dirs except one.

keep = next(p for p in dirs if p != path)

for p in dirs:

if p != keep:

os.rmdir(p)

# Remove other files except one.

keep = next(p for p in files if p != path)

for p in files:

if p != keep:

os.unlink(p)

os.mkdir(TESTFN)

paths = [TESTFN] + [os.path.join(TESTFN, f'child{i}')

for i in range(6)]

dirs = paths[1::2]

files = paths[2::2]

for path in dirs:

os.mkdir(path)

for path in files:

create_file(path)

old_modes = [os.stat(path).st_mode for path in paths]

# Make the parent and the children non-writeable.

new_mode = stat.S_IREAD|stat.S_IEXEC

for path in reversed(paths):

os.chmod(path, new_mode)

try:

shutil.rmtree(TESTFN, onexc=_onexc)

except:

# Test failed, so cleanup artifacts.

for path, mode in zip(paths, old_modes):

try:

os.chmod(path, mode)

except OSError:

pass

shutil.rmtree(TESTFN)

raise

def test_rmtree_above_recursion_limit(self):

recursion_limit = 40

# directory_depth > recursion_limit

directory_depth = recursion_limit + 10

base = os.path.join(TESTFN, *(['d'] * directory_depth))

os.makedirs(base)

with support.infinite_recursion(recursion_limit):

def test_copytree_simple(self):

src_dir = self.mkdtemp()

dst_dir = os.path.join(self.mkdtemp(), 'destination')

self.addCleanup(shutil.rmtree, src_dir)

self.addCleanup(shutil.rmtree, os.path.dirname(dst_dir))

create_file((src_dir, 'test.txt'), '123')

os.mkdir(os.path.join(src_dir, 'test_dir'))

create_file((src_dir, 'test_dir', 'test.txt'), '456')

shutil.copytree(src_dir, dst_dir)

self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'test.txt')))

self.assertTrue(os.path.isdir(os.path.join(dst_dir, 'test_dir')))

self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'test_dir',

'test.txt')))

actual = read_file((dst_dir, 'test.txt'))

self.assertEqual(actual, '123')

actual = read_file((dst_dir, 'test_dir', 'test.txt'))

self.assertEqual(actual, '456')

def test_copytree_dirs_exist_ok(self):

src_dir = self.mkdtemp()

dst_dir = self.mkdtemp()

self.addCleanup(shutil.rmtree, src_dir)

self.addCleanup(shutil.rmtree, dst_dir)

create_file((src_dir, 'nonexisting.txt'), '123')

os.mkdir(os.path.join(src_dir, 'existing_dir'))

os.mkdir(os.path.join(dst_dir, 'existing_dir'))

create_file((dst_dir, 'existing_dir', 'existing.txt'), 'will be replaced')

create_file((src_dir, 'existing_dir', 'existing.txt'), 'has been replaced')

shutil.copytree(src_dir, dst_dir, dirs_exist_ok=True)

self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'nonexisting.txt')))

self.assertTrue(os.path.isdir(os.path.join(dst_dir, 'existing_dir')))

self.assertTrue(os.path.isfile(os.path.join(dst_dir, 'existing_dir',

'existing.txt')))

actual = read_file((dst_dir, 'nonexisting.txt'))

self.assertEqual(actual, '123')

actual = read_file((dst_dir, 'existing_dir', 'existing.txt'))

self.assertEqual(actual, 'has been replaced')

with self.assertRaises(FileExistsError):

shutil.copytree(src_dir, dst_dir, dirs_exist_ok=False)

@os_helper.skip_unless_symlink

def test_copytree_symlinks(self):

tmp_dir = self.mkdtemp()

src_dir = os.path.join(tmp_dir, 'src')

dst_dir = os.path.join(tmp_dir, 'dst')

sub_dir = os.path.join(src_dir, 'sub')

os.mkdir(src_dir)

os.mkdir(sub_dir)

create_file((src_dir, 'file.txt'), 'foo')

src_link = os.path.join(sub_dir, 'link')

dst_link = os.path.join(dst_dir, 'sub/link')

os.symlink(os.path.join(src_dir, 'file.txt'),

src_link)

if hasattr(os, 'lchmod'):

os.lchmod(src_link, stat.S_IRWXU | stat.S_IRWXO)

if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'):

os.lchflags(src_link, stat.UF_NODUMP)

src_stat = os.lstat(src_link)

shutil.copytree(src_dir, dst_dir, symlinks=True)

self.assertTrue(os.path.islink(os.path.join(dst_dir, 'sub', 'link')))

actual = os.readlink(os.path.join(dst_dir, 'sub', 'link'))

# Bad practice to blindly strip the prefix as it may be required to

# correctly refer to the file, but we're only comparing paths here.

if os.name == 'nt' and actual.startswith('\\\\?\\'):

actual = actual[4:]

self.assertEqual(actual, os.path.join(src_dir, 'file.txt'))

dst_stat = os.lstat(dst_link)

if hasattr(os, 'lchmod'):

self.assertEqual(dst_stat.st_mode, src_stat.st_mode)

if hasattr(os, 'lchflags'):

self.assertEqual(dst_stat.st_flags, src_stat.st_flags)

def test_copytree_with_exclude(self):

# creating data

join = os.path.join

exists = os.path.exists

src_dir = self.mkdtemp()

try:

dst_dir = join(self.mkdtemp(), 'destination')

create_file((src_dir, 'test.txt'), '123')

create_file((src_dir, 'test.tmp'), '123')

os.mkdir(join(src_dir, 'test_dir'))

create_file((src_dir, 'test_dir', 'test.txt'), '456')

os.mkdir(join(src_dir, 'test_dir2'))

create_file((src_dir, 'test_dir2', 'test.txt'), '456')

os.mkdir(join(src_dir, 'test_dir2', 'subdir'))

os.mkdir(join(src_dir, 'test_dir2', 'subdir2'))

create_file((src_dir, 'test_dir2', 'subdir', 'test.txt'), '456')

create_file((src_dir, 'test_dir2', 'subdir2', 'test.py'), '456')

# testing glob-like patterns

try:

patterns = shutil.ignore_patterns('*.tmp', 'test_dir2')

shutil.copytree(src_dir, dst_dir, ignore=patterns)

# checking the result: some elements should not be copied

self.assertTrue(exists(join(dst_dir, 'test.txt')))

self.assertFalse(exists(join(dst_dir, 'test.tmp')))

self.assertFalse(exists(join(dst_dir, 'test_dir2')))

finally:

shutil.rmtree(dst_dir)

try:

patterns = shutil.ignore_patterns('*.tmp', 'subdir*')

shutil.copytree(src_dir, dst_dir, ignore=patterns)

# checking the result: some elements should not be copied

self.assertFalse(exists(join(dst_dir, 'test.tmp')))

self.assertFalse(exists(join(dst_dir, 'test_dir2', 'subdir2')))

self.assertFalse(exists(join(dst_dir, 'test_dir2', 'subdir')))

finally:

shutil.rmtree(dst_dir)

# testing callable-style

try:

def _filter(src, names):

res = []

for name in names:

path = os.path.join(src, name)

if (os.path.isdir(path) and

path.split()[-1] == 'subdir'):

res.append(name)

elif os.path.splitext(path)[-1] in ('.py'):

res.append(name)

return res

shutil.copytree(src_dir, dst_dir, ignore=_filter)

# checking the result: some elements should not be copied

self.assertFalse(exists(join(dst_dir, 'test_dir2', 'subdir2',

'test.py')))

self.assertFalse(exists(join(dst_dir, 'test_dir2', 'subdir')))

finally:

shutil.rmtree(dst_dir)

finally:

shutil.rmtree(src_dir)

shutil.rmtree(os.path.dirname(dst_dir))

def test_copytree_arg_types_of_ignore(self):

join = os.path.join

exists = os.path.exists

tmp_dir = self.mkdtemp()

src_dir = join(tmp_dir, "source")

os.mkdir(join(src_dir))

os.mkdir(join(src_dir, 'test_dir'))

os.mkdir(os.path.join(src_dir, 'test_dir', 'subdir'))

create_file((src_dir, 'test_dir', 'subdir', 'test.txt'), '456')

invocations = []

def _ignore(src, names):

invocations.append(src)

self.assertIsInstance(src, str)

self.assertIsInstance(names, list)

self.assertEqual(len(names), len(set(names)))

for name in names:

self.assertIsInstance(name, str)

return []

dst_dir = join(self.mkdtemp(), 'destination')