[Python-checkins] bpo-38223: Reorganize test_shutil. (GH-16281)
Serhiy Storchaka
webhook-mailer at python.org
Thu Sep 26 06:15:13 EDT 2019
https://github.com/python/cpython/commit/4f2eac04e4fee4d7d188ad2eeb0e610e3d7254bb
commit: 4f2eac04e4fee4d7d188ad2eeb0e610e3d7254bb
branch: master
author: Serhiy Storchaka <storchaka at gmail.com>
committer: GitHub <noreply at github.com>
date: 2019-09-26T13:15:08+03:00
summary:
bpo-38223: Reorganize test_shutil. (GH-16281)
* Group tests for specific functions and groups of related functions
into separate classes.
* Clean up creating and cleaning up temporary directories.
* Simplify and make more robust monkey patching of shutil.open.
files:
M Lib/test/test_shutil.py
diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py
index 636e3bd97955..ab0f96d25135 100644
--- a/Lib/test/test_shutil.py
+++ b/Lib/test/test_shutil.py
@@ -159,18 +159,8 @@ def _maxdataOK():
else:
return True
-class TestShutil(unittest.TestCase):
-
- def setUp(self):
- super(TestShutil, self).setUp()
- self.tempdirs = []
-
- def tearDown(self):
- super(TestShutil, self).tearDown()
- while self.tempdirs:
- d = self.tempdirs.pop()
- shutil.rmtree(d, os.name in ('nt', 'cygwin'))
+class BaseTest:
def mkdtemp(self):
"""Create a temporary directory that will be cleaned up.
@@ -178,9 +168,12 @@ def mkdtemp(self):
Returns the path of the directory.
"""
d = tempfile.mkdtemp()
- self.tempdirs.append(d)
+ self.addCleanup(support.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')
@@ -382,141 +375,500 @@ def raiser(fn, *args, **kwargs):
finally:
os.lstat = orig_lstat
- @support.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)
+ 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(hasattr(os, 'lchmod'), 'requires os.lchmod')
- @support.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)
+ def test_rmtree_dont_delete_file(self):
+ # When called on a file instead of a directory, don't delete it.
+ handle, path = tempfile.mkstemp()
+ os.close(handle)
+ self.assertRaises(NotADirectoryError, shutil.rmtree, path)
+ os.remove(path)
- @unittest.skipIf(hasattr(os, 'lchmod'), 'requires os.lchmod to be missing')
@support.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
+ 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 = tempfile.mkdtemp()
+ dst_dir = os.path.join(tempfile.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 = tempfile.mkdtemp()
+ dst_dir = tempfile.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)
@support.skip_unless_symlink
- def test_copystat_symlinks(self):
+ def test_copytree_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)
+ 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_IRWXO)
+ 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_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)
+ 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(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()
+ 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 _chflags_raiser(path, flags, *, follow_symlinks=True):
- ex.errno = err
- raise ex
- return _chflags_raiser
+ def test_copytree_with_exclude(self):
+ # creating data
+ join = os.path.join
+ exists = os.path.exists
+ src_dir = tempfile.mkdtemp()
+ try:
+ dst_dir = join(tempfile.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_retains_permissions(self):
+ tmp_dir = tempfile.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)
+ 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 = tempfile.mkdtemp()
+ dst_dir = os.path.join(tempfile.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 = tempfile.mkdtemp()
+ self.addCleanup(support.rmtree, src)
+ dst = tempfile.mktemp()
+ self.addCleanup(support.rmtree, dst)
+ with open(os.path.join(src, 'foo'), 'w') 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()')
+ @support.skip_unless_symlink
+ 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)
+
+ @support.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))
+
+ @support.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'), 'w'):
+ 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))
+
+
+class TestCopy(BaseTest, unittest.TestCase):
+
+ ### shutil.copymode
+
+ @support.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')
+ @support.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')
+ @support.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
+
+ @support.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:
@@ -528,6 +880,8 @@ def _chflags_raiser(path, flags, *, follow_symlinks=True):
finally:
os.chflags = old_chflags
+ ### shutil.copyxattr
+
@support.skip_unless_xattr
def test_copyxattr(self):
tmp_dir = self.mkdtemp()
@@ -617,6 +971,24 @@ def test_copyxattr_symlinks(self):
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)
+
@support.skip_unless_symlink
def test_copy_symlinks(self):
tmp_dir = self.mkdtemp()
@@ -640,6 +1012,25 @@ def test_copy_symlinks(self):
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'))
+
@support.skip_unless_symlink
def test_copy2_symlinks(self):
tmp_dir = self.mkdtemp()
@@ -688,6 +1079,20 @@ def test_copy2_xattr(self):
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'))
+
+ ### shutil.copyfile
+
@support.skip_unless_symlink
def test_copyfile_symlinks(self):
tmp_dir = self.mkdtemp()
@@ -700,241 +1105,10 @@ def test_copyfile_symlinks(self):
# 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))
-
- 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)
-
- def test_rmtree_dont_delete_file(self):
- # When called on a file instead of a directory, don't delete it.
- handle, path = tempfile.mkstemp()
- os.close(handle)
- self.assertRaises(NotADirectoryError, shutil.rmtree, path)
- os.remove(path)
-
- def test_copytree_simple(self):
- src_dir = tempfile.mkdtemp()
- dst_dir = os.path.join(tempfile.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 = tempfile.mkdtemp()
- dst_dir = tempfile.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)
-
- @support.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 = tempfile.mkdtemp()
- try:
- dst_dir = join(tempfile.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_retains_permissions(self):
- tmp_dir = tempfile.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)
- 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 = tempfile.mkdtemp()
- dst_dir = os.path.join(tempfile.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 = tempfile.mkdtemp()
- self.addCleanup(support.rmtree, src)
- dst = tempfile.mktemp()
- self.addCleanup(support.rmtree, dst)
- with open(os.path.join(src, 'foo'), 'w') as f:
- f.close()
- shutil.copytree(src, dst, copy_function=custom_cpfun)
- self.assertEqual(len(flag), 1)
+ 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):
@@ -976,34 +1150,6 @@ def test_dont_copy_file_onto_symlink_to_itself(self):
finally:
shutil.rmtree(TESTFN, ignore_errors=True)
- @support.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)
-
# Issue #3002: copyfile and copytree block indefinitely on named pipes
@unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
def test_copyfile_named_pipe(self):
@@ -1019,119 +1165,33 @@ def test_copyfile_named_pipe(self):
finally:
os.remove(TESTFN)
- @unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
- @support.skip_unless_symlink
- 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)
-
- @support.skip_unless_symlink
- def test_copytree_dangling_symlinks(self):
-
- # a dangling symlink raises an error at the end
+ def test_copyfile_return_value(self):
+ # copytree returns its destination path.
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))
+ 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))
- @support.skip_unless_symlink
- def test_copytree_symlink_dir(self):
+ def test_copyfile_same_file(self):
+ # copyfile() should raise SameFileError if the source and destination
+ # are the same.
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'), 'w'):
- 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')))
+ 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')
- 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)
+class TestArchives(BaseTest, unittest.TestCase):
- @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'))
+ ### shutil.make_archive
@support.requires_zlib
def test_make_tarball(self):
@@ -1411,6 +1471,8 @@ def test_register_archive_format(self):
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)
@@ -1483,6 +1545,9 @@ def _boo(filename, extract_dir, extra):
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):
@@ -1501,8 +1566,6 @@ def test_disk_usage(self):
@unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support")
@unittest.skipUnless(hasattr(os, 'chown'), 'requires os.chown')
def test_chown(self):
-
- # cleaned-up automatically by TestShutil.tearDown method
dirname = self.mkdtemp()
filename = tempfile.mktemp(dir=dirname)
write_file(filename, 'testing chown function')
@@ -1557,57 +1620,12 @@ def check_chown(path, uid=None, gid=None):
shutil.chown(dirname, user, group)
check_chown(dirname, uid, gid)
- 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_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')
-
- 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))
-
-class TestWhich(unittest.TestCase):
+class TestWhich(BaseTest, unittest.TestCase):
def setUp(self):
self.temp_dir = tempfile.mkdtemp(prefix="Tmp")
- self.addCleanup(shutil.rmtree, self.temp_dir, True)
+ self.addCleanup(support.rmtree, self.temp_dir)
# 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,
@@ -1784,25 +1802,17 @@ def setUp(self):
self.ext = os.fsencode(self.ext)
-class TestMove(unittest.TestCase):
+class TestMove(BaseTest, unittest.TestCase):
def setUp(self):
filename = "foo"
- self.src_dir = tempfile.mkdtemp()
- self.dst_dir = tempfile.mkdtemp()
+ 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 tearDown(self):
- for d in (self.src_dir, self.dst_dir):
- try:
- if d:
- shutil.rmtree(d)
- except:
- pass
-
def _check_move_file(self, src, dst, real_dst):
with open(src, "rb") as f:
contents = f.read()
@@ -1841,10 +1851,7 @@ def test_move_dir(self):
try:
self._check_move_dir(self.src_dir, dst_dir, dst_dir)
finally:
- try:
- shutil.rmtree(dst_dir)
- except:
- pass
+ support.rmtree(dst_dir)
@mock_rename
def test_move_dir_other_fs(self):
@@ -1891,7 +1898,7 @@ def test_destinsrc_false_negative(self):
msg='_destinsrc() wrongly concluded that '
'dst (%s) is not in src (%s)' % (dst, src))
finally:
- shutil.rmtree(TESTFN, ignore_errors=True)
+ support.rmtree(TESTFN)
def test_destinsrc_false_positive(self):
os.mkdir(TESTFN)
@@ -1903,7 +1910,7 @@ def test_destinsrc_false_positive(self):
msg='_destinsrc() wrongly concluded that '
'dst (%s) is in src (%s)' % (dst, src))
finally:
- shutil.rmtree(TESTFN, ignore_errors=True)
+ support.rmtree(TESTFN)
@support.skip_unless_symlink
@mock_rename
@@ -1975,10 +1982,24 @@ def _copy(src, dst):
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.
-class TestCopyFile(unittest.TestCase):
+ 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)
- _delete = False
+ try:
+ shutil.move(self.src_dir, dst_dir)
+ self.assertTrue(os.path.isdir(dst_dir))
+ finally:
+ os.rmdir(dst_dir)
+
+
+class TestCopyFile(unittest.TestCase):
class Faux(object):
_entered = False
@@ -1998,27 +2019,18 @@ def __exit__(self, exc_type, exc_val, exc_tb):
raise OSError("Cannot close")
return self._suppress_at_exit
- def tearDown(self):
- if self._delete:
- del shutil.open
-
- def _set_shutil_open(self, func):
- shutil.open = func
- self._delete = True
-
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.
- self._set_shutil_open(_open)
-
- self.assertRaises(OSError, shutil.copyfile, 'srcfile', 'destfile')
+ 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'):
@@ -2028,9 +2040,8 @@ def _open(filename, mode='r'):
raise OSError('Cannot open "destfile"')
assert 0 # shouldn't reach here.
- self._set_shutil_open(_open)
-
- shutil.copyfile('srcfile', 'destfile')
+ 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,
@@ -2038,7 +2049,6 @@ def _open(filename, mode='r'):
@unittest.skipIf(MACOS, "skipped on macOS")
def test_w_dest_close_fails(self):
-
srcfile = self.Faux()
destfile = self.Faux(True)
@@ -2049,9 +2059,8 @@ def _open(filename, mode='r'):
return destfile
assert 0 # shouldn't reach here.
- self._set_shutil_open(_open)
-
- shutil.copyfile('srcfile', 'destfile')
+ with support.swap_attr(shutil, 'open', _open):
+ shutil.copyfile('srcfile', 'destfile')
self.assertTrue(srcfile._entered)
self.assertTrue(destfile._entered)
self.assertTrue(destfile._raised)
@@ -2072,33 +2081,15 @@ def _open(filename, mode='r'):
return destfile
assert 0 # shouldn't reach here.
- self._set_shutil_open(_open)
-
- self.assertRaises(OSError,
- shutil.copyfile, 'srcfile', 'destfile')
+ 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)
- def test_move_dir_caseinsensitive(self):
- # Renames a folder to the same name
- # but a different case.
-
- self.src_dir = tempfile.mkdtemp()
- self.addCleanup(shutil.rmtree, self.src_dir, True)
- 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)
-
class TestCopyFileObj(unittest.TestCase):
FILESIZE = 2 * 1024 * 1024
@@ -2401,7 +2392,7 @@ def zerocopy_fun(self, src, dst):
return shutil._fastcopy_fcopyfile(src, dst, posix._COPYFILE_DATA)
-class TermsizeTests(unittest.TestCase):
+class TestGetTerminalSize(unittest.TestCase):
def test_does_not_crash(self):
"""Check if get_terminal_size() returns a meaningful value.
More information about the Python-checkins
mailing list