mirror of
https://github.com/python/cpython
synced 2024-10-14 09:01:41 +00:00
bpo-29982: Add "ignore_cleanup_errors" param to tempfile.TemporaryDirectory() (GH-24793)
This commit is contained in:
parent
d48848c83e
commit
bd2fa3c416
|
@ -118,12 +118,12 @@ The module defines the following user-callable items:
|
|||
Added *errors* parameter.
|
||||
|
||||
|
||||
.. function:: TemporaryDirectory(suffix=None, prefix=None, dir=None)
|
||||
.. function:: TemporaryDirectory(suffix=None, prefix=None, dir=None, ignore_cleanup_errors=False)
|
||||
|
||||
This function securely creates a temporary directory using the same rules as :func:`mkdtemp`.
|
||||
The resulting object can be used as a context manager (see
|
||||
:ref:`tempfile-examples`). On completion of the context or destruction
|
||||
of the temporary directory object the newly created temporary directory
|
||||
of the temporary directory object, the newly created temporary directory
|
||||
and all its contents are removed from the filesystem.
|
||||
|
||||
The directory name can be retrieved from the :attr:`name` attribute of the
|
||||
|
@ -132,12 +132,21 @@ The module defines the following user-callable items:
|
|||
the :keyword:`with` statement, if there is one.
|
||||
|
||||
The directory can be explicitly cleaned up by calling the
|
||||
:func:`cleanup` method.
|
||||
:func:`cleanup` method. If *ignore_cleanup_errors* is true, any unhandled
|
||||
exceptions during explicit or implicit cleanup (such as a
|
||||
:exc:`PermissionError` removing open files on Windows) will be ignored,
|
||||
and the remaining removable items deleted on a "best-effort" basis.
|
||||
Otherwise, errors will be raised in whatever context cleanup occurs
|
||||
(the :func:`cleanup` call, exiting the context manager, when the object
|
||||
is garbage-collected or during interpreter shutdown).
|
||||
|
||||
.. audit-event:: tempfile.mkdtemp fullpath tempfile.TemporaryDirectory
|
||||
|
||||
.. versionadded:: 3.2
|
||||
|
||||
.. versionchanged:: 3.10
|
||||
Added *ignore_cleanup_errors* parameter.
|
||||
|
||||
|
||||
.. function:: mkstemp(suffix=None, prefix=None, dir=None, text=False)
|
||||
|
||||
|
|
|
@ -768,7 +768,7 @@ def writelines(self, iterable):
|
|||
return rv
|
||||
|
||||
|
||||
class TemporaryDirectory(object):
|
||||
class TemporaryDirectory:
|
||||
"""Create and return a temporary directory. This has the same
|
||||
behavior as mkdtemp but can be used as a context manager. For
|
||||
example:
|
||||
|
@ -780,14 +780,17 @@ class TemporaryDirectory(object):
|
|||
in it are removed.
|
||||
"""
|
||||
|
||||
def __init__(self, suffix=None, prefix=None, dir=None):
|
||||
def __init__(self, suffix=None, prefix=None, dir=None,
|
||||
ignore_cleanup_errors=False):
|
||||
self.name = mkdtemp(suffix, prefix, dir)
|
||||
self._ignore_cleanup_errors = ignore_cleanup_errors
|
||||
self._finalizer = _weakref.finalize(
|
||||
self, self._cleanup, self.name,
|
||||
warn_message="Implicitly cleaning up {!r}".format(self))
|
||||
warn_message="Implicitly cleaning up {!r}".format(self),
|
||||
ignore_errors=self._ignore_cleanup_errors)
|
||||
|
||||
@classmethod
|
||||
def _rmtree(cls, name):
|
||||
def _rmtree(cls, name, ignore_errors=False):
|
||||
def onerror(func, path, exc_info):
|
||||
if issubclass(exc_info[0], PermissionError):
|
||||
def resetperms(path):
|
||||
|
@ -806,19 +809,20 @@ def resetperms(path):
|
|||
_os.unlink(path)
|
||||
# PermissionError is raised on FreeBSD for directories
|
||||
except (IsADirectoryError, PermissionError):
|
||||
cls._rmtree(path)
|
||||
cls._rmtree(path, ignore_errors=ignore_errors)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
elif issubclass(exc_info[0], FileNotFoundError):
|
||||
pass
|
||||
else:
|
||||
raise
|
||||
if not ignore_errors:
|
||||
raise
|
||||
|
||||
_shutil.rmtree(name, onerror=onerror)
|
||||
|
||||
@classmethod
|
||||
def _cleanup(cls, name, warn_message):
|
||||
cls._rmtree(name)
|
||||
def _cleanup(cls, name, warn_message, ignore_errors=False):
|
||||
cls._rmtree(name, ignore_errors=ignore_errors)
|
||||
_warnings.warn(warn_message, ResourceWarning)
|
||||
|
||||
def __repr__(self):
|
||||
|
@ -831,7 +835,7 @@ def __exit__(self, exc, value, tb):
|
|||
self.cleanup()
|
||||
|
||||
def cleanup(self):
|
||||
if self._finalizer.detach():
|
||||
self._rmtree(self.name)
|
||||
if self._finalizer.detach() or _os.path.exists(self.name):
|
||||
self._rmtree(self.name, ignore_errors=self._ignore_cleanup_errors)
|
||||
|
||||
__class_getitem__ = classmethod(_types.GenericAlias)
|
||||
|
|
|
@ -1365,13 +1365,17 @@ def __exit__(self, *exc_info):
|
|||
d.clear()
|
||||
d.update(c)
|
||||
|
||||
|
||||
class TestTemporaryDirectory(BaseTestCase):
|
||||
"""Test TemporaryDirectory()."""
|
||||
|
||||
def do_create(self, dir=None, pre="", suf="", recurse=1, dirs=1, files=1):
|
||||
def do_create(self, dir=None, pre="", suf="", recurse=1, dirs=1, files=1,
|
||||
ignore_cleanup_errors=False):
|
||||
if dir is None:
|
||||
dir = tempfile.gettempdir()
|
||||
tmp = tempfile.TemporaryDirectory(dir=dir, prefix=pre, suffix=suf)
|
||||
tmp = tempfile.TemporaryDirectory(
|
||||
dir=dir, prefix=pre, suffix=suf,
|
||||
ignore_cleanup_errors=ignore_cleanup_errors)
|
||||
self.nameCheck(tmp.name, dir, pre, suf)
|
||||
self.do_create2(tmp.name, recurse, dirs, files)
|
||||
return tmp
|
||||
|
@ -1410,6 +1414,30 @@ def test_explicit_cleanup(self):
|
|||
finally:
|
||||
os.rmdir(dir)
|
||||
|
||||
def test_explict_cleanup_ignore_errors(self):
|
||||
"""Test that cleanup doesn't return an error when ignoring them."""
|
||||
with tempfile.TemporaryDirectory() as working_dir:
|
||||
temp_dir = self.do_create(
|
||||
dir=working_dir, ignore_cleanup_errors=True)
|
||||
temp_path = pathlib.Path(temp_dir.name)
|
||||
self.assertTrue(temp_path.exists(),
|
||||
f"TemporaryDirectory {temp_path!s} does not exist")
|
||||
with open(temp_path / "a_file.txt", "w+t") as open_file:
|
||||
open_file.write("Hello world!\n")
|
||||
temp_dir.cleanup()
|
||||
self.assertEqual(len(list(temp_path.glob("*"))),
|
||||
int(sys.platform.startswith("win")),
|
||||
"Unexpected number of files in "
|
||||
f"TemporaryDirectory {temp_path!s}")
|
||||
self.assertEqual(
|
||||
temp_path.exists(),
|
||||
sys.platform.startswith("win"),
|
||||
f"TemporaryDirectory {temp_path!s} existance state unexpected")
|
||||
temp_dir.cleanup()
|
||||
self.assertFalse(
|
||||
temp_path.exists(),
|
||||
f"TemporaryDirectory {temp_path!s} exists after cleanup")
|
||||
|
||||
@os_helper.skip_unless_symlink
|
||||
def test_cleanup_with_symlink_to_a_directory(self):
|
||||
# cleanup() should not follow symlinks to directories (issue #12464)
|
||||
|
@ -1444,6 +1472,27 @@ def test_del_on_collection(self):
|
|||
finally:
|
||||
os.rmdir(dir)
|
||||
|
||||
@support.cpython_only
|
||||
def test_del_on_collection_ignore_errors(self):
|
||||
"""Test that ignoring errors works when TemporaryDirectory is gced."""
|
||||
with tempfile.TemporaryDirectory() as working_dir:
|
||||
temp_dir = self.do_create(
|
||||
dir=working_dir, ignore_cleanup_errors=True)
|
||||
temp_path = pathlib.Path(temp_dir.name)
|
||||
self.assertTrue(temp_path.exists(),
|
||||
f"TemporaryDirectory {temp_path!s} does not exist")
|
||||
with open(temp_path / "a_file.txt", "w+t") as open_file:
|
||||
open_file.write("Hello world!\n")
|
||||
del temp_dir
|
||||
self.assertEqual(len(list(temp_path.glob("*"))),
|
||||
int(sys.platform.startswith("win")),
|
||||
"Unexpected number of files in "
|
||||
f"TemporaryDirectory {temp_path!s}")
|
||||
self.assertEqual(
|
||||
temp_path.exists(),
|
||||
sys.platform.startswith("win"),
|
||||
f"TemporaryDirectory {temp_path!s} existance state unexpected")
|
||||
|
||||
def test_del_on_shutdown(self):
|
||||
# A TemporaryDirectory may be cleaned up during shutdown
|
||||
with self.do_create() as dir:
|
||||
|
@ -1476,6 +1525,43 @@ def test_del_on_shutdown(self):
|
|||
self.assertNotIn("Exception ", err)
|
||||
self.assertIn("ResourceWarning: Implicitly cleaning up", err)
|
||||
|
||||
def test_del_on_shutdown_ignore_errors(self):
|
||||
"""Test ignoring errors works when a tempdir is gc'ed on shutdown."""
|
||||
with tempfile.TemporaryDirectory() as working_dir:
|
||||
code = """if True:
|
||||
import pathlib
|
||||
import sys
|
||||
import tempfile
|
||||
import warnings
|
||||
|
||||
temp_dir = tempfile.TemporaryDirectory(
|
||||
dir={working_dir!r}, ignore_cleanup_errors=True)
|
||||
sys.stdout.buffer.write(temp_dir.name.encode())
|
||||
|
||||
temp_dir_2 = pathlib.Path(temp_dir.name) / "test_dir"
|
||||
temp_dir_2.mkdir()
|
||||
with open(temp_dir_2 / "test0.txt", "w") as test_file:
|
||||
test_file.write("Hello world!")
|
||||
open_file = open(temp_dir_2 / "open_file.txt", "w")
|
||||
open_file.write("Hello world!")
|
||||
|
||||
warnings.filterwarnings("always", category=ResourceWarning)
|
||||
""".format(working_dir=working_dir)
|
||||
__, out, err = script_helper.assert_python_ok("-c", code)
|
||||
temp_path = pathlib.Path(out.decode().strip())
|
||||
self.assertEqual(len(list(temp_path.glob("*"))),
|
||||
int(sys.platform.startswith("win")),
|
||||
"Unexpected number of files in "
|
||||
f"TemporaryDirectory {temp_path!s}")
|
||||
self.assertEqual(
|
||||
temp_path.exists(),
|
||||
sys.platform.startswith("win"),
|
||||
f"TemporaryDirectory {temp_path!s} existance state unexpected")
|
||||
err = err.decode('utf-8', 'backslashreplace')
|
||||
self.assertNotIn("Exception", err)
|
||||
self.assertNotIn("Error", err)
|
||||
self.assertIn("ResourceWarning: Implicitly cleaning up", err)
|
||||
|
||||
def test_exit_on_shutdown(self):
|
||||
# Issue #22427
|
||||
with self.do_create() as dir:
|
||||
|
|
|
@ -0,0 +1,3 @@
|
|||
Add optional parameter *ignore_cleanup_errors* to
|
||||
:func:`tempfile.TemporaryDirectory` and allow multiple :func:`cleanup` attempts.
|
||||
Contributed by C.A.M. Gerlach.
|
Loading…
Reference in a new issue