Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions changelog/14935.feature.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
Added a new ``tmp_path_layout`` ini option controlling the directory
layout under ``pytest-of-<user>/``.

* ``"flat"`` (default): unchanged historical behaviour.
* ``"per-rootdir"``: numbered dirs are nested one level deeper, under a
stable token derived from the pytest ``rootpath``
(``<name>-<8-char-hash>``). ``tmp_path_retention_count`` then applies
per project, so runs in one checkout no longer evict scratch created
by runs in another. Particularly useful for users who work in multiple
git worktrees, multiple clones of the same repository, or several
unrelated projects under one user account.
5 changes: 5 additions & 0 deletions changelog/14935.improvement.1.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
Each pytest session directory (``pytest-of-<user>/pytest-N``) now contains
a ``.origin`` sidecar recording the ``rootpath``, pytest version, PID, and
hostname of the session that created it. This lets external cleanup tooling
attribute a session directory to a specific project by reading a file,
rather than by inspecting live process state.
102 changes: 102 additions & 0 deletions src/_pytest/tmpdir.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,12 @@
from collections.abc import Generator
from contextlib import ExitStack
import dataclasses
import hashlib
import os
from pathlib import Path
import re
from shutil import rmtree
import socket
import stat
import tempfile
from typing import Any
Expand All @@ -22,6 +24,7 @@
from .pathlib import make_numbered_dir
from .pathlib import make_numbered_dir_with_cleanup
from .pathlib import rm_rf
import _pytest
from _pytest.compat import get_user_id
from _pytest.config import Config
from _pytest.config import ExitCode
Expand All @@ -38,6 +41,7 @@

tmppath_result_key = StashKey[dict[str, bool]]()
RetentionType = Literal["all", "failed", "none"]
LayoutType = Literal["flat", "per-rootdir"]


@final
Expand All @@ -53,6 +57,8 @@ class TempPathFactory:
_basetemp: Path | None
_retention_count: int
_retention_policy: RetentionType
_rootpath: Path | None
_layout: LayoutType

def __init__(
self,
Expand All @@ -61,6 +67,8 @@ def __init__(
retention_policy: RetentionType,
trace,
basetemp: Path | None = None,
rootpath: Path | None = None,
layout: LayoutType = "flat",
*,
_ispytest: bool = False,
) -> None:
Expand All @@ -76,6 +84,8 @@ def __init__(
self._retention_count = retention_count
self._retention_policy = retention_policy
self._basetemp = basetemp
self._rootpath = rootpath
self._layout = layout
# Register cleanups for session finish. Also called atexit as a last
# resort if sessionfinish for some reason doesn't happen.
self._exit_stack = ExitStack()
Expand All @@ -99,12 +109,15 @@ def from_config(
)

policy: RetentionType = config.getini("tmp_path_retention_policy")
layout: LayoutType = config.getini("tmp_path_layout")

return cls(
given_basetemp=config.option.basetemp,
trace=config.trace.get("tmpdir"),
retention_count=count,
retention_policy=policy,
rootpath=config.rootpath,
layout=layout,
_ispytest=True,
)

Expand Down Expand Up @@ -203,6 +216,14 @@ def getbasetemp(self) -> Path:
rootdir_stat.st_mode & ~0o077,
follow_symlinks=chmod_follow_symlinks,
)
# Per-rootdir layout: nest the numbered dirs one level deeper
# under a stable token derived from the rootpath. Retention then
# applies per project instead of across every rootdir a user has
# ever run pytest against. Opt-in via ``tmp_path_layout``.
if self._layout == "per-rootdir" and self._rootpath is not None:
token = _rootdir_token(self._rootpath)
rootdir = rootdir / token
rootdir.mkdir(mode=0o700, exist_ok=True)
keep = self._retention_count
if self._retention_policy == "none":
keep = 0
Expand All @@ -221,6 +242,7 @@ def getbasetemp(self) -> Path:
self._exit_stack.callback(atexit.unregister, self._exit_stack.close)
assert basetemp is not None, basetemp
self._basetemp = basetemp
_write_origin_sidecar(basetemp, self._rootpath)
self._trace("new basetemp", basetemp)
return basetemp

Expand All @@ -237,6 +259,71 @@ def get_user() -> str | None:
return None


ORIGIN_FILENAME = ".origin"


_ROOTDIR_SLUG_MAX = 32
_ROOTDIR_HASH_LEN = 8


def _rootdir_slug(rootpath: Path) -> str:
"""Return a portable, filesystem-safe slug for a rootpath's name."""
name = rootpath.name or "root"
slug = "".join(c if c.isalnum() or c in "-_." else "-" for c in name)
slug = slug.strip("-.") or "root"
return slug[:_ROOTDIR_SLUG_MAX]


def _rootdir_token(rootpath: Path) -> str:
"""Return a stable ``<slug>-<hash>`` token for a rootpath.

The slug keeps the directory name human-readable in ``/tmp`` listings; the
hash disambiguates homonyms (``~/work/foo`` vs ``~/play/foo``) so two
projects with the same basename get distinct subdirectories.
"""
digest = hashlib.blake2b(
str(rootpath).encode("utf-8"), digest_size=_ROOTDIR_HASH_LEN // 2
).hexdigest()
return f"{_rootdir_slug(rootpath)}-{digest}"


def _write_origin_sidecar(basetemp: Path, rootpath: Path | None) -> None:
"""Write an ``.origin`` file next to ``.lock`` recording who owns this
session directory.

The file is a best-effort record used by external tooling (workstation
janitors, CI cleanup, editor temp sweeps) that needs to attribute a
session directory to a specific project without walking ``/proc`` or
``lsof``. Writing it must never fail a test run, so all errors are
swallowed.

Fields are ``key=value`` lines, newline-terminated, UTF-8:

- ``rootpath``: absolute pytest rootpath, if known.
- ``version``: pytest version string.
- ``pid``: PID of the process that created this basetemp.
- ``host``: hostname of the machine that created this basetemp.
"""
try:
lines = []
if rootpath is not None:
lines.append(f"rootpath={rootpath}")
lines.append(f"version={_pytest.__version__}")
lines.append(f"pid={os.getpid()}")
try:
host = socket.gethostname()
except OSError:
host = ""
if host:
lines.append(f"host={host}")
lines.append("")
(basetemp / ORIGIN_FILENAME).write_text("\n".join(lines), encoding="utf-8")
except OSError:
# Best-effort: never break a test run because a sidecar could not
# be written (read-only mount, permissions, full disk, etc.).
pass


def pytest_configure(config: Config) -> None:
"""Create a TempPathFactory and attach it to the config object.

Expand Down Expand Up @@ -266,6 +353,21 @@ def pytest_addoption(parser: Parser) -> None:
default="all",
)

parser.addini(
"tmp_path_layout",
help=(
"Layout for numbered dirs under ``pytest-of-<user>/``. "
"'flat' (default) keeps the historical shared pool, where "
"``tmp_path_retention_count`` applies across every project that "
"shares a user account. 'per-rootdir' nests numbered dirs under "
"a stable token derived from ``rootpath``, so retention becomes "
"scoped per project and runs in one checkout do not evict scratch "
"from another."
),
type=LayoutType,
default="flat",
)


@fixture(scope="session")
def tmp_path_factory(request: FixtureRequest) -> TempPathFactory:
Expand Down
Loading
Loading