mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 00:48:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

GH-73991: Rework pathlib.Path.rmtree() into delete() (#122368)

Browse source 
Rename `pathlib.Path.rmtree()` to `delete()`, and add support for deleting
non-directories. This simplifies the interface for users, and nicely
complements the upcoming `move()` and `copy()` methods (which will also
accept any type of file.)
This commit is contained in:
Barney Gale 2024-08-07 01:34:44 +01:00 • committed by GitHub
parent b5e142ba7c
commit 98dba73010
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 139 additions and 149 deletions
Download patch file Download diff file Expand all files Collapse all files

77
Lib/test/test_pathlib/test_pathlib.py
View file

@ -32,7 +32,7 @@ root_in_posix = False
if hasattr(os, 'geteuid'):
root_in_posix = (os.geteuid() == 0)
rmtree_use_fd_functions = (
delete_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)
@ -862,8 +862,9 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
self.assertEqual(expected_gid, gid_2)
self.assertEqual(expected_name, link.group(follow_symlinks=False))
def test_rmtree_uses_safe_fd_version_if_available(self):
if rmtree_use_fd_functions:
def test_delete_uses_safe_fd_version_if_available(self):
if delete_use_fd_functions:
self.assertTrue(self.cls.delete.avoids_symlink_attacks)
d = self.cls(self.base, 'a')
d.mkdir()
try:
@ -876,16 +877,18 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
raise Called
os.open = _raiser
self.assertRaises(Called, d.rmtree)
self.assertRaises(Called, d.delete)
finally:
os.open = real_open
else:
self.assertFalse(self.cls.delete.avoids_symlink_attacks)
@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_rmtree_unwritable(self):
tmp = self.cls(self.base, 'rmtree')
def test_delete_unwritable(self):
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
child_file_path = tmp / 'a'
child_dir_path = tmp / 'b'
@ -902,7 +905,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
tmp.chmod(new_mode)
errors = []
tmp.rmtree(on_error=errors.append)
tmp.delete(on_error=errors.append)
# Test whether onerror has actually been called.
self.assertEqual(len(errors), 3)
finally:
@ -911,9 +914,9 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
child_dir_path.chmod(old_child_dir_mode)
@needs_windows
def test_rmtree_inner_junction(self):
def test_delete_inner_junction(self):
import _winapi
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
dir1 = tmp / 'dir1'
dir2 = dir1 / 'dir2'
@ -929,15 +932,15 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
link3 = dir1 / 'link3'
_winapi.CreateJunction(str(file1), str(link3))
# make sure junctions are removed but not followed
dir1.rmtree()
dir1.delete()
self.assertFalse(dir1.exists())
self.assertTrue(dir3.exists())
self.assertTrue(file1.exists())
@needs_windows
def test_rmtree_outer_junction(self):
def test_delete_outer_junction(self):
import _winapi
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
try:
src = tmp / 'cheese'
@ -946,22 +949,22 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
spam = src / 'spam'
spam.write_text('')
_winapi.CreateJunction(str(src), str(dst))
self.assertRaises(OSError, dst.rmtree)
dst.rmtree(ignore_errors=True)
self.assertRaises(OSError, dst.delete)
dst.delete(ignore_errors=True)
finally:
tmp.rmtree(ignore_errors=True)
tmp.delete(ignore_errors=True)
@needs_windows
def test_rmtree_outer_junction_on_error(self):
def test_delete_outer_junction_on_error(self):
import _winapi
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
dir_ = tmp / 'dir'
dir_.mkdir()
link = tmp / 'link'
_winapi.CreateJunction(str(dir_), str(link))
try:
self.assertRaises(OSError, link.rmtree)
self.assertRaises(OSError, link.delete)
self.assertTrue(dir_.exists())
self.assertTrue(link.exists(follow_symlinks=False))
errors = []
@ -969,18 +972,18 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
def on_error(error):
errors.append(error)
link.rmtree(on_error=on_error)
link.delete(on_error=on_error)
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0], OSError)
self.assertEqual(errors[0].filename, str(link))
finally:
os.unlink(str(link))
@unittest.skipUnless(rmtree_use_fd_functions, "requires safe rmtree")
def test_rmtree_fails_on_close(self):
@unittest.skipUnless(delete_use_fd_functions, "requires safe delete")
def test_delete_fails_on_close(self):
# Test that the error handler is called for failed os.close() and that
# os.close() is only called once for a file descriptor.
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
dir1 = tmp / 'dir1'
dir1.mkdir()
@ -996,7 +999,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
close_count = 0
with swap_attr(os, 'close', close) as orig_close:
with self.assertRaises(OSError):
dir1.rmtree()
dir1.delete()
self.assertTrue(dir2.is_dir())
self.assertEqual(close_count, 2)
@ -1004,7 +1007,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
errors = []
with swap_attr(os, 'close', close) as orig_close:
dir1.rmtree(on_error=errors.append)
dir1.delete(on_error=errors.append)
self.assertEqual(len(errors), 2)
self.assertEqual(errors[0].filename, str(dir2))
self.assertEqual(errors[1].filename, str(dir1))
@ -1013,27 +1016,23 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
@unittest.skipUnless(hasattr(os, "mkfifo"), 'requires os.mkfifo()')
@unittest.skipIf(sys.platform == "vxworks",
"fifo requires special path on VxWorks")
def test_rmtree_on_named_pipe(self):
def test_delete_on_named_pipe(self):
p = self.cls(self.base, 'pipe')
os.mkfifo(p)
try:
with self.assertRaises(NotADirectoryError):
p.rmtree()
self.assertTrue(p.exists())
finally:
p.unlink()
p.delete()
self.assertFalse(p.exists())
p = self.cls(self.base, 'dir')
p.mkdir()
os.mkfifo(p / 'mypipe')
p.rmtree()
p.delete()
self.assertFalse(p.exists())
@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_rmtree_deleted_race_condition(self):
def test_delete_deleted_race_condition(self):
# bpo-37260
#
# Test that a file or a directory deleted after it is enumerated
@ -1057,7 +1056,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
if p != keep:
p.unlink()
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
paths = [tmp] + [tmp / f'child{i}' for i in range(6)]
dirs = paths[1::2]
@ -1075,7 +1074,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
path.chmod(new_mode)
try:
tmp.rmtree(on_error=on_error)
tmp.delete(on_error=on_error)
except:
# Test failed, so cleanup artifacts.
for path, mode in zip(paths, old_modes):
@ -1083,13 +1082,13 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
path.chmod(mode)
except OSError:
pass
tmp.rmtree()
tmp.delete()
raise
def test_rmtree_does_not_choke_on_failing_lstat(self):
def test_delete_does_not_choke_on_failing_lstat(self):
try:
orig_lstat = os.lstat
tmp = self.cls(self.base, 'rmtree')
tmp = self.cls(self.base, 'delete')
def raiser(fn, *args, **kwargs):
if fn != str(tmp):
@ -1102,7 +1101,7 @@ class PathTest(test_pathlib_abc.DummyPathTest, PurePathTest):
tmp.mkdir()
foo = tmp / 'foo'
foo.write_text('')
tmp.rmtree()
tmp.delete()
finally:
os.lstat = orig_lstat

