solpbc.org
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
18import pytest
19
20_TMPDIR_FALLBACK_NOTICE: str | None = None
21
22
23def _apply_tmpdir_fallback() -> None:
24 """Route tmp dirs to /var/tmp when TMPDIR is not exported.
25
26 make test sets TMPDIR=/var/tmp at the shell level (see Makefile). Direct
27 pytest invocations (.venv/bin/pytest, python -m pytest, uv run pytest)
28 bypass that. This prelude closes the gap by setting TMPDIR and
29 tempfile.tempdir at module-import time, before pytest builds its
30 tmp_path_factory.
31 """
32 global _TMPDIR_FALLBACK_NOTICE
33
34 if "TMPDIR" in os.environ:
35 return
36
37 # Test-only override so unwritable-target branch can be exercised
38 # without chmod 000 /var/tmp.
39 target = os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_TARGET", "/var/tmp")
40
41 if not (os.path.isdir(target) and os.access(target, os.W_OK)):
42 if os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED") != "1":
43 _TMPDIR_FALLBACK_NOTICE = (
44 f"solstone: pytest invoked without TMPDIR export and fallback "
45 f"target {target} is not writable; leaving TMPDIR unset.\n"
46 )
47 os.environ["_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED"] = "1"
48 return
49
50 os.environ["TMPDIR"] = target
51 tempfile.tempdir = target
52
53 if os.environ.get("_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED") != "1":
54 _TMPDIR_FALLBACK_NOTICE = (
55 f"solstone: pytest invoked without TMPDIR export; routing tmp dirs "
56 f"to {target}. Prefer 'make test' to set TMPDIR at the shell level.\n"
57 )
58 os.environ["_SOLSTONE_TMPDIR_FALLBACK_NOTIFIED"] = "1"
59
60
61_apply_tmpdir_fallback()
62
63_FIXTURE_ROOT = "tests/fixtures"
64_BASELINE: set[tuple[str, str]] | None = None
65_GIT_AVAILABLE = True
66
67
68def _capture_status(repo_root: Path) -> set[tuple[str, str]] | None:
69 """Return the set of (status_XY, path) tuples from fixture-tree git status.
70
71 Returns None when git is unavailable or the command fails (e.g. not a git
72 repo).
73 """
74 try:
75 result = subprocess.run(
76 ["git", "status", "--porcelain", "--", _FIXTURE_ROOT],
77 cwd=repo_root,
78 capture_output=True,
79 text=True,
80 check=False,
81 )
82 except FileNotFoundError:
83 return None
84 if result.returncode != 0:
85 return None
86
87 entries: set[tuple[str, str]] = set()
88 for line in result.stdout.splitlines():
89 if len(line) < 4:
90 continue
91 status = line[:2]
92 path = line[3:]
93 if " -> " in path:
94 path = path.split(" -> ", 1)[1]
95 entries.add((status, path))
96 return entries
97
98
99def _format_leak_message(new_entries: set[tuple[str, str]]) -> str:
100 lines = [f" {status} {path}" for status, path in sorted(new_entries)]
101 return (
102 "\n"
103 "solstone fixture-leak detector: tests left tests/fixtures/ dirty\n"
104 + "\n".join(lines)
105 + "\n\n"
106 "To fix, use one of these isolation mechanisms:\n"
107 " - journal_copy fixture (tests/conftest.py:188) — copies tracked fixtures to tmp_path\n"
108 " - point SOLSTONE_JOURNAL at a tmp_path directly\n"
109 " - mock the subprocess/write path so code never touches tests/fixtures/\n"
110 "\n"
111 "Prior incidents: f6f382a6, 2996e072\n"
112 )
113
114
115def pytest_sessionstart(session):
116 global _BASELINE, _GIT_AVAILABLE, _TMPDIR_FALLBACK_NOTICE
117
118 if hasattr(session.config, "workerinput"):
119 return
120
121 if _TMPDIR_FALLBACK_NOTICE is not None:
122 sys.stderr.write(_TMPDIR_FALLBACK_NOTICE)
123 _TMPDIR_FALLBACK_NOTICE = None
124
125 repo_root = session.config.rootpath
126 _BASELINE = _capture_status(repo_root)
127 if _BASELINE is None:
128 _GIT_AVAILABLE = False
129 sys.stderr.write("solstone fixture-leak detector: git unavailable, skipping\n")
130
131
132def pytest_sessionfinish(session, exitstatus):
133 if hasattr(session.config, "workerinput"):
134 return
135
136 if not _GIT_AVAILABLE or _BASELINE is None:
137 return
138
139 repo_root = session.config.rootpath
140 current = _capture_status(repo_root)
141 if current is None:
142 return
143
144 new_entries = current - _BASELINE
145 if not new_entries:
146 return
147
148 sys.stderr.write(_format_leak_message(new_entries))
149 session.exitstatus = 1
150
151
152@pytest.fixture(autouse=True)
153def _reset_local_slot_sizing_state():
154 """Keep per-process local sizing and admission state from leaking across tests."""
155
156 def _reset() -> None:
157 admission = sys.modules.get("solstone.think.admission")
158 if admission is not None:
159 admission.reset_admission_state()
160 local_server = sys.modules.get("solstone.think.providers.local_server")
161 if local_server is not None:
162 local_server.reset_parallel_slots_cache()
163 fanout_policy = sys.modules.get("solstone.think.providers.fanout_policy")
164 if fanout_policy is not None:
165 fanout_policy.reset_default_cap_log_state()
166 thinking = sys.modules.get("solstone.think.thinking")
167 if thinking is not None:
168 thinking.reset_dispatch_admission_state()
169
170 _reset()
171 yield
172 _reset()
173
174
175@pytest.fixture(autouse=True)
176def _default_local_vulkan_gpu():
177 """Keep local-provider readiness deterministic on GPU-less test hosts."""
178 from solstone.think.providers import local_vulkan
179
180 original_detect_gpus = local_vulkan.detect_gpus
181 local_vulkan.reset_detect_cache()
182
183 def fake_detect_gpus():
184 return [
185 local_vulkan.VulkanDevice(
186 1,
187 "NVIDIA GeForce GTX 1660 Ti",
188 local_vulkan.VK_TYPE_DISCRETE,
189 6390,
190 )
191 ]
192
193 local_vulkan.detect_gpus = fake_detect_gpus
194 yield
195 local_vulkan.detect_gpus = original_detect_gpus
196 local_vulkan.reset_detect_cache()