Skip to content
Merged
Prev Previous commit
Next Next commit
Add new tests for TemporaryDirectory cleanup when ignoring errors
  • Loading branch information
CAM-Gerlach committed Mar 8, 2021
commit 571e918a5a01d66c0c853ad727842a4573b2f0fe
75 changes: 73 additions & 2 deletions Lib/test/test_tempfile.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1410,6 +1414,25 @@ 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.assertIn(len(list(temp_path.glob("*"))), {0, 1},
"Not all closed files cleaned up in "
f"TemporaryDirectory {temp_path!s}")
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)
Expand Down Expand Up @@ -1444,6 +1467,22 @@ 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.assertIn(len(list(temp_path.glob("*"))), {0, 1},
"Not all closed files cleaned up in "
f"TemporaryDirectory {temp_path!s}")

def test_del_on_shutdown(self):
# A TemporaryDirectory may be cleaned up during shutdown
with self.do_create() as dir:
Expand Down Expand Up @@ -1476,6 +1515,38 @@ 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.assertIn(len(list(temp_path.glob("*"))), {0, 1},
"Not all closed files cleaned up in "
f"TemporaryDirectory {temp_path!s}")
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:
Expand Down