91
Lib/test/test_pathlib/test_pathlib_abc.py
View file

@ -2641,85 +2641,43 @@ class DummyPathTest(DummyPurePathTest):
self.assertFileNotFound(p.stat)
self.assertFileNotFound(p.unlink)
def test_rmtree(self):
def test_delete_file(self):
p = self.cls(self.base) / 'fileA'
p.delete()
self.assertFileNotFound(p.stat)
self.assertFileNotFound(p.unlink)
def test_delete_dir(self):
base = self.cls(self.base)
base.joinpath('dirA').rmtree()
base.joinpath('dirA').delete()
self.assertRaises(FileNotFoundError, base.joinpath('dirA').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirA', 'linkC').lstat)
base.joinpath('dirB').rmtree()
base.joinpath('dirB').delete()
self.assertRaises(FileNotFoundError, base.joinpath('dirB').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirB', 'fileB').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirB', 'linkD').lstat)
base.joinpath('dirC').rmtree()
base.joinpath('dirC').delete()
self.assertRaises(FileNotFoundError, base.joinpath('dirC').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirC', 'dirD').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirC', 'dirD', 'fileD').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirC', 'fileC').stat)
self.assertRaises(FileNotFoundError, base.joinpath('dirC', 'novel.txt').stat)
def test_rmtree_errors(self):
tmp = self.cls(self.base, 'rmtree')
tmp.mkdir()
# filename is guaranteed not to exist
filename = tmp / 'foo'
self.assertRaises(FileNotFoundError, filename.rmtree)
# test that ignore_errors option is honored
filename.rmtree(ignore_errors=True)
# existing file
filename = tmp / "tstfile"
filename.write_text("")
with self.assertRaises(NotADirectoryError) as cm:
filename.rmtree()
self.assertEqual(cm.exception.filename, str(filename))
self.assertTrue(filename.exists())
# test that ignore_errors option is honored
filename.rmtree(ignore_errors=True)
self.assertTrue(filename.exists())
def test_rmtree_on_error(self):
tmp = self.cls(self.base, 'rmtree')
tmp.mkdir()
filename = tmp / "tstfile"
filename.write_text("")
errors = []
def on_error(error):
errors.append(error)
filename.rmtree(on_error=on_error)
self.assertEqual(len(errors), 2)
# First from scandir()
self.assertIsInstance(errors[0], NotADirectoryError)
self.assertEqual(errors[0].filename, str(filename))
# Then from munlink()
self.assertIsInstance(errors[1], NotADirectoryError)
self.assertEqual(errors[1].filename, str(filename))
@needs_symlinks
def test_rmtree_outer_symlink(self):
tmp = self.cls(self.base, 'rmtree')
def test_delete_symlink(self):
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
dir_ = tmp / 'dir'
dir_.mkdir()
link = tmp / 'link'
link.symlink_to(dir_)
self.assertRaises(OSError, link.rmtree)
link.delete()
self.assertTrue(dir_.exists())
self.assertTrue(link.exists(follow_symlinks=False))
errors = []
def on_error(error):
errors.append(error)
link.rmtree(on_error=on_error)
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0], OSError)
self.assertEqual(errors[0].filename, str(link))
self.assertFalse(link.exists(follow_symlinks=False))
@needs_symlinks
def test_rmtree_inner_symlink(self):
tmp = self.cls(self.base, 'rmtree')
def test_delete_inner_symlink(self):
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
dir1 = tmp / 'dir1'
dir2 = dir1 / 'dir2'
@ -2735,11 +2693,26 @@ class DummyPathTest(DummyPurePathTest):
link3 = dir1 / 'link3'
link3.symlink_to(file1)
# make sure symlinks are removed but not followed
dir1.rmtree()
dir1.delete()
self.assertFalse(dir1.exists())
self.assertTrue(dir3.exists())
self.assertTrue(file1.exists())
def test_delete_missing(self):
tmp = self.cls(self.base, 'delete')
tmp.mkdir()
# filename is guaranteed not to exist
filename = tmp / 'foo'
self.assertRaises(FileNotFoundError, filename.delete)
# test that ignore_errors option is honored
filename.delete(ignore_errors=True)
# test on_error
errors = []
filename.delete(on_error=errors.append)
self.assertEqual(len(errors), 1)
self.assertIsInstance(errors[0], FileNotFoundError)
self.assertEqual(errors[0].filename, str(filename))
def setUpWalk(self):
# Build:
# TESTFN/