diff --git a/src/_pytest/pathlib.py b/src/_pytest/pathlib.py index 1fc16f5cd..4cd71e4b1 100644 --- a/src/_pytest/pathlib.py +++ b/src/_pytest/pathlib.py @@ -1,4 +1,3 @@ -import atexit import contextlib import fnmatch import importlib.util @@ -244,7 +243,7 @@ def create_cleanup_lock(p: Path) -> Path: return lock_path -def register_cleanup_lock_removal(lock_path: Path, register=atexit.register): +def register_cleanup_lock_removal(lock_path: Path, register): """Register a cleanup function for removing a lock, by default on atexit.""" pid = os.getpid() @@ -362,6 +361,7 @@ def make_numbered_dir_with_cleanup( keep: int, lock_timeout: float, mode: int, + register, ) -> Path: """Create a numbered dir with a cleanup lock and remove old ones.""" e = None @@ -371,13 +371,13 @@ def make_numbered_dir_with_cleanup( # Only lock the current dir when keep is not 0 if keep != 0: lock_path = create_cleanup_lock(p) - register_cleanup_lock_removal(lock_path) + register_cleanup_lock_removal(lock_path, register=register) except Exception as exc: e = exc else: consider_lock_dead_if_created_before = p.stat().st_mtime - lock_timeout # Register a cleanup for program exit - atexit.register( + register( cleanup_numbered_dir, root, prefix, diff --git a/src/_pytest/tmpdir.py b/src/_pytest/tmpdir.py index 57671483f..724b29eb4 100644 --- a/src/_pytest/tmpdir.py +++ b/src/_pytest/tmpdir.py @@ -3,6 +3,7 @@ import os import re import sys import tempfile +from contextlib import ExitStack from pathlib import Path from shutil import rmtree from typing import Dict @@ -77,6 +78,7 @@ class TempPathFactory: self._retention_count = retention_count self._retention_policy = retention_policy self._basetemp = basetemp + self._exit_stack = ExitStack() @classmethod def from_config( @@ -196,6 +198,7 @@ class TempPathFactory: keep=keep, lock_timeout=LOCK_TIMEOUT, mode=0o700, + register=self._exit_stack.callback, ) assert basetemp is not None, basetemp self._basetemp = basetemp @@ -303,19 +306,23 @@ def pytest_sessionfinish(session, exitstatus: Union[int, ExitCode]): the policy is "failed", and the basetemp is not specified by a user. """ tmp_path_factory: TempPathFactory = session.config._tmp_path_factory - if tmp_path_factory._basetemp is None: - return - policy = tmp_path_factory._retention_policy - if ( - exitstatus == 0 - and policy == "failed" - and tmp_path_factory._given_basetemp is None - ): - passed_dir = tmp_path_factory._basetemp - if passed_dir.exists(): - # We do a "best effort" to remove files, but it might not be possible due to some leaked resource, - # permissions, etc, in which case we ignore it. - rmtree(passed_dir, ignore_errors=True) + + # tmporal directory cleanup, which is registered to + # this ExitStack, will be executed at the end of this scope + with tmp_path_factory._exit_stack: + if tmp_path_factory._basetemp is None: + return + policy = tmp_path_factory._retention_policy + if ( + exitstatus == 0 + and policy == "failed" + and tmp_path_factory._given_basetemp is None + ): + passed_dir = tmp_path_factory._basetemp + if passed_dir.exists(): + # We do a "best effort" to remove files, but it might not be possible due to some leaked resource, + # permissions, etc, in which case we ignore it. + rmtree(passed_dir, ignore_errors=True) @hookimpl(tryfirst=True, hookwrapper=True) diff --git a/testing/test_tmpdir.py b/testing/test_tmpdir.py index 57f442b04..1daf04613 100644 --- a/testing/test_tmpdir.py +++ b/testing/test_tmpdir.py @@ -92,6 +92,73 @@ class TestConfigTmpPath: assert mytemp.exists() assert not mytemp.joinpath("hello").exists() + def test_policy_none_delete_all(self, pytester: Pytester) -> None: + p = pytester.makepyfile( + """ + def test_1(tmp_path): + assert 0 == 0 + """ + ) + p_failed = pytester.makepyfile( + another_file_name=""" + def test_1(tmp_path): + assert 0 == 1 + """ + ) + pytester.makepyprojecttoml( + """ + [tool.pytest.ini_options] + tmp_path_retention_policy = "none" + """ + ) + + pytester.inline_run(p) + pytester.inline_run(p_failed) + + root = pytester._test_tmproot + for child in root.iterdir(): + base_dir = list(child.iterdir()) + # Check the base dir itself is gone without depending on test results + assert base_dir == [] + + @pytest.mark.parametrize("policy", ['"failed"', '"all"']) + @pytest.mark.parametrize("count", [0, 1, 3]) + def test_retention_count(self, pytester: Pytester, policy, count) -> None: + p = pytester.makepyfile( + """ + def test_1(tmp_path): + assert 0 == 0 + """ + ) + p_failed = pytester.makepyfile( + another_file_name=""" + def test_1(tmp_path): + assert 0 == 1 + """ + ) + + pytester.makepyprojecttoml( + f""" + [tool.pytest.ini_options] + tmp_path_retention_policy = {policy} + tmp_path_retention_count = {count} + """ + ) + + pytester.inline_run(p) + pytester.inline_run(p_failed) + pytester.inline_run(p) + pytester.inline_run(p_failed) + pytester.inline_run(p) + pytester.inline_run(p_failed) + pytester.inline_run(p) + pytester.inline_run(p_failed) + + root = pytester._test_tmproot + for child in root.iterdir(): + base_dir = filter(lambda x: not x.is_symlink(), child.iterdir()) + assert len(list(base_dir)) == count + def test_policy_failed_removes_only_passed_dir(self, pytester: Pytester) -> None: p = pytester.makepyfile( """ @@ -119,26 +186,6 @@ class TestConfigTmpPath: assert len(test_dir) == 1 assert test_dir[0].name == "test_20" - def test_policy_failed_removes_basedir_when_all_passed( - self, pytester: Pytester - ) -> None: - p = pytester.makepyfile( - """ - def test_1(tmp_path): - assert 0 == 0 - """ - ) - - pytester.inline_run(p) - root = pytester._test_tmproot - for child in root.iterdir(): - # This symlink will be deleted by cleanup_numbered_dir **after** - # the test finishes because it's triggered by atexit. - # So it has to be ignored here. - base_dir = filter(lambda x: not x.is_symlink(), child.iterdir()) - # Check the base dir itself is gone - assert len(list(base_dir)) == 0 - # issue #10502 def test_policy_failed_removes_dir_when_skipped_from_fixture( self, pytester: Pytester @@ -160,10 +207,8 @@ class TestConfigTmpPath: # Check if the whole directory is removed root = pytester._test_tmproot for child in root.iterdir(): - base_dir = list( - filter(lambda x: x.is_dir() and not x.is_symlink(), child.iterdir()) - ) - assert len(base_dir) == 0 + base_dir = list(child.iterdir()) + assert base_dir == [] # issue #10502 def test_policy_all_keeps_dir_when_skipped_from_fixture(