diff options
Diffstat (limited to 'Lib/test/test_shutil.py')
| -rw-r--r-- | Lib/test/test_shutil.py | 511 |
1 files changed, 452 insertions, 59 deletions
diff --git a/Lib/test/test_shutil.py b/Lib/test/test_shutil.py index becd91c..30d9e07 100644 --- a/Lib/test/test_shutil.py +++ b/Lib/test/test_shutil.py @@ -9,9 +9,80 @@ import os import os.path from test import support from test.support import TESTFN +from os.path import splitdrive +from distutils.spawn import find_executable, spawn +from shutil import (_make_tarball, _make_zipfile, make_archive, + register_archive_format, unregister_archive_format, + get_archive_formats, Error, unpack_archive, + register_unpack_format, RegistryError, + unregister_unpack_format, get_unpack_formats) +import tarfile +import warnings + +from test import support +from test.support import TESTFN, check_warnings, captured_stdout + +try: + import bz2 + BZ2_SUPPORTED = True +except ImportError: + BZ2_SUPPORTED = False + TESTFN2 = TESTFN + "2" +try: + import grp + import pwd + UID_GID_SUPPORT = True +except ImportError: + UID_GID_SUPPORT = False + +try: + import zlib +except ImportError: + zlib = None + +try: + import zipfile + ZIP_SUPPORT = True +except ImportError: + ZIP_SUPPORT = find_executable('zip') + 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')) + + def write_file(self, path, content='xxx'): + """Writes a file in the given path. + + + path can be a string or a sequence. + """ + if isinstance(path, (list, tuple)): + path = os.path.join(*path) + f = open(path, 'w') + try: + f.write(content) + finally: + f.close() + + def mkdtemp(self): + """Create a temporary directory that will be cleaned up. + + Returns the path of the directory. + """ + d = tempfile.mkdtemp() + self.tempdirs.append(d) + return d + def test_rmtree_errors(self): # filename is guaranteed not to exist filename = tempfile.mktemp() @@ -78,11 +149,12 @@ class TestShutil(unittest.TestCase): self.assertRaises(OSError, shutil.rmtree, path) os.remove(path) + def _write_data(self, path, data): + f = open(path, "w") + f.write(data) + f.close() + def test_copytree_simple(self): - def write_data(path, data): - f = open(path, "w") - f.write(data) - f.close() def read_data(path): f = open(path) @@ -92,11 +164,9 @@ class TestShutil(unittest.TestCase): src_dir = tempfile.mkdtemp() dst_dir = os.path.join(tempfile.mkdtemp(), 'destination') - - write_data(os.path.join(src_dir, 'test.txt'), '123') - + self._write_data(os.path.join(src_dir, 'test.txt'), '123') os.mkdir(os.path.join(src_dir, 'test_dir')) - write_data(os.path.join(src_dir, 'test_dir', 'test.txt'), '456') + self._write_data(os.path.join(src_dir, 'test_dir', 'test.txt'), '456') try: shutil.copytree(src_dir, dst_dir) @@ -125,11 +195,6 @@ class TestShutil(unittest.TestCase): def test_copytree_with_exclude(self): - def write_data(path, data): - f = open(path, "w") - f.write(data) - f.close() - def read_data(path): f = open(path) data = f.read() @@ -142,16 +207,18 @@ class TestShutil(unittest.TestCase): src_dir = tempfile.mkdtemp() try: dst_dir = join(tempfile.mkdtemp(), 'destination') - write_data(join(src_dir, 'test.txt'), '123') - write_data(join(src_dir, 'test.tmp'), '123') + self._write_data(join(src_dir, 'test.txt'), '123') + self._write_data(join(src_dir, 'test.tmp'), '123') os.mkdir(join(src_dir, 'test_dir')) - write_data(join(src_dir, 'test_dir', 'test.txt'), '456') + self._write_data(join(src_dir, 'test_dir', 'test.txt'), '456') os.mkdir(join(src_dir, 'test_dir2')) - write_data(join(src_dir, 'test_dir2', 'test.txt'), '456') + self._write_data(join(src_dir, 'test_dir2', 'test.txt'), '456') os.mkdir(join(src_dir, 'test_dir2', 'subdir')) os.mkdir(join(src_dir, 'test_dir2', 'subdir2')) - write_data(join(src_dir, 'test_dir2', 'subdir', 'test.txt'), '456') - write_data(join(src_dir, 'test_dir2', 'subdir2', 'test.py'), '456') + self._write_data(join(src_dir, 'test_dir2', 'subdir', 'test.txt'), + '456') + self._write_data(join(src_dir, 'test_dir2', 'subdir2', 'test.py'), + '456') # testing glob-like patterns @@ -204,48 +271,58 @@ class TestShutil(unittest.TestCase): shutil.rmtree(src_dir) shutil.rmtree(os.path.dirname(dst_dir)) - if hasattr(os, "symlink"): - 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: - f = open(src, 'w') + @unittest.skipUnless(hasattr(os, 'link'), 'requires os.link') + def test_dont_copy_file_onto_link_to_itself(self): + # Temporarily disable test on Windows. + if os.name == 'nt': + return + # bug 851123. + os.mkdir(TESTFN) + src = os.path.join(TESTFN, 'cheese') + dst = os.path.join(TESTFN, 'shop') + try: + with open(src, 'w') as f: f.write('cheddar') - f.close() - - os.link(src, dst) - self.assertRaises(shutil.Error, shutil.copyfile, src, dst) - with open(src, 'r') as f: - self.assertEqual(f.read(), 'cheddar') - os.remove(dst) - - # 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.Error, shutil.copyfile, src, dst) - with open(src, 'r') as f: - self.assertEqual(f.read(), 'cheddar') - os.remove(dst) - finally: - try: - shutil.rmtree(TESTFN) - except OSError: - pass + os.link(src, dst) + self.assertRaises(shutil.Error, shutil.copyfile, src, dst) + with open(src, 'r') as f: + self.assertEqual(f.read(), 'cheddar') + os.remove(dst) + finally: + shutil.rmtree(TESTFN, ignore_errors=True) - 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) - finally: - shutil.rmtree(TESTFN, ignore_errors=True) + @support.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') 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.Error, shutil.copyfile, src, dst) + with open(src, 'r') as f: + self.assertEqual(f.read(), 'cheddar') + os.remove(dst) + 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) + finally: + shutil.rmtree(TESTFN, ignore_errors=True) if hasattr(os, "mkfifo"): # Issue #3002: copyfile and copytree block indefinitely on named pipes @@ -259,6 +336,7 @@ class TestShutil(unittest.TestCase): finally: os.remove(TESTFN) + @support.skip_unless_symlink def test_copytree_named_pipe(self): os.mkdir(TESTFN) try: @@ -279,6 +357,320 @@ class TestShutil(unittest.TestCase): 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') + self._write_data(os.path.join(src_dir, 'test.txt'), '123') + os.mkdir(os.path.join(src_dir, 'test_dir')) + self._write_data(os.path.join(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')) + self._write_data(os.path.join(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)) + + @unittest.skipUnless(zlib, "requires zlib") + def test_make_tarball(self): + # creating something to tar + tmpdir = self.mkdtemp() + self.write_file([tmpdir, 'file1'], 'xxx') + self.write_file([tmpdir, 'file2'], 'xxx') + os.mkdir(os.path.join(tmpdir, 'sub')) + self.write_file([tmpdir, 'sub', 'file3'], 'xxx') + + tmpdir2 = self.mkdtemp() + unittest.skipUnless(splitdrive(tmpdir)[0] == splitdrive(tmpdir2)[0], + "source and target should be on same drive") + + base_name = os.path.join(tmpdir2, 'archive') + + # working with relative paths to avoid tar warnings + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + _make_tarball(splitdrive(base_name)[1], '.') + finally: + os.chdir(old_dir) + + # check if the compressed tarball was created + tarball = base_name + '.tar.gz' + self.assertTrue(os.path.exists(tarball)) + + # trying an uncompressed one + base_name = os.path.join(tmpdir2, 'archive') + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + _make_tarball(splitdrive(base_name)[1], '.', compress=None) + finally: + os.chdir(old_dir) + tarball = base_name + '.tar' + self.assertTrue(os.path.exists(tarball)) + + def _tarinfo(self, path): + tar = tarfile.open(path) + try: + names = tar.getnames() + names.sort() + return tuple(names) + finally: + tar.close() + + def _create_files(self): + # creating something to tar + tmpdir = self.mkdtemp() + dist = os.path.join(tmpdir, 'dist') + os.mkdir(dist) + self.write_file([dist, 'file1'], 'xxx') + self.write_file([dist, 'file2'], 'xxx') + os.mkdir(os.path.join(dist, 'sub')) + self.write_file([dist, 'sub', 'file3'], 'xxx') + os.mkdir(os.path.join(dist, 'sub2')) + tmpdir2 = self.mkdtemp() + base_name = os.path.join(tmpdir2, 'archive') + return tmpdir, tmpdir2, base_name + + @unittest.skipUnless(zlib, "Requires zlib") + @unittest.skipUnless(find_executable('tar') and find_executable('gzip'), + 'Need the tar command to run') + def test_tarfile_vs_tar(self): + tmpdir, tmpdir2, base_name = self._create_files() + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + _make_tarball(base_name, 'dist') + finally: + os.chdir(old_dir) + + # check if the compressed tarball was created + tarball = base_name + '.tar.gz' + self.assertTrue(os.path.exists(tarball)) + + # now create another tarball using `tar` + tarball2 = os.path.join(tmpdir, 'archive2.tar.gz') + tar_cmd = ['tar', '-cf', 'archive2.tar', 'dist'] + gzip_cmd = ['gzip', '-f9', 'archive2.tar'] + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + with captured_stdout() as s: + spawn(tar_cmd) + spawn(gzip_cmd) + finally: + os.chdir(old_dir) + + self.assertTrue(os.path.exists(tarball2)) + # let's compare both tarballs + self.assertEqual(self._tarinfo(tarball), self._tarinfo(tarball2)) + + # trying an uncompressed one + base_name = os.path.join(tmpdir2, 'archive') + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + _make_tarball(base_name, 'dist', compress=None) + finally: + os.chdir(old_dir) + tarball = base_name + '.tar' + self.assertTrue(os.path.exists(tarball)) + + # now for a dry_run + base_name = os.path.join(tmpdir2, 'archive') + old_dir = os.getcwd() + os.chdir(tmpdir) + try: + _make_tarball(base_name, 'dist', compress=None, dry_run=True) + finally: + os.chdir(old_dir) + tarball = base_name + '.tar' + self.assertTrue(os.path.exists(tarball)) + + @unittest.skipUnless(zlib, "Requires zlib") + @unittest.skipUnless(ZIP_SUPPORT, 'Need zip support to run') + def test_make_zipfile(self): + # creating something to tar + tmpdir = self.mkdtemp() + self.write_file([tmpdir, 'file1'], 'xxx') + self.write_file([tmpdir, 'file2'], 'xxx') + + tmpdir2 = self.mkdtemp() + base_name = os.path.join(tmpdir2, 'archive') + _make_zipfile(base_name, tmpdir) + + # check if the compressed tarball was created + tarball = base_name + '.zip' + self.assertTrue(os.path.exists(tarball)) + + + def test_make_archive(self): + tmpdir = self.mkdtemp() + base_name = os.path.join(tmpdir, 'archive') + self.assertRaises(ValueError, make_archive, base_name, 'xxx') + + @unittest.skipUnless(zlib, "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' + + base_dir, root_dir, base_name = 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.exists(res)) + + res = make_archive(base_name, 'zip', root_dir, base_dir) + self.assertTrue(os.path.exists(res)) + + res = make_archive(base_name, 'tar', root_dir, base_dir, + owner=owner, group=group) + self.assertTrue(os.path.exists(res)) + + res = make_archive(base_name, 'tar', root_dir, base_dir, + owner='kjhkjhkjg', group='oihohoh') + self.assertTrue(os.path.exists(res)) + + + @unittest.skipUnless(zlib, "Requires zlib") + @unittest.skipUnless(UID_GID_SUPPORT, "Requires grp and pwd support") + def test_tarfile_root_owner(self): + tmpdir, tmpdir2, base_name = self._create_files() + old_dir = os.getcwd() + os.chdir(tmpdir) + group = grp.getgrgid(0)[0] + owner = pwd.getpwuid(0)[0] + try: + archive_name = _make_tarball(base_name, 'dist', compress=None, + owner=owner, group=group) + finally: + os.chdir(old_dir) + + # check if the compressed tarball was created + self.assertTrue(os.path.exists(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(self): + current_dir = os.getcwd() + def _breaks(*args, **kw): + raise RuntimeError() + + register_archive_format('xxx', _breaks, [], 'xxx file') + try: + try: + make_archive('xxx', 'xxx', root_dir=self.mkdtemp()) + except Exception: + pass + self.assertEqual(os.getcwd(), current_dir) + finally: + unregister_archive_format('xxx') + + 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) + + def _compare_dirs(self, dir1, dir2): + # check that dir1 and dir2 are equivalent, + # return the diff + diff = [] + for root, dirs, files in os.walk(dir1): + for file_ in files: + path = os.path.join(root, file_) + target_path = os.path.join(dir2, os.path.split(path)[-1]) + if not os.path.exists(target_path): + diff.append(file_) + return diff + + @unittest.skipUnless(zlib, "Requires zlib") + def test_unpack_archive(self): + formats = ['tar', 'gztar', 'zip'] + if BZ2_SUPPORTED: + formats.append('bztar') + + for format in formats: + tmpdir = self.mkdtemp() + base_dir, root_dir, base_name = self._create_files() + tmpdir2 = self.mkdtemp() + filename = make_archive(base_name, format, root_dir, base_dir) + + # let's try to unpack it now + unpack_archive(filename, tmpdir2) + diff = self._compare_dirs(tmpdir, tmpdir2) + self.assertEqual(diff, []) + + def test_unpack_registery(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 TestMove(unittest.TestCase): @@ -419,6 +811,7 @@ class TestMove(unittest.TestCase): finally: shutil.rmtree(TESTFN, ignore_errors=True) + class TestCopyFile(unittest.TestCase): _delete = False |