Stop using atexit and add integration tests for tmpdir fixture

src/_pytest/pathlib.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import atexit
 from collections.abc import Callable
 from collections.abc import Iterable
 from collections.abc import Iterator
@@ -260,10 +259,8 @@ def create_cleanup_lock(p: Path) -> Path:
         return lock_path
 
 
-def register_cleanup_lock_removal(
-    lock_path: Path, register: Any = atexit.register
-) -> Any:
-    """Register a cleanup function for removing a lock, by default on atexit."""
+def register_cleanup_lock_removal(lock_path: Path, register: Any) -> Any:
+    """Register a cleanup function for removing a lock."""
     pid = os.getpid()
 
     def cleanup_on_exit(lock_path: Path = lock_path, original_pid: int = pid) -> None:
@@ -375,27 +372,43 @@ def cleanup_numbered_dir(
 
 
 def make_numbered_dir_with_cleanup(
+    *,
     root: Path,
     prefix: str,
+    mode: int,
     keep: int,
     lock_timeout: float,
-    mode: int,
+    register: Any,
 ) -> Path:
-    """Create a numbered dir with a cleanup lock and remove old ones."""
+    """Create a numbered dir and register its cleanup.
+
+    Similar to make_numbered_dir, but also maintains a lock file indicating that
+    the directory is currently in use, and registers the cleanup of the lock and
+    of stale numbered directories.
+
+    :param keep:
+        The number of sessions to retain the directory.
+    :param lock_timeout:
+        In case of a crash, the lock remains "stuck". The timeout is a time
+        limit after which the lock is considered stale and can be removed.
+    :param register:
+        Called as register(cleanup_func, params...). Should schedule to call
+        passed cleanup functions on session finish.
+    """
     e = None
     for i in range(10):
         try:
             p = make_numbered_dir(root, prefix, mode)
             # 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)
         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,

src/_pytest/tmpdir.py

@@ -3,7 +3,9 @@
 
 from __future__ import annotations
 
+import atexit
 from collections.abc import Generator
+from contextlib import ExitStack
 import dataclasses
 import os
 from pathlib import Path
@@ -74,6 +76,9 @@ def __init__(
         self._retention_count = retention_count
         self._retention_policy = retention_policy
         self._basetemp = basetemp
+        # Register cleanups for session finish. Also called atexit as a last
+        # resort if sessionfinish for some reason doesn't happen.
+        self._exit_stack = ExitStack()
 
     @classmethod
     def from_config(
@@ -211,7 +216,13 @@ def getbasetemp(self) -> Path:
                 keep=keep,
                 lock_timeout=LOCK_TIMEOUT,
                 mode=0o700,
+                register=self._exit_stack.callback,
             )
+            # Ensure that the cleanup is called on exit (#1120 possibly?).
+            # But if the exit stack is closed manually (as it normally should),
+            # unregister the atexit to avoid pile up.
+            atexit.register(self._exit_stack.close)
+            self._exit_stack.callback(atexit.unregister, self._exit_stack.close)
         assert basetemp is not None, basetemp
         self._basetemp = basetemp
         self._trace("new basetemp", basetemp)
@@ -325,6 +336,9 @@ def pytest_sessionfinish(session, exitstatus: int | ExitCode):
     if basetemp.is_dir():
         cleanup_dead_symlinks(basetemp)
 
+    # Run the numbered dirs and lock file cleanups registered on the ExitStack.
+    tmp_path_factory._exit_stack.close()
+
 
 @hookimpl(wrapper=True, tryfirst=True)
 def pytest_runtest_makereport(

testing/test_tmpdir.py

@@ -96,6 +96,73 @@ def test_1(tmp_path):
         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(
             """
@@ -183,10 +250,8 @@ def test_fixt(fixt):
         # 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(