Add configuration options to control how `tmp_path` directories are kept (#10442)

Close #8141

AUTHORS

@@ -374,6 +374,7 @@ Xixi Zhao
 Xuan Luong
 Xuecong Liao
 Yoav Caspi
+Yusuke Kadowaki
 Yuval Shimon
 Zac Hatfield-Dodds
 Zachary Kneupper

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"`.

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

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)
+            # Only lock the current dir when keep is not 0
-            register_cleanup_lock_removal(lock_path)
+            if keep != 0:
+                lock_path = create_cleanup_lock(p)
+                register_cleanup_lock_removal(lock_path)
         except Exception as exc:
             e = exc
         else:

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)

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.