diff --git a/AUTHORS b/AUTHORS index f2f330c4b..d47d75961 100644 --- a/AUTHORS +++ b/AUTHORS @@ -374,6 +374,7 @@ Xixi Zhao Xuan Luong Xuecong Liao Yoav Caspi +Yusuke Kadowaki Yuval Shimon Zac Hatfield-Dodds Zachary Kneupper diff --git a/changelog/8141.feature.rst b/changelog/8141.feature.rst new file mode 100644 index 000000000..70de099e6 --- /dev/null +++ b/changelog/8141.feature.rst @@ -0,0 +1,2 @@ +Added :confval:`tmp_path_retention_count` and :confval:`tmp_path_retention_policy` configuration options to control how directories created by the :fixture:`tmp_path` fixture are kept. +The default behavior has changed to keep only directories for failed tests, equivalent to `tmp_path_retention_policy="failed"`. diff --git a/doc/en/reference/reference.rst b/doc/en/reference/reference.rst index b7bd4d119..25d76568a 100644 --- a/doc/en/reference/reference.rst +++ b/doc/en/reference/reference.rst @@ -1723,6 +1723,40 @@ passed multiple times. The expected format is ``name=value``. For example:: directories when executing from the root directory. +.. confval:: tmp_path_retention_count + + + + How many sessions should we keep the `tmp_path` directories, + according to `tmp_path_retention_policy`. + + .. code-block:: ini + + [pytest] + tmp_path_retention_count = 3 + + Default: 3 + + +.. confval:: tmp_path_retention_policy + + + + Controls which directories created by the `tmp_path` fixture are kept around, + based on test outcome. + + * `all`: retains directories for all tests, regardless of the outcome. + * `failed`: retains directories only for tests with outcome `error` or `failed`. + * `none`: directories are always removed after each test ends, regardless of the outcome. + + .. code-block:: ini + + [pytest] + tmp_path_retention_policy = "all" + + Default: failed + + .. confval:: usefixtures List of fixtures that will be applied to all test functions; this is semantically the same to apply diff --git a/src/_pytest/pathlib.py b/src/_pytest/pathlib.py index c5a411b59..533335c66 100644 --- a/src/_pytest/pathlib.py +++ b/src/_pytest/pathlib.py @@ -335,15 +335,26 @@ def cleanup_candidates(root: Path, prefix: str, keep: int) -> Iterator[Path]: yield path +def cleanup_dead_symlink(root: Path): + for left_dir in root.iterdir(): + if left_dir.is_symlink(): + if not left_dir.resolve().exists(): + left_dir.unlink() + + def cleanup_numbered_dir( root: Path, prefix: str, keep: int, consider_lock_dead_if_created_before: float ) -> None: """Cleanup for lock driven numbered directories.""" + if not root.exists(): + return for path in cleanup_candidates(root, prefix, keep): try_cleanup(path, consider_lock_dead_if_created_before) for path in root.glob("garbage-*"): try_cleanup(path, consider_lock_dead_if_created_before) + cleanup_dead_symlink(root) + def make_numbered_dir_with_cleanup( root: Path, @@ -357,8 +368,10 @@ def make_numbered_dir_with_cleanup( for i in range(10): try: p = make_numbered_dir(root, prefix, mode) - lock_path = create_cleanup_lock(p) - register_cleanup_lock_removal(lock_path) + # 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) except Exception as exc: e = exc else: diff --git a/src/_pytest/tmpdir.py b/src/_pytest/tmpdir.py index 9497a0d49..7b8e42017 100644 --- a/src/_pytest/tmpdir.py +++ b/src/_pytest/tmpdir.py @@ -4,16 +4,30 @@ import re import sys import tempfile from pathlib import Path +from shutil import rmtree +from typing import Generator from typing import Optional +from typing import TYPE_CHECKING +from typing import Union + +if TYPE_CHECKING: + from typing_extensions import Literal + + RetentionType = Literal["all", "failed", "none"] + import attr +from _pytest.config.argparsing import Parser from .pathlib import LOCK_TIMEOUT from .pathlib import make_numbered_dir from .pathlib import make_numbered_dir_with_cleanup from .pathlib import rm_rf +from .pathlib import cleanup_dead_symlink from _pytest.compat import final from _pytest.config import Config +from _pytest.config import ExitCode +from _pytest.config import hookimpl from _pytest.deprecated import check_ispytest from _pytest.fixtures import fixture from _pytest.fixtures import FixtureRequest @@ -31,10 +45,14 @@ class TempPathFactory: _given_basetemp = attr.ib(type=Optional[Path]) _trace = attr.ib() _basetemp = attr.ib(type=Optional[Path]) + _retention_count = attr.ib(type=int) + _retention_policy = attr.ib(type="RetentionType") def __init__( self, given_basetemp: Optional[Path], + retention_count: int, + retention_policy: "RetentionType", trace, basetemp: Optional[Path] = None, *, @@ -49,6 +67,8 @@ class TempPathFactory: # Path.absolute() exists, but it is not public (see https://bugs.python.org/issue25012). self._given_basetemp = Path(os.path.abspath(str(given_basetemp))) self._trace = trace + self._retention_count = retention_count + self._retention_policy = retention_policy self._basetemp = basetemp @classmethod @@ -63,9 +83,23 @@ class TempPathFactory: :meta private: """ check_ispytest(_ispytest) + count = int(config.getini("tmp_path_retention_count")) + if count < 0: + raise ValueError( + f"tmp_path_retention_count must be >= 0. Current input: {count}." + ) + + policy = config.getini("tmp_path_retention_policy") + if policy not in ("all", "failed", "none"): + raise ValueError( + f"tmp_path_retention_policy must be either all, failed, none. Current intput: {policy}." + ) + return cls( given_basetemp=config.option.basetemp, trace=config.trace.get("tmpdir"), + retention_count=count, + retention_policy=policy, _ispytest=True, ) @@ -146,10 +180,13 @@ class TempPathFactory: ) if (rootdir_stat.st_mode & 0o077) != 0: os.chmod(rootdir, rootdir_stat.st_mode & ~0o077) + keep = self._retention_count + if self._retention_policy == "none": + keep = 0 basetemp = make_numbered_dir_with_cleanup( prefix="pytest-", root=rootdir, - keep=3, + keep=keep, lock_timeout=LOCK_TIMEOUT, mode=0o700, ) @@ -184,6 +221,21 @@ def pytest_configure(config: Config) -> None: mp.setattr(config, "_tmp_path_factory", _tmp_path_factory, raising=False) +def pytest_addoption(parser: Parser) -> None: + parser.addini( + "tmp_path_retention_count", + help="How many sessions should we keep the `tmp_path` directories, according to `tmp_path_retention_policy`.", + default=3, + ) + + parser.addini( + "tmp_path_retention_policy", + help="Controls which directories created by the `tmp_path` fixture are kept around, based on test outcome. " + "(all/failed/none)", + default="failed", + ) + + @fixture(scope="session") def tmp_path_factory(request: FixtureRequest) -> TempPathFactory: """Return a :class:`pytest.TempPathFactory` instance for the test session.""" @@ -200,7 +252,9 @@ def _mk_tmp(request: FixtureRequest, factory: TempPathFactory) -> Path: @fixture -def tmp_path(request: FixtureRequest, tmp_path_factory: TempPathFactory) -> Path: +def tmp_path( + request: FixtureRequest, tmp_path_factory: TempPathFactory +) -> Generator[Path, None, None]: """Return a temporary directory path object which is unique to each test function invocation, created as a sub directory of the base temporary directory. @@ -213,4 +267,46 @@ def tmp_path(request: FixtureRequest, tmp_path_factory: TempPathFactory) -> Path The returned object is a :class:`pathlib.Path` object. """ - return _mk_tmp(request, tmp_path_factory) + path = _mk_tmp(request, tmp_path_factory) + yield path + + # Remove the tmpdir if the policy is "failed" and the test passed. + tmp_path_factory: TempPathFactory = request.session.config._tmp_path_factory # type: ignore + policy = tmp_path_factory._retention_policy + if policy == "failed" and request.node._tmp_path_result_call.passed: + # 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(path, ignore_errors=True) + + # remove dead symlink + basetemp = tmp_path_factory._basetemp + if basetemp is None: + return + cleanup_dead_symlink(basetemp) + + +def pytest_sessionfinish(session, exitstatus: Union[int, ExitCode]): + """After each session, remove base directory if all the tests passed, + 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) + + +@hookimpl(tryfirst=True, hookwrapper=True) +def pytest_runtest_makereport(item, call): + outcome = yield + result = outcome.get_result() + setattr(item, "_tmp_path_result_" + result.when, result) diff --git a/testing/test_tmpdir.py b/testing/test_tmpdir.py index 4f7c53847..e681470a4 100644 --- a/testing/test_tmpdir.py +++ b/testing/test_tmpdir.py @@ -42,6 +42,14 @@ class FakeConfig: def get(self, key): return lambda *k: None + def getini(self, name): + if name == "tmp_path_retention_count": + return 3 + elif name == "tmp_path_retention_policy": + return "failed" + else: + assert False + @property def option(self): return self @@ -84,6 +92,53 @@ class TestConfigTmpPath: assert mytemp.exists() assert not mytemp.joinpath("hello").exists() + def test_policy_failed_removes_only_passed_dir(self, pytester: Pytester) -> None: + p = pytester.makepyfile( + """ + def test_1(tmp_path): + assert 0 == 0 + def test_2(tmp_path): + assert 0 == 1 + """ + ) + + pytester.inline_run(p) + 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) == 1 + test_dir = list( + filter( + lambda x: x.is_dir() and not x.is_symlink(), base_dir[0].iterdir() + ) + ) + # Check only the failed one remains + 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 + testdata = [ ("mypath", True), @@ -275,12 +330,12 @@ class TestNumberedDir: assert not lock.exists() - def _do_cleanup(self, tmp_path: Path) -> None: + def _do_cleanup(self, tmp_path: Path, keep: int = 2) -> None: self.test_make(tmp_path) cleanup_numbered_dir( root=tmp_path, prefix=self.PREFIX, - keep=2, + keep=keep, consider_lock_dead_if_created_before=0, ) @@ -289,6 +344,11 @@ class TestNumberedDir: a, b = (x for x in tmp_path.iterdir() if not x.is_symlink()) print(a, b) + def test_cleanup_keep_0(self, tmp_path: Path): + self._do_cleanup(tmp_path, 0) + dir_num = len(list(tmp_path.iterdir())) + assert dir_num == 0 + def test_cleanup_locked(self, tmp_path): p = make_numbered_dir(root=tmp_path, prefix=self.PREFIX) @@ -446,7 +506,7 @@ def test_tmp_path_factory_create_directory_with_safe_permissions( """Verify that pytest creates directories under /tmp with private permissions.""" # Use the test's tmp_path as the system temproot (/tmp). monkeypatch.setenv("PYTEST_DEBUG_TEMPROOT", str(tmp_path)) - tmp_factory = TempPathFactory(None, lambda *args: None, _ispytest=True) + tmp_factory = TempPathFactory(None, 3, "failed", lambda *args: None, _ispytest=True) basetemp = tmp_factory.getbasetemp() # No world-readable permissions. @@ -466,14 +526,14 @@ def test_tmp_path_factory_fixes_up_world_readable_permissions( """ # Use the test's tmp_path as the system temproot (/tmp). monkeypatch.setenv("PYTEST_DEBUG_TEMPROOT", str(tmp_path)) - tmp_factory = TempPathFactory(None, lambda *args: None, _ispytest=True) + tmp_factory = TempPathFactory(None, 3, "failed", lambda *args: None, _ispytest=True) basetemp = tmp_factory.getbasetemp() # Before - simulate bad perms. os.chmod(basetemp.parent, 0o777) assert (basetemp.parent.stat().st_mode & 0o077) != 0 - tmp_factory = TempPathFactory(None, lambda *args: None, _ispytest=True) + tmp_factory = TempPathFactory(None, 3, "failed", lambda *args: None, _ispytest=True) basetemp = tmp_factory.getbasetemp() # After - fixed.