personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3"""Pytest fixture-leak detector.
4
5Captures `git status --porcelain -- tests/fixtures/` at session start and
6diffs at session end. Fails the session with a named-path error when tests
7leave the fixture tree dirty.
8"""
9
10from __future__ import annotations
11
12import os
13import subprocess
14import sys
15import tempfile
16from pathlib import Path
17
18_TMPDIR_FALLBACK_NOTICE: str | None = None
19
20
21def _apply_tmpdir_fallback() -> None:
22 """Route tmp dirs to /var/tmp when TMPDIR is not exported.
23
24 make test sets TMPDIR=/var/tmp at the shell level (see Makefile). Direct
25 pytest invocations (.venv/bin/pytest, python -m pytest, uv run pytest)
26 bypass that. This prelude closes the gap by setting TMPDIR and
27 tempfile.tempdir at module-import time, before pytest builds its
28 tmp_path_factory.
29 """
30 global _TMPDIR_FALLBACK_NOTICE
31
32 if "TMPDIR" in os.environ:
33 return
34
35 # Test-only override so unwritable-target branch can be exercised
36 # without chmod 000 /var/tmp.
37 target = os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_TARGET", "/var/tmp")
38
39 if not (os.path.isdir(target) and os.access(target, os.W_OK)):
40 if os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED") != "1":
41 _TMPDIR_FALLBACK_NOTICE = (
42 f"solstone: pytest invoked without TMPDIR export and fallback "
43 f"target {target} is not writable; leaving TMPDIR unset.\n"
44 )
45 os.environ["_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED"] = "1"
46 return
47
48 os.environ["TMPDIR"] = target
49 tempfile.tempdir = target
50
51 if os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED") != "1":
52 _TMPDIR_FALLBACK_NOTICE = (
53 f"solstone: pytest invoked without TMPDIR export; routing tmp dirs "
54 f"to {target}. Prefer 'make test' to set TMPDIR at the shell level.\n"
55 )
56 os.environ["_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED"] = "1"
57
58
59_apply_tmpdir_fallback()
60
61_FIXTURE_ROOT = "tests/fixtures"
62_BASELINE: set[tuple[str, str]] | None = None
63_GIT_AVAILABLE = True
64
65
66def _capture_status(repo_root: Path) -> set[tuple[str, str]] | None:
67 """Return the set of (status_XY, path) tuples from fixture-tree git status.
68
69 Returns None when git is unavailable or the command fails (e.g. not a git
70 repo).
71 """
72 try:
73 result = subprocess.run(
74 ["git", "status", "--porcelain", "--", _FIXTURE_ROOT],
75 cwd=repo_root,
76 capture_output=True,
77 text=True,
78 check=False,
79 )
80 except FileNotFoundError:
81 return None
82 if result.returncode != 0:
83 return None
84
85 entries: set[tuple[str, str]] = set()
86 for line in result.stdout.splitlines():
87 if len(line) < 4:
88 continue
89 status = line[:2]
90 path = line[3:]
91 if " -> " in path:
92 path = path.split(" -> ", 1)[1]
93 entries.add((status, path))
94 return entries
95
96
97def _format_leak_message(new_entries: set[tuple[str, str]]) -> str:
98 lines = [f" {status} {path}" for status, path in sorted(new_entries)]
99 return (
100 "\n"
101 "solstone fixture-leak detector: tests left tests/fixtures/ dirty\n"
102 + "\n".join(lines)
103 + "\n\n"
104 "To fix, use one of these isolation mechanisms:\n"
105 " - journal_copy fixture (tests/conftest.py:188) — copies tracked fixtures to tmp_path\n"
106 " - point SOLSTONE_JOURNAL at a tmp_path directly\n"
107 " - mock the subprocess/write path so code never touches tests/fixtures/\n"
108 "\n"
109 "Prior incidents: f6f382a6, 2996e072\n"
110 )
111
112
113def pytest_sessionstart(session):
114 global _BASELINE, _GIT_AVAILABLE, _TMPDIR_FALLBACK_NOTICE
115
116 if _TMPDIR_FALLBACK_NOTICE is not None:
117 sys.stderr.write(_TMPDIR_FALLBACK_NOTICE)
118 _TMPDIR_FALLBACK_NOTICE = None
119
120 repo_root = session.config.rootpath
121 _BASELINE = _capture_status(repo_root)
122 if _BASELINE is None:
123 _GIT_AVAILABLE = False
124 sys.stderr.write("solstone fixture-leak detector: git unavailable, skipping\n")
125
126
127def pytest_sessionfinish(session, exitstatus):
128 if not _GIT_AVAILABLE or _BASELINE is None:
129 return
130
131 repo_root = session.config.rootpath
132 current = _capture_status(repo_root)
133 if current is None:
134 return
135
136 new_entries = current - _BASELINE
137 if not new_entries:
138 return
139
140 sys.stderr.write(_format_leak_message(new_entries))
141 session.exitstatus = 1