diff options
Diffstat (limited to 'Lib')
| -rw-r--r-- | Lib/tempfile.py | 24 | ||||
| -rw-r--r-- | Lib/test/test_tempfile.py | 90 |
2 files changed, 102 insertions, 12 deletions
diff --git a/Lib/tempfile.py b/Lib/tempfile.py index dc088d9..4b2547c 100644 --- a/Lib/tempfile.py +++ b/Lib/tempfile.py @@ -768,7 +768,7 @@ class SpooledTemporaryFile: 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 @@ class TemporaryDirectory(object): _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 @@ class TemporaryDirectory(object): 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) diff --git a/Lib/test/test_tempfile.py b/Lib/test/test_tempfile.py index 5822c75..3a3f6a9 100644 --- a/Lib/test/test_tempfile.py +++ b/Lib/test/test_tempfile.py @@ -1365,13 +1365,17 @@ class NulledModules: 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 @@ class TestTemporaryDirectory(BaseTestCase): 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 @@ class TestTemporaryDirectory(BaseTestCase): 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 @@ class TestTemporaryDirectory(BaseTestCase): 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: |