我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用shutil.open()。
def test_on_error(self): self.errorState = 0 os.mkdir(TESTFN) self.childpath = os.path.join(TESTFN, 'a') f = open(self.childpath, 'w') f.close() old_dir_mode = os.stat(TESTFN).st_mode old_child_mode = os.stat(self.childpath).st_mode # Make unwritable. os.chmod(self.childpath, stat.S_IREAD) os.chmod(TESTFN, stat.S_IREAD) shutil.rmtree(TESTFN, onerror=self.check_args_to_onerror) # Test whether onerror has actually been called. self.assertEqual(self.errorState, 2, "Expected call to onerror function did not happen.") # Make writable again. os.chmod(TESTFN, old_dir_mode) os.chmod(self.childpath, old_child_mode) # Clean up. shutil.rmtree(TESTFN)
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') 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_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_w_dest_open_fails(self): srcfile = self.Faux() def _open(filename, mode='r'): if filename == 'srcfile': return srcfile if filename == 'destfile': raise IOError('Cannot open "destfile"') assert 0 # shouldn't reach here. self._set_shutil_open(_open) shutil.copyfile('srcfile', 'destfile') self.assertTrue(srcfile._entered) self.assertTrue(srcfile._exited_with[0] is IOError) self.assertEqual(srcfile._exited_with[1].args, ('Cannot open "destfile"',))
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 support.change_cwd(root_dir): 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 setUp(self): filename = "foo" self.src_dir = tempfile.mkdtemp() self.dst_dir = tempfile.mkdtemp() self.src_file = os.path.join(self.src_dir, filename) self.dst_file = os.path.join(self.dst_dir, filename) # Try to create a dir in the current directory, hoping that it is # not located on the same filesystem as the system tmp dir. try: self.dir_other_fs = tempfile.mkdtemp( dir=os.path.dirname(__file__)) self.file_other_fs = os.path.join(self.dir_other_fs, filename) except OSError: self.dir_other_fs = None with open(self.src_file, "wb") as f: f.write("spam")
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_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)
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.SameFileError, 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_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. self._set_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"',))
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 _write_data(self, path, data): f = open(path, "w") f.write(data) f.close()
def test_copytree_simple(self): def read_data(path): f = open(path) data = f.read() f.close() return data src_dir = tempfile.mkdtemp() dst_dir = os.path.join(tempfile.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') try: 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_data(os.path.join(dst_dir, 'test.txt')) self.assertEqual(actual, '123') actual = read_data(os.path.join(dst_dir, 'test_dir', 'test.txt')) self.assertEqual(actual, '456') finally: for path in ( os.path.join(src_dir, 'test.txt'), os.path.join(dst_dir, 'test.txt'), os.path.join(src_dir, 'test_dir', 'test.txt'), os.path.join(dst_dir, 'test_dir', 'test.txt'), ): if os.path.exists(path): os.remove(path) for path in (src_dir, os.path.dirname(dst_dir) ): if os.path.exists(path): shutil.rmtree(path)
def _tarinfo(self, path): tar = tarfile.open(path) try: names = tar.getnames() names.sort() return tuple(names) finally: tar.close()
def setUp(self): filename = "foo" self.src_dir = tempfile.mkdtemp() self.dst_dir = tempfile.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 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_w_source_open_fails(self): def _open(filename, mode='r'): if filename == 'srcfile': raise IOError('Cannot open "srcfile"') assert 0 # shouldn't reach here. self._set_shutil_open(_open) self.assertRaises(IOError, shutil.copyfile, 'srcfile', 'destfile')
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') 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
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 support.change_cwd(work_dir): 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.assertEqual(sorted(tf.getnames()), ['.', './file1', './file2', './sub', './sub/file3', './sub2']) # trying an uncompressed one with support.change_cwd(work_dir): 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.assertEqual(sorted(tf.getnames()), ['.', './file1', './file2', './sub', './sub/file3', './sub2'])
def _set_shutil_open(self, func): shutil.open = func self._delete = True
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) with open(path, 'wb' if binary else 'w') as fp: fp.write(content)
