# Copyright (C) 2003 Python Software Foundation import unittest import unittest.mock import shutil import tempfile import sys import stat import os import os.path import errno import functools import pathlib import subprocess import random import string import contextlib import io from shutil import (make_archive, register_archive_format, unregister_archive_format, get_archive_formats, Error, unpack_archive, register_unpack_format, RegistryError, unregister_unpack_format, get_unpack_formats, SameFileError, _GiveupOnFastCopy) import tarfile import zipfile try: import posix except ImportError: posix = None from test import support from test.support import os_helper from test.support.os_helper import TESTFN, FakePath TESTFN2 = TESTFN + "2" TESTFN_SRC = TESTFN + "_SRC" TESTFN_DST = TESTFN + "_DST" MACOS = sys.platform.startswith("darwin") SOLARIS = sys.platform.startswith("sunos") AIX = sys.platform[:3] == 'aix' try: import grp import pwd UID_GID_SUPPORT = True except ImportError: UID_GID_SUPPORT = False try: import _winapi except ImportError: _winapi = None no_chdir = unittest.mock.patch('os.chdir', side_effect=AssertionError("shouldn't call os.chdir()")) def _fake_rename(*args, **kwargs): # Pretend the destination path is on a different filesystem. raise OSError(getattr(errno, 'EXDEV', 18), "Invalid cross-device link") def mock_rename(func): @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 return wrap def write_file(path, content, binary=False): """Write *content* to a file located at *path*. If *path* is a tuple instead of a string, os.path.join will be used to make a path. If *binary* is true, the file will be opened in binary mode. """ if isinstance(path, tuple): path = os.path.join(*path) mode = 'wb' if binary else 'w' encoding = None if binary else "utf-8" with open(path, mode, encoding=encoding) as fp: fp.write(content) def write_test_file(path, size): """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) assert os.path.getsize(path) == size def read_file(path, binary=False): """Return contents from a file located at *path*. If *path* is a tuple instead of a string, os.path.join will be used to make a path. If *binary* is true, the file will be opened in binary mode. """ 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: return fp.read() def rlistdir(path): 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) return res def supports_file2file_sendfile(): # ...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: os_helper.unlink(dstname) SUPPORTS_SENDFILE = supports_file2file_sendfile() # AIX 32-bit mode, by default, lacks enough memory for the xz/lzma compiler test # The AIX command 'dump -o program' gives XCOFF header information # The second word of the last line in the maxdata value # when 32-bit maxdata must be greater than 0x1000000 for the xz test to succeed def _maxdataOK(): 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: return True class BaseTest: def mkdtemp(self, prefix=None): """Create a temporary directory that will be cleaned up. Returns the path of the directory. """ d = tempfile.mkdtemp(prefix=prefix, dir=os.getcwd()) self.addCleanup(os_helper.rmtree, d) return d class TestRmTree(BaseTest, unittest.TestCase): def test_rmtree_works_on_bytes(self): tmp = self.mkdtemp() victim = os.path.join(tmp, 'killme') os.mkdir(victim) write_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(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_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') write_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(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_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') write_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() write_file((tmpdir, "tstfile"), "") filename = os.path.join(tmpdir, "tstfile") 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)) 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) @unittest.skipIf(sys.platform[:6] == 'cygwin', "This test can't be run on Cygwin (issue #1071513).") @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.assertTrue(issubclass(exc[0], OSError)) self.errorState += 1 else: self.assertEqual(func, os.rmdir) self.assertEqual(arg, TESTFN) self.assertTrue(issubclass(exc[0], OSError)) self.errorState = 3 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) write_file((TESTFN, 'foo'), 'foo') shutil.rmtree(TESTFN) finally: os.lstat = orig_lstat def test_rmtree_uses_safe_fd_version_if_available(self): _use_fd_functions = ({os.open, os.stat, os.unlink, os.rmdir} <= os.supports_dir_fd and os.listdir in os.supports_fd and os.stat in os.supports_follow_symlinks) if _use_fd_functions: self.assertTrue(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_rmtree = shutil._rmtree_safe_fd class Called(Exception): pass def _raiser(*args, **kwargs): raise Called shutil._rmtree_safe_fd = _raiser self.assertRaises(Called, shutil.rmtree, d) finally: shutil._rmtree_safe_fd = real_rmtree else: self.assertFalse(shutil._use_fd_functions) self.assertFalse(shutil.rmtree.avoids_symlink_attacks) @unittest.skipUnless(shutil._use_fd_functions, "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')) write_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._use_fd_functions, "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) open(os.path.join(src, 'spam'), 'wb').close() _winapi.CreateJunction(src, dst) self.assertRaises(OSError, shutil.rmtree, dst) shutil.rmtree(dst, ignore_errors=True) finally: shutil.rmtree(TESTFN, ignore_errors=True) class TestCopyTree(BaseTest, unittest.TestCase): 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)) write_file((src_dir, 'test.txt'), '123') os.mkdir(os.path.join(src_dir, 'test_dir')) write_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) write_file((src_dir, 'nonexisting.txt'), '123') os.mkdir(os.path.join(src_dir, 'existing_dir')) os.mkdir(os.path.join(dst_dir, 'existing_dir')) write_file((dst_dir, 'existing_dir', 'existing.txt'), 'will be replaced') write_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) write_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') write_file((src_dir, 'test.txt'), '123') write_file((src_dir, 'test.tmp'), '123') os.mkdir(join(src_dir, 'test_dir')) write_file((src_dir, 'test_dir', 'test.txt'), '456') os.mkdir(join(src_dir, 'test_dir2')) write_file((src_dir, 'test_dir2', 'test.txt'), '456') os.mkdir(join(src_dir, 'test_dir2', 'subdir')) os.mkdir(join(src_dir, 'test_dir2', 'subdir2')) write_file((src_dir, 'test_dir2', 'subdir', 'test.txt'), '456') write_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')) write_file((src_dir, 'test_dir', 'subdir', 'test.txt'), '456') invokations = [] def _ignore(src, names): invokations.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') shutil.copytree(src_dir, dst_dir, ignore=_ignore) self.assertTrue(exists(join(dst_dir, 'test_dir', 'subdir', 'test.txt'))) dst_dir = join(self.mkdtemp(), 'destination') shutil.copytree(pathlib.Path(src_dir), dst_dir, ignore=_ignore) self.assertTrue(exists(join(dst_dir, 'test_dir', 'subdir', 'test.txt'))) dst_dir = join(self.mkdtemp(), 'destination') src_dir_entry = list(os.scandir(tmp_dir))[0] self.assertIsInstance(src_dir_entry, os.DirEntry) shutil.copytree(src_dir_entry, dst_dir, ignore=_ignore) self.assertTrue(exists(join(dst_dir, 'test_dir', 'subdir', 'test.txt'))) self.assertEqual(len(invokations), 9) def test_copytree_retains_permissions(self): tmp_dir = self.mkdtemp() src_dir = os.path.join(tmp_dir, 'source') os.mkdir(src_dir) dst_dir = os.path.join(tmp_dir, 'destination') self.addCleanup(shutil.rmtree, tmp_dir) os.chmod(src_dir, 0o777) write_file((src_dir, 'permissive.txt'), '123') os.chmod(os.path.join(src_dir, 'permissive.txt'), 0o777) write_file((src_dir, 'restrictive.txt'), '456') os.chmod(os.path.join(src_dir, 'restrictive.txt'), 0o600) restrictive_subdir = tempfile.mkdtemp(dir=src_dir) self.addCleanup(os_helper.rmtree, restrictive_subdir) os.chmod(restrictive_subdir, 0o600) shutil.copytree(src_dir, dst_dir) self.assertEqual(os.stat(src_dir).st_mode, os.stat(dst_dir).st_mode) self.assertEqual(os.stat(os.path.join(src_dir, 'permissive.txt')).st_mode, os.stat(os.path.join(dst_dir, 'permissive.txt')).st_mode) self.assertEqual(os.stat(os.path.join(src_dir, 'restrictive.txt')).st_mode, os.stat(os.path.join(dst_dir, 'restrictive.txt')).st_mode) restrictive_subdir_dst = os.path.join(dst_dir, os.path.split(restrictive_subdir)[1]) self.assertEqual(os.stat(restrictive_subdir).st_mode, os.stat(restrictive_subdir_dst).st_mode) @unittest.mock.patch('os.chmod') def test_copytree_winerror(self, mock_patch): # When copying to VFAT, copystat() raises OSError. On Windows, the # exception object has a meaningful 'winerror' attribute, but not # on other operating systems. Do not assume 'winerror' is set. 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)) mock_patch.side_effect = PermissionError('ka-boom') with self.assertRaises(shutil.Error): shutil.copytree(src_dir, dst_dir) def test_copytree_custom_copy_function(self): # See: https://bugs.python.org/issue35648 def custom_cpfun(a, b): flag.append(None) self.assertIsInstance(a, str) self.assertIsInstance(b, str) self.assertEqual(a, os.path.join(src, 'foo')) self.assertEqual(b, os.path.join(dst, 'foo')) flag = [] src = self.mkdtemp() dst = tempfile.mktemp(dir=self.mkdtemp()) with open(os.path.join(src, 'foo'), 'w', encoding='utf-8') as f: f.close() shutil.copytree(src, dst, copy_function=custom_cpfun) self.assertEqual(len(flag), 1) # Issue #3002: copyfile and copytree block indefinitely on named pipes @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()') @os_helper.skip_unless_symlink @unittest.skipIf(sys.platform == "vxworks", "fifo requires special path on VxWorks") def test_copytree_named_pipe(self): os.mkdir(TESTFN) try: subdir = os.path.join(TESTFN, "subdir") os.mkdir(subdir) pipe = os.path.join(subdir, "mypipe") try: os.mkfifo(pipe) except PermissionError as e: self.skipTest('os.mkfifo(): %s' % e) try: shutil.copytree(TESTFN, TESTFN2) except shutil.Error as e: errors = e.args[0] self.assertEqual(len(errors), 1) src, dst, error_msg = errors[0] self.assertEqual("`%s` is a named pipe" % pipe, error_msg) else: self.fail("shutil.Error should have been raised") finally: shutil.rmtree(TESTFN, ignore_errors=True) shutil.rmtree(TESTFN2, ignore_errors=True) def test_copytree_special_func(self): src_dir = self.mkdtemp() dst_dir = os.path.join(self.mkdtemp(), 'destination') write_file((src_dir, 'test.txt'), '123') os.mkdir(os.path.join(src_dir, 'test_dir')) write_file((src_dir, 'test_dir', 'test.txt'), '456') copied = [] def _copy(src, dst): copied.append((src, dst)) shutil.copytree(src_dir, dst_dir, copy_function=_copy) self.assertEqual(len(copied), 2) @os_helper.skip_unless_symlink def test_copytree_dangling_symlinks(self): # a dangling symlink raises an error at the end src_dir = self.mkdtemp() dst_dir = os.path.join(self.mkdtemp(), 'destination') os.symlink('IDONTEXIST', os.path.join(src_dir, 'test.txt')) os.mkdir(os.path.join(src_dir, 'test_dir')) write_file((src_dir, 'test_dir', 'test.txt'), '456') self.assertRaises(Error, shutil.copytree, src_dir, dst_dir) # a dangling symlink is ignored with the proper flag dst_dir = os.path.join(self.mkdtemp(), 'destination2') shutil.copytree(src_dir, dst_dir, ignore_dangling_symlinks=True) self.assertNotIn('test.txt', os.listdir(dst_dir)) # a dangling symlink is copied if symlinks=True dst_dir = os.path.join(self.mkdtemp(), 'destination3') shutil.copytree(src_dir, dst_dir, symlinks=True) self.assertIn('test.txt', os.listdir(dst_dir)) @os_helper.skip_unless_symlink def test_copytree_symlink_dir(self): src_dir = self.mkdtemp() dst_dir = os.path.join(self.mkdtemp(), 'destination') os.mkdir(os.path.join(src_dir, 'real_dir')) with open(os.path.join(src_dir, 'real_dir', 'test.txt'), 'wb'): pass os.symlink(os.path.join(src_dir, 'real_dir'), os.path.join(src_dir, 'link_to_dir'), target_is_directory=True) shutil.copytree(src_dir, dst_dir, symlinks=False) self.assertFalse(os.path.islink(os.path.join(dst_dir, 'link_to_dir'))) self.assertIn('test.txt', os.listdir(os.path.join(dst_dir, 'link_to_dir'))) dst_dir = os.path.join(self.mkdtemp(), 'destination2') shutil.copytree(src_dir, dst_dir, symlinks=True) self.assertTrue(os.path.islink(os.path.join(dst_dir, 'link_to_dir'))) self.assertIn('test.txt', os.listdir(os.path.join(dst_dir, 'link_to_dir'))) def test_copytree_return_value(self): # copytree returns its destination path. src_dir = self.mkdtemp() dst_dir = src_dir + "dest" self.addCleanup(shutil.rmtree, dst_dir, True) src = os.path.join(src_dir, 'foo') write_file(src, 'foo') rv = shutil.copytree(src_dir, dst_dir) self.assertEqual(['foo'], os.listdir(rv)) def test_copytree_subdirectory(self): # copytree where dst is a subdirectory of src, see Issue 38688 base_dir = self.mkdtemp() self.addCleanup(shutil.rmtree, base_dir, ignore_errors=True) src_dir = os.path.join(base_dir, "t", "pg") dst_dir = os.path.join(src_dir, "somevendor", "1.0") os.makedirs(src_dir) src = os.path.join(src_dir, 'pol') write_file(src, 'pol') rv = shutil.copytree(src_dir, dst_dir) self.assertEqual(['pol'], os.listdir(rv)) class TestCopy(BaseTest, unittest.TestCase): ### shutil.copymode @os_helper.skip_unless_symlink def test_copymode_follow_symlinks(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') dst_link = os.path.join(tmp_dir, 'quux') write_file(src, 'foo') write_file(dst, 'foo') os.symlink(src, src_link) os.symlink(dst, dst_link) os.chmod(src, stat.S_IRWXU|stat.S_IRWXG) # file to file os.chmod(dst, stat.S_IRWXO) self.assertNotEqual(os.stat(src).st_mode, os.stat(dst).st_mode) shutil.copymode(src, dst) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) # On Windows, os.chmod does not follow symlinks (issue #15411) if os.name != 'nt': # follow src link os.chmod(dst, stat.S_IRWXO) shutil.copymode(src_link, dst) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) # follow dst link os.chmod(dst, stat.S_IRWXO) shutil.copymode(src, dst_link) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) # follow both links os.chmod(dst, stat.S_IRWXO) shutil.copymode(src_link, dst_link) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) @unittest.skipUnless(hasattr(os, 'lchmod'), 'requires os.lchmod') @os_helper.skip_unless_symlink def test_copymode_symlink_to_symlink(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') dst_link = os.path.join(tmp_dir, 'quux') write_file(src, 'foo') write_file(dst, 'foo') os.symlink(src, src_link) os.symlink(dst, dst_link) os.chmod(src, stat.S_IRWXU|stat.S_IRWXG) os.chmod(dst, stat.S_IRWXU) os.lchmod(src_link, stat.S_IRWXO|stat.S_IRWXG) # link to link os.lchmod(dst_link, stat.S_IRWXO) shutil.copymode(src_link, dst_link, follow_symlinks=False) self.assertEqual(os.lstat(src_link).st_mode, os.lstat(dst_link).st_mode) self.assertNotEqual(os.stat(src).st_mode, os.stat(dst).st_mode) # src link - use chmod os.lchmod(dst_link, stat.S_IRWXO) shutil.copymode(src_link, dst, follow_symlinks=False) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) # dst link - use chmod os.lchmod(dst_link, stat.S_IRWXO) shutil.copymode(src, dst_link, follow_symlinks=False) self.assertEqual(os.stat(src).st_mode, os.stat(dst).st_mode) @unittest.skipIf(hasattr(os, 'lchmod'), 'requires os.lchmod to be missing') @os_helper.skip_unless_symlink def test_copymode_symlink_to_symlink_wo_lchmod(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') dst_link = os.path.join(tmp_dir, 'quux') write_file(src, 'foo') write_file(dst, 'foo') os.symlink(src, src_link) os.symlink(dst, dst_link) shutil.copymode(src_link, dst_link, follow_symlinks=False) # silent fail ### shutil.copystat @os_helper.skip_unless_symlink def test_copystat_symlinks(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') dst_link = os.path.join(tmp_dir, 'qux') write_file(src, 'foo') src_stat = os.stat(src) os.utime(src, (src_stat.st_atime, src_stat.st_mtime - 42.0)) # ensure different mtimes write_file(dst, 'bar') self.assertNotEqual(os.stat(src).st_mtime, os.stat(dst).st_mtime) os.symlink(src, src_link) os.symlink(dst, dst_link) if hasattr(os, 'lchmod'): os.lchmod(src_link, stat.S_IRWXO) if hasattr(os, 'lchflags') and hasattr(stat, 'UF_NODUMP'): os.lchflags(src_link, stat.UF_NODUMP) src_link_stat = os.lstat(src_link) # follow if hasattr(os, 'lchmod'): shutil.copystat(src_link, dst_link, follow_symlinks=True) self.assertNotEqual(src_link_stat.st_mode, os.stat(dst).st_mode) # don't follow shutil.copystat(src_link, dst_link, follow_symlinks=False) dst_link_stat = os.lstat(dst_link) if os.utime in os.supports_follow_symlinks: for attr in 'st_atime', 'st_mtime': # The modification times may be truncated in the new file. self.assertLessEqual(getattr(src_link_stat, attr), getattr(dst_link_stat, attr) + 1) if hasattr(os, 'lchmod'): self.assertEqual(src_link_stat.st_mode, dst_link_stat.st_mode) if hasattr(os, 'lchflags') and hasattr(src_link_stat, 'st_flags'): self.assertEqual(src_link_stat.st_flags, dst_link_stat.st_flags) # tell to follow but dst is not a link shutil.copystat(src_link, dst, follow_symlinks=False) self.assertTrue(abs(os.stat(src).st_mtime - os.stat(dst).st_mtime) < 00000.1) @unittest.skipUnless(hasattr(os, 'chflags') and hasattr(errno, 'EOPNOTSUPP') and hasattr(errno, 'ENOTSUP'), "requires os.chflags, EOPNOTSUPP & ENOTSUP") def test_copystat_handles_harmless_chflags_errors(self): tmpdir = self.mkdtemp() file1 = os.path.join(tmpdir, 'file1') file2 = os.path.join(tmpdir, 'file2') write_file(file1, 'xxx') write_file(file2, 'xxx') def make_chflags_raiser(err): ex = OSError() def _chflags_raiser(path, flags, *, follow_symlinks=True): ex.errno = err raise ex return _chflags_raiser old_chflags = os.chflags try: for err in errno.EOPNOTSUPP, errno.ENOTSUP: os.chflags = make_chflags_raiser(err) shutil.copystat(file1, file2) # assert others errors break it os.chflags = make_chflags_raiser(errno.EOPNOTSUPP + errno.ENOTSUP) self.assertRaises(OSError, shutil.copystat, file1, file2) finally: os.chflags = old_chflags ### shutil.copyxattr @os_helper.skip_unless_xattr def test_copyxattr(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') write_file(src, 'foo') dst = os.path.join(tmp_dir, 'bar') write_file(dst, 'bar') # no xattr == no problem shutil._copyxattr(src, dst) # common case os.setxattr(src, 'user.foo', b'42') os.setxattr(src, 'user.bar', b'43') shutil._copyxattr(src, dst) self.assertEqual(sorted(os.listxattr(src)), sorted(os.listxattr(dst))) self.assertEqual( os.getxattr(src, 'user.foo'), os.getxattr(dst, 'user.foo')) # check errors don't affect other attrs os.remove(dst) write_file(dst, 'bar') os_error = OSError(errno.EPERM, 'EPERM') def _raise_on_user_foo(fname, attr, val, **kwargs): if attr == 'user.foo': raise os_error else: orig_setxattr(fname, attr, val, **kwargs) try: orig_setxattr = os.setxattr os.setxattr = _raise_on_user_foo shutil._copyxattr(src, dst) self.assertIn('user.bar', os.listxattr(dst)) finally: os.setxattr = orig_setxattr # the source filesystem not supporting xattrs should be ok, too. def _raise_on_src(fname, *, follow_symlinks=True): if fname == src: raise OSError(errno.ENOTSUP, 'Operation not supported') return orig_listxattr(fname, follow_symlinks=follow_symlinks) try: orig_listxattr = os.listxattr os.listxattr = _raise_on_src shutil._copyxattr(src, dst) finally: os.listxattr = orig_listxattr # test that shutil.copystat copies xattrs src = os.path.join(tmp_dir, 'the_original') srcro = os.path.join(tmp_dir, 'the_original_ro') write_file(src, src) write_file(srcro, srcro) os.setxattr(src, 'user.the_value', b'fiddly') os.setxattr(srcro, 'user.the_value', b'fiddly') os.chmod(srcro, 0o444) dst = os.path.join(tmp_dir, 'the_copy') dstro = os.path.join(tmp_dir, 'the_copy_ro') write_file(dst, dst) write_file(dstro, dstro) shutil.copystat(src, dst) shutil.copystat(srcro, dstro) self.assertEqual(os.getxattr(dst, 'user.the_value'), b'fiddly') self.assertEqual(os.getxattr(dstro, 'user.the_value'), b'fiddly') @os_helper.skip_unless_symlink @os_helper.skip_unless_xattr @os_helper.skip_unless_dac_override def test_copyxattr_symlinks(self): # On Linux, it's only possible to access non-user xattr for symlinks; # which in turn require root privileges. This test should be expanded # as soon as other platforms gain support for extended attributes. tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') src_link = os.path.join(tmp_dir, 'baz') write_file(src, 'foo') os.symlink(src, src_link) os.setxattr(src, 'trusted.foo', b'42') os.setxattr(src_link, 'trusted.foo', b'43', follow_symlinks=False) dst = os.path.join(tmp_dir, 'bar') dst_link = os.path.join(tmp_dir, 'qux') write_file(dst, 'bar') os.symlink(dst, dst_link) shutil._copyxattr(src_link, dst_link, follow_symlinks=False) self.assertEqual(os.getxattr(dst_link, 'trusted.foo', follow_symlinks=False), b'43') self.assertRaises(OSError, os.getxattr, dst, 'trusted.foo') shutil._copyxattr(src_link, dst, follow_symlinks=False) self.assertEqual(os.getxattr(dst, 'trusted.foo'), b'43') ### shutil.copy def _copy_file(self, method): fname = 'test.txt' tmpdir = self.mkdtemp() write_file((tmpdir, fname), 'xxx') file1 = os.path.join(tmpdir, fname) tmpdir2 = self.mkdtemp() method(file1, tmpdir2) file2 = os.path.join(tmpdir2, fname) return (file1, file2) def test_copy(self): # Ensure that the copied file exists and has the same mode bits. file1, file2 = self._copy_file(shutil.copy) self.assertTrue(os.path.exists(file2)) self.assertEqual(os.stat(file1).st_mode, os.stat(file2).st_mode) @os_helper.skip_unless_symlink def test_copy_symlinks(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') write_file(src, 'foo') os.symlink(src, src_link) if hasattr(os, 'lchmod'): os.lchmod(src_link, stat.S_IRWXU | stat.S_IRWXO) # don't follow shutil.copy(src_link, dst, follow_symlinks=True) self.assertFalse(os.path.islink(dst)) self.assertEqual(read_file(src), read_file(dst)) os.remove(dst) # follow shutil.copy(src_link, dst, follow_symlinks=False) self.assertTrue(os.path.islink(dst)) self.assertEqual(os.readlink(dst), os.readlink(src_link)) if hasattr(os, 'lchmod'): self.assertEqual(os.lstat(src_link).st_mode, os.lstat(dst).st_mode) ### shutil.copy2 @unittest.skipUnless(hasattr(os, 'utime'), 'requires os.utime') def test_copy2(self): # Ensure that the copied file exists and has the same mode and # modification time bits. file1, file2 = self._copy_file(shutil.copy2) self.assertTrue(os.path.exists(file2)) file1_stat = os.stat(file1) file2_stat = os.stat(file2) self.assertEqual(file1_stat.st_mode, file2_stat.st_mode) for attr in 'st_atime', 'st_mtime': # The modification times may be truncated in the new file. self.assertLessEqual(getattr(file1_stat, attr), getattr(file2_stat, attr) + 1) if hasattr(os, 'chflags') and hasattr(file1_stat, 'st_flags'): self.assertEqual(getattr(file1_stat, 'st_flags'), getattr(file2_stat, 'st_flags')) @os_helper.skip_unless_symlink def test_copy2_symlinks(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') src_link = os.path.join(tmp_dir, 'baz') write_file(src, 'foo') os.symlink(src, 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.stat(src) src_link_stat = os.lstat(src_link) # follow shutil.copy2(src_link, dst, follow_symlinks=True) self.assertFalse(os.path.islink(dst)) self.assertEqual(read_file(src), read_file(dst)) os.remove(dst) # don't follow shutil.copy2(src_link, dst, follow_symlinks=False) self.assertTrue(os.path.islink(dst)) self.assertEqual(os.readlink(dst), os.readlink(src_link)) dst_stat = os.lstat(dst) if os.utime in os.supports_follow_symlinks: for attr in 'st_atime', 'st_mtime': # The modification times may be truncated in the new file. self.assertLessEqual(getattr(src_link_stat, attr), getattr(dst_stat, attr) + 1) if hasattr(os, 'lchmod'): self.assertEqual(src_link_stat.st_mode, dst_stat.st_mode) self.assertNotEqual(src_stat.st_mode, dst_stat.st_mode) if hasattr(os, 'lchflags') and hasattr(src_link_stat, 'st_flags'): self.assertEqual(src_link_stat.st_flags, dst_stat.st_flags) @os_helper.skip_unless_xattr def test_copy2_xattr(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'foo') dst = os.path.join(tmp_dir, 'bar') write_file(src, 'foo') os.setxattr(src, 'user.foo', b'42') shutil.copy2(src, dst) self.assertEqual( os.getxattr(src, 'user.foo'), os.getxattr(dst, 'user.foo')) os.remove(dst) def test_copy_return_value(self): # copy and copy2 both return their destination path. for fn in (shutil.copy, shutil.copy2): src_dir = self.mkdtemp() dst_dir = self.mkdtemp() src = os.path.join(src_dir, 'foo') write_file(src, 'foo') rv = fn(src, dst_dir) self.assertEqual(rv, os.path.join(dst_dir, 'foo')) rv = fn(src, os.path.join(dst_dir, 'bar')) self.assertEqual(rv, os.path.join(dst_dir, 'bar')) def test_copy_dir(self): self._test_copy_dir(shutil.copy) def test_copy2_dir(self): self._test_copy_dir(shutil.copy2) def _test_copy_dir(self, copy_func): src_dir = self.mkdtemp() src_file = os.path.join(src_dir, 'foo') dir2 = self.mkdtemp() dst = os.path.join(src_dir, 'does_not_exist/') write_file(src_file, 'foo') if sys.platform == "win32": err = PermissionError else: err = IsADirectoryError self.assertRaises(err, copy_func, dir2, src_dir) # raise *err* because of src rather than FileNotFoundError because of dst self.assertRaises(err, copy_func, dir2, dst) copy_func(src_file, dir2) # should not raise exceptions ### shutil.copyfile @os_helper.skip_unless_symlink def test_copyfile_symlinks(self): tmp_dir = self.mkdtemp() src = os.path.join(tmp_dir, 'src') dst = os.path.join(tmp_dir, 'dst') dst_link = os.path.join(tmp_dir, 'dst_link') link = os.path.join(tmp_dir, 'link') write_file(src, 'foo') os.symlink(src, link) # don't follow shutil.copyfile(link, dst_link, follow_symlinks=False) self.assertTrue(os.path.islink(dst_link)) self.assertEqual(os.readlink(link), os.readlink(dst_link)) # follow shutil.copyfile(link, dst) self.assertFalse(os.path.islink(dst)) @unittest.skipUnless(hasattr(os, 'link'), 'requires os.link') def test_dont_copy_file_onto_link_to_itself(self): # bug 851123. os.mkdir(TESTFN) src = os.path.join(TESTFN, 'cheese') dst = os.path.join(TESTFN, 'shop') try: with open(src, 'w', encoding='utf-8') as f: f.write('cheddar') try: os.link(src, dst) except PermissionError as e: self.skipTest('os.link(): %s' % e) self.assertRaises(shutil.SameFileError, shutil.copyfile, src, dst) with open(src, 'r', encoding='utf-8') as f: self.assertEqual(f.read(), 'cheddar') os.remove(dst) finally: shutil.rmtree(TESTFN, ignore_errors=True) @os_helper.skip_unless_symlink def test_dont_copy_file_onto_symlink_to_itself(self): # bug 851123. os.mkdir(TESTFN) src = os.path.join(TESTFN, 'cheese') dst = os.path.join(TESTFN, 'shop') try: with open(src, 'w', encoding='utf-8') as f: f.write('cheddar') # Using `src` here would mean we end up with a symlink pointing # to TESTFN/TESTFN/cheese, while it should point at # TESTFN/cheese. os.symlink('cheese', dst) self.assertRaises(shutil.SameFileError, shutil.copyfile, src, dst) with open(src, 'r', encoding='utf-8') as f: self.assertEqual(f.read(), 'cheddar') os.remove(dst) finally: shutil.rmtree(TESTFN, ignore_errors=True) # Issue #3002: copyfile and copytree block indefinitely on named pipes @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()') @unittest.skipIf(sys.platform == "vxworks", "fifo requires special path on VxWorks") def test_copyfile_named_pipe(self): try: os.mkfifo(TESTFN) except PermissionError as e: self.skipTest('os.mkfifo(): %s' % e) try: self.assertRaises(shutil.SpecialFileError, shutil.copyfile, TESTFN, TESTFN2) self.assertRaises(shutil.SpecialFileError, shutil.copyfile, __file__, TESTFN) finally: os.remove(TESTFN) def test_copyfile_return_value(self): # copytree returns its destination path. src_dir = self.mkdtemp() dst_dir = self.mkdtemp() dst_file = os.path.join(dst_dir, 'bar') src_file = os.path.join(src_dir, 'foo') write_file(src_file, 'foo') rv = shutil.copyfile(src_file, dst_file) self.assertTrue(os.path.exists(rv)) self.assertEqual(read_file(src_file), read_file(dst_file)) def test_copyfile_same_file(self): # copyfile() should raise SameFileError if the source and destination # are the same. src_dir = self.mkdtemp() src_file = os.path.join(src_dir, 'foo') write_file(src_file, 'foo') self.assertRaises(SameFileError, shutil.copyfile, src_file, src_file) # But Error should work too, to stay backward compatible. self.assertRaises(Error, shutil.copyfile, src_file, src_file) # Make sure file is not corrupted. self.assertEqual(read_file(src_file), 'foo') @unittest.skipIf(MACOS or SOLARIS or _winapi, 'On MACOS, Solaris and Windows the errors are not confusing (though different)') # gh-92670: The test uses a trailing slash to force the OS consider # the path as a directory, but on AIX the trailing slash has no effect # and is considered as a file. @unittest.skipIf(AIX, 'Not valid on AIX, see gh-92670') def test_copyfile_nonexistent_dir(self): # Issue 43219 src_dir = self.mkdtemp() src_file = os.path.join(src_dir, 'foo') dst = os.path.join(src_dir, 'does_not_exist/') write_file(src_file, 'foo') self.assertRaises(FileNotFoundError, shutil.copyfile, src_file, dst) def test_copyfile_copy_dir(self): # Issue 45234 # test copy() and copyfile() raising proper exceptions when src and/or # dst are directories src_dir = self.mkdtemp() src_file = os.path.join(src_dir, 'foo') dir2 = self.mkdtemp() dst = os.path.join(src_dir, 'does_not_exist/') write_file(src_file, 'foo') if sys.platform == "win32": err = PermissionError else: err = IsADirectoryError self.assertRaises(err, shutil.copyfile, src_dir, dst) self.assertRaises(err, shutil.copyfile, src_file, src_dir) self.assertRaises(err, shutil.copyfile, dir2, src_dir) class TestArchives(BaseTest, unittest.TestCase): ### shutil.make_archive @support.requires_zlib() def test_make_tarball(self): # creating something to tar root_dir, base_dir = self._create_files('') tmpdir2 = self.mkdtemp() # force shutil to create the directory os.rmdir(tmpdir2) # working with relative paths work_dir = os.path.dirname(tmpdir2) rel_base_name = os.path.join(os.path.basename(tmpdir2), 'archive') with os_helper.change_cwd(work_dir), no_chdir: base_name = os.path.abspath(rel_base_name) tarball = make_archive(rel_base_name, 'gztar', root_dir, '.') # check if the compressed tarball was created self.assertEqual(tarball, base_name + '.tar.gz') self.assertTrue(os.path.isfile(tarball)) self.assertTrue(tarfile.is_tarfile(tarball)) with tarfile.open(tarball, 'r:gz') as tf: self.assertCountEqual(tf.getnames(), ['.', './sub', './sub2', './file1', './file2', './sub/file3']) # trying an uncompressed one with os_helper.change_cwd(work_dir), no_chdir: tarball = make_archive(rel_base_name, 'tar', root_dir, '.') self.assertEqual(tarball, base_name + '.tar') self.assertTrue(os.path.isfile(tarball)) self.assertTrue(tarfile.is_tarfile(tarball)) with tarfile.open(tarball, 'r') as tf: self.assertCountEqual(tf.getnames(), ['.', './sub', './sub2', './file1', './file2', './sub/file3']) def _tarinfo(self, path): with tarfile.open(path) as tar: names = tar.getnames() names.sort() return tuple(names) def _create_files(self, base_dir='dist'): # creating something to tar root_dir = self.mkdtemp() dist = os.path.join(root_dir, base_dir) os.makedirs(dist, exist_ok=True) write_file((dist, 'file1'), 'xxx') write_file((dist, 'file2'), 'xxx') os.mkdir(os.path.join(dist, 'sub')) write_file((dist, 'sub', 'file3'), 'xxx') os.mkdir(os.path.join(dist, 'sub2')) if base_dir: write_file((root_dir, 'outer'), 'xxx') return root_dir, base_dir @support.requires_zlib() @unittest.skipUnless(shutil.which('tar'), 'Need the tar command to run') def test_tarfile_vs_tar(self): root_dir, base_dir = self._create_files() base_name = os.path.join(self.mkdtemp(), 'archive') with no_chdir: tarball = make_archive(base_name, 'gztar', root_dir, base_dir) # check if the compressed tarball was created self.assertEqual(tarball, base_name + '.tar.gz') self.assertTrue(os.path.isfile(tarball)) # now create another tarball using `tar` tarball2 = os.path.join(root_dir, 'archive2.tar') tar_cmd = ['tar', '-cf', 'archive2.tar', base_dir] subprocess.check_call(tar_cmd, cwd=root_dir, stdout=subprocess.DEVNULL) self.assertTrue(os.path.isfile(tarball2)) # let's compare both tarballs self.assertEqual(self._tarinfo(tarball), self._tarinfo(tarball2)) # trying an uncompressed one with no_chdir: tarball = make_archive(base_name, 'tar', root_dir, base_dir) self.assertEqual(tarball, base_name + '.tar') self.assertTrue(os.path.isfile(tarball)) # now for a dry_run with no_chdir: tarball = make_archive(base_name, 'tar', root_dir, base_dir, dry_run=True) self.assertEqual(tarball, base_name + '.tar') self.assertTrue(os.path.isfile(tarball)) @support.requires_zlib() def test_make_zipfile(self): # creating something to zip root_dir, base_dir = self._create_files() tmpdir2 = self.mkdtemp() # force shutil to create the directory os.rmdir(tmpdir2) # working with relative paths work_dir = os.path.dirname(tmpdir2) rel_base_name = os.path.join(os.path.basename(tmpdir2), 'archive') with os_helper.change_cwd(work_dir), no_chdir: base_name = os.path.abspath(rel_base_name) res = make_archive(rel_base_name, 'zip', root_dir) self.assertEqual(res, base_name + '.zip') self.assertTrue(os.path.isfile(res)) self.assertTrue(zipfile.is_zipfile(res)) with zipfile.ZipFile(res) as zf: self.assertCountEqual(zf.namelist(), ['dist/', 'dist/sub/', 'dist/sub2/', 'dist/file1', 'dist/file2', 'dist/sub/file3', 'outer']) with os_helper.change_cwd(work_dir), no_chdir: base_name = os.path.abspath(rel_base_name) res = make_archive(rel_base_name, 'zip', root_dir, base_dir) self.assertEqual(res, base_name + '.zip') self.assertTrue(os.path.isfile(res)) self.assertTrue(zipfile.is_zipfile(res)) with zipfile.ZipFile(res) as zf: self.assertCountEqual(zf.namelist(), ['dist/', 'dist/sub/', 'dist/sub2/', 'dist/file1', 'dist/file2', 'dist/sub/file3']) @support.requires_zlib() @unittest.skipUnless(shutil.which('zip'), 'Need the zip command to run') def test_zipfile_vs_zip(self): root_dir, base_dir = self._create_files() base_name = os.path.join(self.mkdtemp(), 'archive') with no_chdir: archive = make_archive(base_name, 'zip', root_dir, base_dir) # check if ZIP file was created self.assertEqual(archive, base_name + '.zip') self.assertTrue(os.path.isfile(archive)) # now create another ZIP file using `zip` archive2 = os.path.join(root_dir, 'archive2.zip') zip_cmd = ['zip', '-q', '-r', 'archive2.zip', base_dir] subprocess.check_call(zip_cmd, cwd=root_dir, stdout=subprocess.DEVNULL) self.assertTrue(os.path.isfile(archive2)) # let's compare both ZIP files with zipfile.ZipFile(archive) as zf: names = zf.namelist() with zipfile.ZipFile(archive2) as zf: names2 = zf.namelist() self.assertEqual(sorted(names), sorted(names2)) @support.requires_zlib() @unittest.skipUnless(shutil.which('unzip'), 'Need the unzip command to run') def test_unzip_zipfile(self): root_dir, base_dir = self._create_files() base_name = os.path.join(self.mkdtemp(), 'archive') with no_chdir: archive = make_archive(base_name, 'zip', root_dir, base_dir) # check if ZIP file was created self.assertEqual(archive, base_name + '.zip') self.assertTrue(os.path.isfile(archive)) # now check the ZIP file using `unzip -t` zip_cmd = ['unzip', '-t', archive] with os_helper.change_cwd(root_dir): try: subprocess.check_output(zip_cmd, stderr=subprocess.STDOUT) except subprocess.CalledProcessError as exc: details = exc.output.decode(errors="replace") if 'unrecognized option: t' in details: self.skipTest("unzip doesn't support -t") msg = "{}\n\n**Unzip Output**\n{}" self.fail(msg.format(exc, details)) def test_make_archive(self): tmpdir = self.mkdtemp() base_name = os.path.join(tmpdir, 'archive') self.assertRaises(ValueError, make_archive, base_name, 'xxx') @support.requires_zlib() def test_make_archive_owner_group(self): # testing make_archive with owner and group, with various combinations # this works even if there's not gid/uid support if UID_GID_SUPPORT: group = grp.getgrgid(0)[0] owner = pwd.getpwuid(0)[0] else: group = owner = 'root' root_dir, base_dir = self._create_files() base_name = os.path.join(self.mkdtemp(), 'archive') res = make_archive(base_name, 'zip', root_dir, base_dir, owner=owner, group=group) self.assertTrue(os.path.isfile(res)) res = make_archive(base_name, 'zip', root_dir, base_dir) self.assertTrue(os.path.isfile(res)) res = make_archive(base_name, 'tar', root_dir, base_dir, owner=owner, group=group) self.assertTrue(os.path.isfile(res)) res = make_archive(base_name, 'tar', root_dir, base_dir, owner='kjhkjhkjg', group='oihohoh') self.assertTrue(os.path.isfile(res)) @support.requires_zlib() @unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support") def test_tarfile_root_owner(self): root_dir, base_dir = self._create_files() base_name = os.path.join(self.mkdtemp(), 'archive') group = grp.getgrgid(0)[0] owner = pwd.getpwuid(0)[0] with os_helper.change_cwd(root_dir), no_chdir: archive_name = make_archive(base_name, 'gztar', root_dir, 'dist', owner=owner, group=group) # check if the compressed tarball was created self.assertTrue(os.path.isfile(archive_name)) # now checks the rights archive = tarfile.open(archive_name) try: for member in archive.getmembers(): self.assertEqual(member.uid, 0) self.assertEqual(member.gid, 0) finally: archive.close() def test_make_archive_cwd_default(self): current_dir = os.getcwd() def archiver(base_name, base_dir, **kw): self.assertNotIn('root_dir', kw) self.assertEqual(base_name, 'basename') self.assertEqual(os.getcwd(), current_dir) raise RuntimeError() register_archive_format('xxx', archiver, [], 'xxx file') try: with no_chdir: with self.assertRaises(RuntimeError): make_archive('basename', 'xxx') self.assertEqual(os.getcwd(), current_dir) finally: unregister_archive_format('xxx') def test_make_archive_cwd(self): current_dir = os.getcwd() root_dir = self.mkdtemp() def archiver(base_name, base_dir, **kw): self.assertNotIn('root_dir', kw) self.assertEqual(base_name, os.path.join(current_dir, 'basename')) self.assertEqual(os.getcwd(), root_dir) raise RuntimeError() dirs = [] def _chdir(path): dirs.append(path) orig_chdir(path) register_archive_format('xxx', archiver, [], 'xxx file') try: with support.swap_attr(os, 'chdir', _chdir) as orig_chdir: with self.assertRaises(RuntimeError): make_archive('basename', 'xxx', root_dir=root_dir) self.assertEqual(os.getcwd(), current_dir) self.assertEqual(dirs, [root_dir, current_dir]) finally: unregister_archive_format('xxx') def test_make_archive_cwd_supports_root_dir(self): current_dir = os.getcwd() root_dir = self.mkdtemp() def archiver(base_name, base_dir, **kw): self.assertEqual(base_name, 'basename') self.assertEqual(kw['root_dir'], root_dir) self.assertEqual(os.getcwd(), current_dir) raise RuntimeError() archiver.supports_root_dir = True register_archive_format('xxx', archiver, [], 'xxx file') try: with no_chdir: with self.assertRaises(RuntimeError): make_archive('basename', 'xxx', root_dir=root_dir) self.assertEqual(os.getcwd(), current_dir) finally: unregister_archive_format('xxx') def test_make_tarfile_in_curdir(self): # Issue #21280 root_dir = self.mkdtemp() with os_helper.change_cwd(root_dir), no_chdir: self.assertEqual(make_archive('test', 'tar'), 'test.tar') self.assertTrue(os.path.isfile('test.tar')) @support.requires_zlib() def test_make_zipfile_in_curdir(self): # Issue #21280 root_dir = self.mkdtemp() with os_helper.change_cwd(root_dir), no_chdir: self.assertEqual(make_archive('test', 'zip'), 'test.zip') self.assertTrue(os.path.isfile('test.zip')) def test_register_archive_format(self): self.assertRaises(TypeError, register_archive_format, 'xxx', 1) self.assertRaises(TypeError, register_archive_format, 'xxx', lambda: x, 1) self.assertRaises(TypeError, register_archive_format, 'xxx', lambda: x, [(1, 2), (1, 2, 3)]) register_archive_format('xxx', lambda: x, [(1, 2)], 'xxx file') formats = [name for name, params in get_archive_formats()] self.assertIn('xxx', formats) unregister_archive_format('xxx') formats = [name for name, params in get_archive_formats()] self.assertNotIn('xxx', formats) ### shutil.unpack_archive def check_unpack_archive(self, format): self.check_unpack_archive_with_converter(format, lambda path: path) self.check_unpack_archive_with_converter(format, pathlib.Path) self.check_unpack_archive_with_converter(format, FakePath) def check_unpack_archive_with_converter(self, format, converter): root_dir, base_dir = self._create_files() expected = rlistdir(root_dir) expected.remove('outer') base_name = os.path.join(self.mkdtemp(), 'archive') filename = make_archive(base_name, format, root_dir, base_dir) # let's try to unpack it now tmpdir2 = self.mkdtemp() unpack_archive(converter(filename), converter(tmpdir2)) self.assertEqual(rlistdir(tmpdir2), expected) # and again, this time with the format specified tmpdir3 = self.mkdtemp() unpack_archive(converter(filename), converter(tmpdir3), format=format) self.assertEqual(rlistdir(tmpdir3), expected) self.assertRaises(shutil.ReadError, unpack_archive, converter(TESTFN)) self.assertRaises(ValueError, unpack_archive, converter(TESTFN), format='xxx') def test_unpack_archive_tar(self): self.check_unpack_archive('tar') @support.requires_zlib() def test_unpack_archive_gztar(self): self.check_unpack_archive('gztar') @support.requires_bz2() def test_unpack_archive_bztar(self): self.check_unpack_archive('bztar') @support.requires_lzma() @unittest.skipIf(AIX and not _maxdataOK(), "AIX MAXDATA must be 0x20000000 or larger") def test_unpack_archive_xztar(self): self.check_unpack_archive('xztar') @support.requires_zlib() def test_unpack_archive_zip(self): self.check_unpack_archive('zip') def test_unpack_registry(self): formats = get_unpack_formats() def _boo(filename, extract_dir, extra): self.assertEqual(extra, 1) self.assertEqual(filename, 'stuff.boo') self.assertEqual(extract_dir, 'xx') register_unpack_format('Boo', ['.boo', '.b2'], _boo, [('extra', 1)]) unpack_archive('stuff.boo', 'xx') # trying to register a .boo unpacker again self.assertRaises(RegistryError, register_unpack_format, 'Boo2', ['.boo'], _boo) # should work now unregister_unpack_format('Boo') register_unpack_format('Boo2', ['.boo'], _boo) self.assertIn(('Boo2', ['.boo'], ''), get_unpack_formats()) self.assertNotIn(('Boo', ['.boo'], ''), get_unpack_formats()) # let's leave a clean state unregister_unpack_format('Boo2') self.assertEqual(get_unpack_formats(), formats) class TestMisc(BaseTest, unittest.TestCase): @unittest.skipUnless(hasattr(shutil, 'disk_usage'), "disk_usage not available on this platform") def test_disk_usage(self): usage = shutil.disk_usage(os.path.dirname(__file__)) for attr in ('total', 'used', 'free'): self.assertIsInstance(getattr(usage, attr), int) self.assertGreater(usage.total, 0) self.assertGreater(usage.used, 0) self.assertGreaterEqual(usage.free, 0) self.assertGreaterEqual(usage.total, usage.used) self.assertGreater(usage.total, usage.free) # bpo-32557: Check that disk_usage() also accepts a filename shutil.disk_usage(__file__) @unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support") @unittest.skipUnless(hasattr(os, 'chown'), 'requires os.chown') def test_chown(self): dirname = self.mkdtemp() filename = tempfile.mktemp(dir=dirname) write_file(filename, 'testing chown function') with self.assertRaises(ValueError): shutil.chown(filename) with self.assertRaises(LookupError): shutil.chown(filename, user='non-existing username') with self.assertRaises(LookupError): shutil.chown(filename, group='non-existing groupname') with self.assertRaises(TypeError): shutil.chown(filename, b'spam') with self.assertRaises(TypeError): shutil.chown(filename, 3.14) uid = os.getuid() gid = os.getgid() def check_chown(path, uid=None, gid=None): s = os.stat(filename) if uid is not None: self.assertEqual(uid, s.st_uid) if gid is not None: self.assertEqual(gid, s.st_gid) shutil.chown(filename, uid, gid) check_chown(filename, uid, gid) shutil.chown(filename, uid) check_chown(filename, uid) shutil.chown(filename, user=uid) check_chown(filename, uid) shutil.chown(filename, group=gid) check_chown(filename, gid=gid) shutil.chown(dirname, uid, gid) check_chown(dirname, uid, gid) shutil.chown(dirname, uid) check_chown(dirname, uid) shutil.chown(dirname, user=uid) check_chown(dirname, uid) shutil.chown(dirname, group=gid) check_chown(dirname, gid=gid) try: user = pwd.getpwuid(uid)[0] group = grp.getgrgid(gid)[0] except KeyError: # On some systems uid/gid cannot be resolved. pass else: shutil.chown(filename, user, group) check_chown(filename, uid, gid) shutil.chown(dirname, user, group) check_chown(dirname, uid, gid) class TestWhich(BaseTest, unittest.TestCase): def setUp(self): self.temp_dir = self.mkdtemp(prefix="Tmp") # Give the temp_file an ".exe" suffix for all. # It's needed on Windows and not harmful on other platforms. self.temp_file = tempfile.NamedTemporaryFile(dir=self.temp_dir, prefix="Tmp", suffix=".Exe") os.chmod(self.temp_file.name, stat.S_IXUSR) self.addCleanup(self.temp_file.close) self.dir, self.file = os.path.split(self.temp_file.name) self.env_path = self.dir self.curdir = os.curdir self.ext = ".EXE" def test_basic(self): # Given an EXE in a directory, it should be returned. rv = shutil.which(self.file, path=self.dir) self.assertEqual(rv, self.temp_file.name) def test_absolute_cmd(self): # When given the fully qualified path to an executable that exists, # it should be returned. rv = shutil.which(self.temp_file.name, path=self.temp_dir) self.assertEqual(rv, self.temp_file.name) def test_relative_cmd(self): # When given the relative path with a directory part to an executable # that exists, it should be returned. base_dir, tail_dir = os.path.split(self.dir) relpath = os.path.join(tail_dir, self.file) with os_helper.change_cwd(path=base_dir): rv = shutil.which(relpath, path=self.temp_dir) self.assertEqual(rv, relpath) # But it shouldn't be searched in PATH directories (issue #16957). with os_helper.change_cwd(path=self.dir): rv = shutil.which(relpath, path=base_dir) self.assertIsNone(rv) def test_cwd(self): # Issue #16957 base_dir = os.path.dirname(self.dir) with os_helper.change_cwd(path=self.dir): rv = shutil.which(self.file, path=base_dir) if sys.platform == "win32": # Windows: current directory implicitly on PATH self.assertEqual(rv, os.path.join(self.curdir, self.file)) else: # Other platforms: shouldn't match in the current directory. self.assertIsNone(rv) @os_helper.skip_if_dac_override def test_non_matching_mode(self): # Set the file read-only and ask for writeable files. os.chmod(self.temp_file.name, stat.S_IREAD) if os.access(self.temp_file.name, os.W_OK): self.skipTest("can't set the file read-only") rv = shutil.which(self.file, path=self.dir, mode=os.W_OK) self.assertIsNone(rv) def test_relative_path(self): base_dir, tail_dir = os.path.split(self.dir) with os_helper.change_cwd(path=base_dir): rv = shutil.which(self.file, path=tail_dir) self.assertEqual(rv, os.path.join(tail_dir, self.file)) def test_nonexistent_file(self): # Return None when no matching executable file is found on the path. rv = shutil.which("foo.exe", path=self.dir) self.assertIsNone(rv) @unittest.skipUnless(sys.platform == "win32", "pathext check is Windows-only") def test_pathext_checking(self): # Ask for the file without the ".exe" extension, then ensure that # it gets found properly with the extension. rv = shutil.which(self.file[:-4], path=self.dir) self.assertEqual(rv, self.temp_file.name[:-4] + self.ext) def test_environ_path(self): with os_helper.EnvironmentVarGuard() as env: env['PATH'] = self.env_path rv = shutil.which(self.file) self.assertEqual(rv, self.temp_file.name) def test_environ_path_empty(self): # PATH='': no match with os_helper.EnvironmentVarGuard() as env: env['PATH'] = '' with unittest.mock.patch('os.confstr', return_value=self.dir, \ create=True), \ support.swap_attr(os, 'defpath', self.dir), \ os_helper.change_cwd(self.dir): rv = shutil.which(self.file) self.assertIsNone(rv) def test_environ_path_cwd(self): expected_cwd = os.path.basename(self.temp_file.name) if sys.platform == "win32": curdir = os.curdir if isinstance(expected_cwd, bytes): curdir = os.fsencode(curdir) expected_cwd = os.path.join(curdir, expected_cwd) # PATH=':': explicitly looks in the current directory with os_helper.EnvironmentVarGuard() as env: env['PATH'] = os.pathsep with unittest.mock.patch('os.confstr', return_value=self.dir, \ create=True), \ support.swap_attr(os, 'defpath', self.dir): rv = shutil.which(self.file) self.assertIsNone(rv) # look in current directory with os_helper.change_cwd(self.dir): rv = shutil.which(self.file) self.assertEqual(rv, expected_cwd) def test_environ_path_missing(self): with os_helper.EnvironmentVarGuard() as env: env.pop('PATH', None) # without confstr with unittest.mock.patch('os.confstr', side_effect=ValueError, \ create=True), \ support.swap_attr(os, 'defpath', self.dir): rv = shutil.which(self.file) self.assertEqual(rv, self.temp_file.name) # with confstr with unittest.mock.patch('os.confstr', return_value=self.dir, \ create=True), \ support.swap_attr(os, 'defpath', ''): rv = shutil.which(self.file) self.assertEqual(rv, self.temp_file.name) def test_empty_path(self): base_dir = os.path.dirname(self.dir) with os_helper.change_cwd(path=self.dir), \ os_helper.EnvironmentVarGuard() as env: env['PATH'] = self.env_path rv = shutil.which(self.file, path='') self.assertIsNone(rv) def test_empty_path_no_PATH(self): with os_helper.EnvironmentVarGuard() as env: env.pop('PATH', None) rv = shutil.which(self.file) self.assertIsNone(rv) @unittest.skipUnless(sys.platform == "win32", 'test specific to Windows') def test_pathext(self): ext = ".xyz" temp_filexyz = tempfile.NamedTemporaryFile(dir=self.temp_dir, prefix="Tmp2", suffix=ext) os.chmod(temp_filexyz.name, stat.S_IXUSR) self.addCleanup(temp_filexyz.close) # strip path and extension program = os.path.basename(temp_filexyz.name) program = os.path.splitext(program)[0] with os_helper.EnvironmentVarGuard() as env: env['PATHEXT'] = ext rv = shutil.which(program, path=self.temp_dir) self.assertEqual(rv, temp_filexyz.name) # Issue 40592: See https://bugs.python.org/issue40592 @unittest.skipUnless(sys.platform == "win32", 'test specific to Windows') def test_pathext_with_empty_str(self): ext = ".xyz" temp_filexyz = tempfile.NamedTemporaryFile(dir=self.temp_dir, prefix="Tmp2", suffix=ext) self.addCleanup(temp_filexyz.close) # strip path and extension program = os.path.basename(temp_filexyz.name) program = os.path.splitext(program)[0] with os_helper.EnvironmentVarGuard() as env: env['PATHEXT'] = f"{ext};" # note the ; rv = shutil.which(program, path=self.temp_dir) self.assertEqual(rv, temp_filexyz.name) class TestWhichBytes(TestWhich): def setUp(self): TestWhich.setUp(self) self.dir = os.fsencode(self.dir) self.file = os.fsencode(self.file) self.temp_file.name = os.fsencode(self.temp_file.name) self.curdir = os.fsencode(self.curdir) self.ext = os.fsencode(self.ext) class TestMove(BaseTest, unittest.TestCase): def setUp(self): filename = "foo" self.src_dir = self.mkdtemp() self.dst_dir = self.mkdtemp() self.src_file = os.path.join(self.src_dir, filename) self.dst_file = os.path.join(self.dst_dir, filename) with open(self.src_file, "wb") as f: f.write(b"spam") def _check_move_file(self, src, dst, real_dst): with open(src, "rb") as f: contents = f.read() shutil.move(src, dst) with open(real_dst, "rb") as f: self.assertEqual(contents, f.read()) self.assertFalse(os.path.exists(src)) def _check_move_dir(self, src, dst, real_dst): contents = sorted(os.listdir(src)) shutil.move(src, dst) self.assertEqual(contents, sorted(os.listdir(real_dst))) self.assertFalse(os.path.exists(src)) def test_move_file(self): # Move a file to another location on the same filesystem. self._check_move_file(self.src_file, self.dst_file, self.dst_file) def test_move_file_to_dir(self): # Move a file inside an existing dir on the same filesystem. self._check_move_file(self.src_file, self.dst_dir, self.dst_file) def test_move_file_to_dir_pathlike_src(self): # Move a pathlike file to another location on the same filesystem. src = pathlib.Path(self.src_file) self._check_move_file(src, self.dst_dir, self.dst_file) def test_move_file_to_dir_pathlike_dst(self): # Move a file to another pathlike location on the same filesystem. dst = pathlib.Path(self.dst_dir) self._check_move_file(self.src_file, dst, self.dst_file) @mock_rename def test_move_file_other_fs(self): # Move a file to an existing dir on another filesystem. self.test_move_file() @mock_rename def test_move_file_to_dir_other_fs(self): # Move a file to another location on another filesystem. self.test_move_file_to_dir() def test_move_dir(self): # Move a dir to another location on the same filesystem. dst_dir = tempfile.mktemp(dir=self.mkdtemp()) try: self._check_move_dir(self.src_dir, dst_dir, dst_dir) finally: os_helper.rmtree(dst_dir) @mock_rename def test_move_dir_other_fs(self): # Move a dir to another location on another filesystem. self.test_move_dir() def test_move_dir_to_dir(self): # Move a dir inside an existing dir on the same filesystem. self._check_move_dir(self.src_dir, self.dst_dir, os.path.join(self.dst_dir, os.path.basename(self.src_dir))) @mock_rename def test_move_dir_to_dir_other_fs(self): # Move a dir inside an existing dir on another filesystem. self.test_move_dir_to_dir() def test_move_dir_sep_to_dir(self): self._check_move_dir(self.src_dir + os.path.sep, self.dst_dir, os.path.join(self.dst_dir, os.path.basename(self.src_dir))) @unittest.skipUnless(os.path.altsep, 'requires os.path.altsep') def test_move_dir_altsep_to_dir(self): self._check_move_dir(self.src_dir + os.path.altsep, self.dst_dir, os.path.join(self.dst_dir, os.path.basename(self.src_dir))) def test_existing_file_inside_dest_dir(self): # A file with the same name inside the destination dir already exists. with open(self.dst_file, "wb"): pass self.assertRaises(shutil.Error, shutil.move, self.src_file, self.dst_dir) def test_dont_move_dir_in_itself(self): # Moving a dir inside itself raises an Error. dst = os.path.join(self.src_dir, "bar") self.assertRaises(shutil.Error, shutil.move, self.src_dir, dst) def test_destinsrc_false_negative(self): os.mkdir(TESTFN) try: for src, dst in [('srcdir', 'srcdir/dest')]: src = os.path.join(TESTFN, src) dst = os.path.join(TESTFN, dst) self.assertTrue(shutil._destinsrc(src, dst), msg='_destinsrc() wrongly concluded that ' 'dst (%s) is not in src (%s)' % (dst, src)) finally: os_helper.rmtree(TESTFN) def test_destinsrc_false_positive(self): os.mkdir(TESTFN) try: for src, dst in [('srcdir', 'src/dest'), ('srcdir', 'srcdir.new')]: src = os.path.join(TESTFN, src) dst = os.path.join(TESTFN, dst) self.assertFalse(shutil._destinsrc(src, dst), msg='_destinsrc() wrongly concluded that ' 'dst (%s) is in src (%s)' % (dst, src)) finally: os_helper.rmtree(TESTFN) @os_helper.skip_unless_symlink @mock_rename def test_move_file_symlink(self): dst = os.path.join(self.src_dir, 'bar') os.symlink(self.src_file, dst) shutil.move(dst, self.dst_file) self.assertTrue(os.path.islink(self.dst_file)) self.assertTrue(os.path.samefile(self.src_file, self.dst_file)) @os_helper.skip_unless_symlink @mock_rename def test_move_file_symlink_to_dir(self): filename = "bar" dst = os.path.join(self.src_dir, filename) os.symlink(self.src_file, dst) shutil.move(dst, self.dst_dir) final_link = os.path.join(self.dst_dir, filename) self.assertTrue(os.path.islink(final_link)) self.assertTrue(os.path.samefile(self.src_file, final_link)) @os_helper.skip_unless_symlink @mock_rename def test_move_dangling_symlink(self): src = os.path.join(self.src_dir, 'baz') dst = os.path.join(self.src_dir, 'bar') os.symlink(src, dst) dst_link = os.path.join(self.dst_dir, 'quux') shutil.move(dst, dst_link) self.assertTrue(os.path.islink(dst_link)) self.assertEqual(os.path.realpath(src), os.path.realpath(dst_link)) @os_helper.skip_unless_symlink @mock_rename def test_move_dir_symlink(self): src = os.path.join(self.src_dir, 'baz') dst = os.path.join(self.src_dir, 'bar') os.mkdir(src) os.symlink(src, dst) dst_link = os.path.join(self.dst_dir, 'quux') shutil.move(dst, dst_link) self.assertTrue(os.path.islink(dst_link)) self.assertTrue(os.path.samefile(src, dst_link)) def test_move_return_value(self): rv = shutil.move(self.src_file, self.dst_dir) self.assertEqual(rv, os.path.join(self.dst_dir, os.path.basename(self.src_file))) def test_move_as_rename_return_value(self): rv = shutil.move(self.src_file, os.path.join(self.dst_dir, 'bar')) self.assertEqual(rv, os.path.join(self.dst_dir, 'bar')) @mock_rename def test_move_file_special_function(self): moved = [] def _copy(src, dst): moved.append((src, dst)) shutil.move(self.src_file, self.dst_dir, copy_function=_copy) self.assertEqual(len(moved), 1) @mock_rename def test_move_dir_special_function(self): moved = [] def _copy(src, dst): moved.append((src, dst)) os_helper.create_empty_file(os.path.join(self.src_dir, 'child')) os_helper.create_empty_file(os.path.join(self.src_dir, 'child1')) shutil.move(self.src_dir, self.dst_dir, copy_function=_copy) self.assertEqual(len(moved), 3) def test_move_dir_caseinsensitive(self): # Renames a folder to the same name # but a different case. self.src_dir = self.mkdtemp() dst_dir = os.path.join( os.path.dirname(self.src_dir), os.path.basename(self.src_dir).upper()) self.assertNotEqual(self.src_dir, dst_dir) try: shutil.move(self.src_dir, dst_dir) self.assertTrue(os.path.isdir(dst_dir)) finally: os.rmdir(dst_dir) @os_helper.skip_unless_dac_override @unittest.skipUnless(hasattr(os, 'lchflags') and hasattr(stat, 'SF_IMMUTABLE') and hasattr(stat, 'UF_OPAQUE'), 'requires lchflags') def test_move_dir_permission_denied(self): # bpo-42782: shutil.move should not create destination directories # if the source directory cannot be removed. try: os.mkdir(TESTFN_SRC) os.lchflags(TESTFN_SRC, stat.SF_IMMUTABLE) # Testing on an empty immutable directory # TESTFN_DST should not exist if shutil.move failed self.assertRaises(PermissionError, shutil.move, TESTFN_SRC, TESTFN_DST) self.assertFalse(TESTFN_DST in os.listdir()) # Create a file and keep the directory immutable os.lchflags(TESTFN_SRC, stat.UF_OPAQUE) os_helper.create_empty_file(os.path.join(TESTFN_SRC, 'child')) os.lchflags(TESTFN_SRC, stat.SF_IMMUTABLE) # Testing on a non-empty immutable directory # TESTFN_DST should not exist if shutil.move failed self.assertRaises(PermissionError, shutil.move, TESTFN_SRC, TESTFN_DST) self.assertFalse(TESTFN_DST in os.listdir()) finally: if os.path.exists(TESTFN_SRC): os.lchflags(TESTFN_SRC, stat.UF_OPAQUE) os_helper.rmtree(TESTFN_SRC) if os.path.exists(TESTFN_DST): os.lchflags(TESTFN_DST, stat.UF_OPAQUE) os_helper.rmtree(TESTFN_DST) class TestCopyFile(unittest.TestCase): class Faux(object): _entered = False _exited_with = None _raised = False def __init__(self, raise_in_exit=False, suppress_at_exit=True): self._raise_in_exit = raise_in_exit self._suppress_at_exit = suppress_at_exit def read(self, *args): return '' def __enter__(self): self._entered = True def __exit__(self, exc_type, exc_val, exc_tb): self._exited_with = exc_type, exc_val, exc_tb if self._raise_in_exit: self._raised = True raise OSError("Cannot close") return self._suppress_at_exit def test_w_source_open_fails(self): def _open(filename, mode='r'): if filename == 'srcfile': raise OSError('Cannot open "srcfile"') assert 0 # shouldn't reach here. with support.swap_attr(shutil, 'open', _open): with self.assertRaises(OSError): shutil.copyfile('srcfile', 'destfile') @unittest.skipIf(MACOS, "skipped on macOS") def test_w_dest_open_fails(self): srcfile = self.Faux() def _open(filename, mode='r'): if filename == 'srcfile': return srcfile if filename == 'destfile': raise OSError('Cannot open "destfile"') assert 0 # shouldn't reach here. with support.swap_attr(shutil, 'open', _open): shutil.copyfile('srcfile', 'destfile') self.assertTrue(srcfile._entered) self.assertTrue(srcfile._exited_with[0] is OSError) self.assertEqual(srcfile._exited_with[1].args, ('Cannot open "destfile"',)) @unittest.skipIf(MACOS, "skipped on macOS") def test_w_dest_close_fails(self): srcfile = self.Faux() destfile = self.Faux(True) def _open(filename, mode='r'): if filename == 'srcfile': return srcfile if filename == 'destfile': return destfile assert 0 # shouldn't reach here. with support.swap_attr(shutil, 'open', _open): shutil.copyfile('srcfile', 'destfile') self.assertTrue(srcfile._entered) self.assertTrue(destfile._entered) self.assertTrue(destfile._raised) self.assertTrue(srcfile._exited_with[0] is OSError) self.assertEqual(srcfile._exited_with[1].args, ('Cannot close',)) @unittest.skipIf(MACOS, "skipped on macOS") def test_w_source_close_fails(self): srcfile = self.Faux(True) destfile = self.Faux() def _open(filename, mode='r'): if filename == 'srcfile': return srcfile if filename == 'destfile': return destfile assert 0 # shouldn't reach here. with support.swap_attr(shutil, 'open', _open): with self.assertRaises(OSError): shutil.copyfile('srcfile', 'destfile') self.assertTrue(srcfile._entered) self.assertTrue(destfile._entered) self.assertFalse(destfile._raised) self.assertTrue(srcfile._exited_with[0] is None) self.assertTrue(srcfile._raised) class TestCopyFileObj(unittest.TestCase): FILESIZE = 2 * 1024 * 1024 @classmethod def setUpClass(cls): write_test_file(TESTFN, cls.FILESIZE) @classmethod def tearDownClass(cls): os_helper.unlink(TESTFN) os_helper.unlink(TESTFN2) def tearDown(self): os_helper.unlink(TESTFN2) @contextlib.contextmanager def get_files(self): with open(TESTFN, "rb") as src: with open(TESTFN2, "wb") as dst: yield (src, dst) def assert_files_eq(self, src, dst): with open(src, 'rb') as fsrc: with open(dst, 'rb') as fdst: self.assertEqual(fsrc.read(), fdst.read()) def test_content(self): with self.get_files() as (src, dst): shutil.copyfileobj(src, dst) self.assert_files_eq(TESTFN, TESTFN2) def test_file_not_closed(self): with self.get_files() as (src, dst): shutil.copyfileobj(src, dst) assert not src.closed assert not dst.closed def test_file_offset(self): with self.get_files() as (src, dst): shutil.copyfileobj(src, dst) self.assertEqual(src.tell(), self.FILESIZE) self.assertEqual(dst.tell(), self.FILESIZE) @unittest.skipIf(os.name != 'nt', "Windows only") def test_win_impl(self): # Make sure alternate Windows implementation is called. with unittest.mock.patch("shutil._copyfileobj_readinto") as m: shutil.copyfile(TESTFN, TESTFN2) assert m.called # File size is 2 MiB but max buf size should be 1 MiB. self.assertEqual(m.call_args[0][2], 1 * 1024 * 1024) # If file size < 1 MiB memoryview() length must be equal to # the actual file size. with tempfile.NamedTemporaryFile(dir=os.getcwd(), delete=False) as f: f.write(b'foo') fname = f.name self.addCleanup(os_helper.unlink, fname) with unittest.mock.patch("shutil._copyfileobj_readinto") as m: shutil.copyfile(fname, TESTFN2) self.assertEqual(m.call_args[0][2], 3) # Empty files should not rely on readinto() variant. with tempfile.NamedTemporaryFile(dir=os.getcwd(), delete=False) as f: pass fname = f.name self.addCleanup(os_helper.unlink, fname) with unittest.mock.patch("shutil._copyfileobj_readinto") as m: shutil.copyfile(fname, TESTFN2) assert not m.called self.assert_files_eq(fname, TESTFN2) class _ZeroCopyFileTest(object): """Tests common to all zero-copy APIs.""" FILESIZE = (10 * 1024 * 1024) # 10 MiB FILEDATA = b"" PATCHPOINT = "" @classmethod def setUpClass(cls): write_test_file(TESTFN, cls.FILESIZE) with open(TESTFN, 'rb') as f: cls.FILEDATA = f.read() assert len(cls.FILEDATA) == cls.FILESIZE @classmethod def tearDownClass(cls): os_helper.unlink(TESTFN) def tearDown(self): os_helper.unlink(TESTFN2) @contextlib.contextmanager def get_files(self): with open(TESTFN, "rb") as src: with open(TESTFN2, "wb") as dst: yield (src, dst) def zerocopy_fun(self, *args, **kwargs): raise NotImplementedError("must be implemented in subclass") def reset(self): self.tearDown() self.tearDownClass() self.setUpClass() self.setUp() # --- def test_regular_copy(self): with self.get_files() as (src, dst): self.zerocopy_fun(src, dst) self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) # Make sure the fallback function is not called. with self.get_files() as (src, dst): with unittest.mock.patch('shutil.copyfileobj') as m: shutil.copyfile(TESTFN, TESTFN2) assert not m.called def test_same_file(self): self.addCleanup(self.reset) with self.get_files() as (src, dst): with self.assertRaises(Exception): self.zerocopy_fun(src, src) # Make sure src file is not corrupted. self.assertEqual(read_file(TESTFN, binary=True), self.FILEDATA) def test_non_existent_src(self): name = tempfile.mktemp(dir=os.getcwd()) with self.assertRaises(FileNotFoundError) as cm: shutil.copyfile(name, "new") self.assertEqual(cm.exception.filename, name) def test_empty_file(self): srcname = TESTFN + 'src' dstname = TESTFN + 'dst' self.addCleanup(lambda: os_helper.unlink(srcname)) self.addCleanup(lambda: os_helper.unlink(dstname)) with open(srcname, "wb"): pass with open(srcname, "rb") as src: with open(dstname, "wb") as dst: self.zerocopy_fun(src, dst) self.assertEqual(read_file(dstname, binary=True), b"") def test_unhandled_exception(self): with unittest.mock.patch(self.PATCHPOINT, side_effect=ZeroDivisionError): self.assertRaises(ZeroDivisionError, shutil.copyfile, TESTFN, TESTFN2) def test_exception_on_first_call(self): # Emulate a case where the first call to the zero-copy # function raises an exception in which case the function is # supposed to give up immediately. with unittest.mock.patch(self.PATCHPOINT, side_effect=OSError(errno.EINVAL, "yo")): with self.get_files() as (src, dst): with self.assertRaises(_GiveupOnFastCopy): self.zerocopy_fun(src, dst) def test_filesystem_full(self): # Emulate a case where filesystem is full and sendfile() fails # on first call. with unittest.mock.patch(self.PATCHPOINT, side_effect=OSError(errno.ENOSPC, "yo")): with self.get_files() as (src, dst): self.assertRaises(OSError, self.zerocopy_fun, src, dst) @unittest.skipIf(not SUPPORTS_SENDFILE, 'os.sendfile() not supported') class TestZeroCopySendfile(_ZeroCopyFileTest, unittest.TestCase): PATCHPOINT = "os.sendfile" def zerocopy_fun(self, fsrc, fdst): return shutil._fastcopy_sendfile(fsrc, fdst) def test_non_regular_file_src(self): with io.BytesIO(self.FILEDATA) as src: with open(TESTFN2, "wb") as dst: with self.assertRaises(_GiveupOnFastCopy): self.zerocopy_fun(src, dst) shutil.copyfileobj(src, dst) self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_non_regular_file_dst(self): with open(TESTFN, "rb") as src: with io.BytesIO() as dst: with self.assertRaises(_GiveupOnFastCopy): self.zerocopy_fun(src, dst) shutil.copyfileobj(src, dst) dst.seek(0) self.assertEqual(dst.read(), self.FILEDATA) def test_exception_on_second_call(self): def sendfile(*args, **kwargs): if not flag: flag.append(None) return orig_sendfile(*args, **kwargs) else: raise OSError(errno.EBADF, "yo") flag = [] orig_sendfile = os.sendfile with unittest.mock.patch('os.sendfile', create=True, side_effect=sendfile): with self.get_files() as (src, dst): with self.assertRaises(OSError) as cm: shutil._fastcopy_sendfile(src, dst) assert flag self.assertEqual(cm.exception.errno, errno.EBADF) def test_cant_get_size(self): # Emulate a case where src file size cannot be determined. # Internally bufsize will be set to a small value and # sendfile() will be called repeatedly. with unittest.mock.patch('os.fstat', side_effect=OSError) as m: with self.get_files() as (src, dst): shutil._fastcopy_sendfile(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_small_chunks(self): # Force internal file size detection to be smaller than the # actual file size. We want to force sendfile() to be called # multiple times, also in order to emulate a src fd which gets # bigger while it is being copied. mock = unittest.mock.Mock() mock.st_size = 65536 + 1 with unittest.mock.patch('os.fstat', return_value=mock) as m: with self.get_files() as (src, dst): shutil._fastcopy_sendfile(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_big_chunk(self): # Force internal file size detection to be +100MB bigger than # the actual file size. Make sure sendfile() does not rely on # file size value except for (maybe) a better throughput / # performance. mock = unittest.mock.Mock() mock.st_size = self.FILESIZE + (100 * 1024 * 1024) with unittest.mock.patch('os.fstat', return_value=mock) as m: with self.get_files() as (src, dst): shutil._fastcopy_sendfile(src, dst) assert m.called self.assertEqual(read_file(TESTFN2, binary=True), self.FILEDATA) def test_blocksize_arg(self): with unittest.mock.patch('os.sendfile', side_effect=ZeroDivisionError) as m: self.assertRaises(ZeroDivisionError, shutil.copyfile, TESTFN, TESTFN2) blocksize = m.call_args[0][3] # Make sure file size and the block size arg passed to # sendfile() are the same. self.assertEqual(blocksize, os.path.getsize(TESTFN)) # ...unless we're dealing with a small file. os_helper.unlink(TESTFN2) write_file(TESTFN2, b"hello", binary=True) self.addCleanup(os_helper.unlink, TESTFN2 + '3') self.assertRaises(ZeroDivisionError, shutil.copyfile, TESTFN2, TESTFN2 + '3') blocksize = m.call_args[0][3] self.assertEqual(blocksize, 2 ** 23) def test_file2file_not_supported(self): # Emulate a case where sendfile() only support file->socket # fds. In such a case copyfile() is supposed to skip the # fast-copy attempt from then on. assert shutil._USE_CP_SENDFILE try: with unittest.mock.patch( self.PATCHPOINT, side_effect=OSError(errno.ENOTSOCK, "yo")) as m: with self.get_files() as (src, dst): with self.assertRaises(_GiveupOnFastCopy): shutil._fastcopy_sendfile(src, dst) assert m.called assert not shutil._USE_CP_SENDFILE with unittest.mock.patch(self.PATCHPOINT) as m: shutil.copyfile(TESTFN, TESTFN2) assert not m.called finally: shutil._USE_CP_SENDFILE = True @unittest.skipIf(not MACOS, 'macOS only') class TestZeroCopyMACOS(_ZeroCopyFileTest, unittest.TestCase): PATCHPOINT = "posix._fcopyfile" def zerocopy_fun(self, src, dst): return shutil._fastcopy_fcopyfile(src, dst, posix._COPYFILE_DATA) class TestGetTerminalSize(unittest.TestCase): def test_does_not_crash(self): """Check if get_terminal_size() returns a meaningful value. There's no easy portable way to actually check the size of the terminal, so let's check if it returns something sensible instead. """ size = shutil.get_terminal_size() self.assertGreaterEqual(size.columns, 0) self.assertGreaterEqual(size.lines, 0) def test_os_environ_first(self): "Check if environment variables have precedence" with os_helper.EnvironmentVarGuard() as env: env['COLUMNS'] = '777' del env['LINES'] size = shutil.get_terminal_size() self.assertEqual(size.columns, 777) with os_helper.EnvironmentVarGuard() as env: del env['COLUMNS'] env['LINES'] = '888' size = shutil.get_terminal_size() self.assertEqual(size.lines, 888) def test_bad_environ(self): with os_helper.EnvironmentVarGuard() as env: env['COLUMNS'] = 'xxx' env['LINES'] = 'yyy' size = shutil.get_terminal_size() self.assertGreaterEqual(size.columns, 0) self.assertGreaterEqual(size.lines, 0) @unittest.skipUnless(os.isatty(sys.__stdout__.fileno()), "not on tty") @unittest.skipUnless(hasattr(os, 'get_terminal_size'), 'need os.get_terminal_size()') def test_stty_match(self): """Check if stty returns the same results ignoring env This test will fail if stdin and stdout are connected to different terminals with different sizes. Nevertheless, such situations should be pretty rare. """ try: size = subprocess.check_output(['stty', 'size']).decode().split() except (FileNotFoundError, PermissionError, subprocess.CalledProcessError): self.skipTest("stty invocation failed") expected = (int(size[1]), int(size[0])) # reversed order with os_helper.EnvironmentVarGuard() as env: del env['LINES'] del env['COLUMNS'] actual = shutil.get_terminal_size() self.assertEqual(expected, actual) @unittest.skipIf(support.is_wasi, "WASI has no /dev/null") def test_fallback(self): with os_helper.EnvironmentVarGuard() as env: del env['LINES'] del env['COLUMNS'] # sys.__stdout__ has no fileno() with support.swap_attr(sys, '__stdout__', None): size = shutil.get_terminal_size(fallback=(10, 20)) self.assertEqual(size.columns, 10) self.assertEqual(size.lines, 20) # sys.__stdout__ is not a terminal on Unix # or fileno() not in (0, 1, 2) on Windows with open(os.devnull, 'w', encoding='utf-8') as f, \ support.swap_attr(sys, '__stdout__', f): size = shutil.get_terminal_size(fallback=(30, 40)) self.assertEqual(size.columns, 30) self.assertEqual(size.lines, 40) class PublicAPITests(unittest.TestCase): """Ensures that the correct values are exposed in the public API.""" def test_module_all_attribute(self): self.assertTrue(hasattr(shutil, '__all__')) target_api = ['copyfileobj', 'copyfile', 'copymode', 'copystat', 'copy', 'copy2', 'copytree', 'move', 'rmtree', 'Error', 'SpecialFileError', 'ExecError', 'make_archive', 'get_archive_formats', 'register_archive_format', 'unregister_archive_format', 'get_unpack_formats', 'register_unpack_format', 'unregister_unpack_format', 'unpack_archive', 'ignore_patterns', 'chown', 'which', 'get_terminal_size', 'SameFileError'] if hasattr(os, 'statvfs') or os.name == 'nt': target_api.append('disk_usage') self.assertEqual(set(shutil.__all__), set(target_api)) if __name__ == '__main__': unittest.main()