diff --git a/Doc/library/tempfile.rst b/Doc/library/tempfile.rst index 00acf4b1792375..f182c0feaa7e7c 100644 --- a/Doc/library/tempfile.rst +++ b/Doc/library/tempfile.rst @@ -105,12 +105,12 @@ The module defines the following user-callable items: the truncate method now accepts a ``size`` argument. -.. 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 @@ -119,10 +119,19 @@ 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). .. versionadded:: 3.2 + .. versionchanged:: 3.10 + Added *ignore_cleanup_errors* parameter. + .. function:: mkstemp(suffix=None, prefix=None, dir=None, text=False) diff --git a/Lib/tempfile.py b/Lib/tempfile.py index f92db5313215e9..2ba92ad9f03466 100644 --- a/Lib/tempfile.py +++ b/Lib/tempfile.py @@ -774,7 +774,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: @@ -786,15 +786,49 @@ 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 _cleanup(cls, name, warn_message): - _shutil.rmtree(name) + def _rmtree(cls, name, ignore_errors=False): + def onerror(func, path, exc_info): + if issubclass(exc_info[0], PermissionError): + def resetperms(path): + try: + _os.chflags(path, 0) + except AttributeError: + pass + _os.chmod(path, 0o700) + + try: + if path != name: + resetperms(_os.path.dirname(path)) + resetperms(path) + + try: + _os.unlink(path) + # PermissionError is raised on FreeBSD for directories + except (IsADirectoryError, PermissionError): + cls._rmtree(path, ignore_errors=ignore_errors) + except FileNotFoundError: + pass + elif issubclass(exc_info[0], FileNotFoundError): + pass + else: + if not ignore_errors: + raise + + _shutil.rmtree(name, onerror=onerror) + + @classmethod + def _cleanup(cls, name, warn_message, ignore_errors=False): + cls._rmtree(name, ignore_errors=ignore_errors) _warnings.warn(warn_message, ResourceWarning) def __repr__(self): @@ -807,5 +841,7 @@ def __exit__(self, exc, value, tb): self.cleanup() def cleanup(self): - if self._finalizer.detach(): - _shutil.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) diff --git a/Lib/test/test_tempfile.py b/Lib/test/test_tempfile.py index 2d47e70bd03288..8171aa6e0ae52a 100644 --- a/Lib/test/test_tempfile.py +++ b/Lib/test/test_tempfile.py @@ -1312,13 +1312,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): + 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) # Create a subdirectory and some files if recurse: @@ -1351,7 +1355,31 @@ def test_explicit_cleanup(self): finally: os.rmdir(dir) - @support.skip_unless_symlink + 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) d1 = self.do_create() @@ -1385,6 +1413,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: @@ -1417,6 +1466,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: diff --git a/Misc/NEWS.d/next/Library/2021-03-07-23-23-03.bpo-29982.Q9iszT.rst b/Misc/NEWS.d/next/Library/2021-03-07-23-23-03.bpo-29982.Q9iszT.rst new file mode 100644 index 00000000000000..fd71bc6e4e0df7 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2021-03-07-23-23-03.bpo-29982.Q9iszT.rst @@ -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.