solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Self-test for scripts/check_journal_io_access.py.
5
6Drives the journal_io access check against throwaway good/bad fixture trees and
7asserts the detector's import-binding discrimination, owner exclusions,
8non-gated symbol handling, and ratcheting allowlist behavior.
9"""
10
11from __future__ import annotations
12
13import importlib.util
14import subprocess
15import sys
16from pathlib import Path
17
18import pytest
19
20REPO_ROOT = Path(__file__).resolve().parents[1]
21SCRIPT = REPO_ROOT / "scripts" / "check_journal_io_access.py"
22
23
24def _load_checker():
25 spec = importlib.util.spec_from_file_location("check_journal_io_access", SCRIPT)
26 assert spec and spec.loader
27 module = importlib.util.module_from_spec(spec)
28 spec.loader.exec_module(module)
29 return module
30
31
32cja = _load_checker()
33
34
35BAD_IMPORT = """\
36# SPDX-License-Identifier: AGPL-3.0-only
37# Copyright (c) 2026 sol pbc
38from solstone.think.journal_io import write_json
39
40
41def persist(path):
42 write_json(path, {"ok": True})
43"""
44
45LOCAL_DISCRIMINATION = """\
46# SPDX-License-Identifier: AGPL-3.0-only
47# Copyright (c) 2026 sol pbc
48
49
50def write_json(path, payload):
51 return path, payload
52
53
54def persist(path):
55 write_json(path, {"ok": True})
56 path.write_text("ok", encoding="utf-8")
57"""
58
59OWNER_IMPORT = """\
60# SPDX-License-Identifier: AGPL-3.0-only
61# Copyright (c) 2026 sol pbc
62from solstone.think.journal_io import write_json
63
64
65def persist(path):
66 write_json(path, {"ok": True})
67"""
68
69NON_GATED_IMPORTS = """\
70# SPDX-License-Identifier: AGPL-3.0-only
71# Copyright (c) 2026 sol pbc
72from solstone.think.journal_io import contained_path, day_path, read_json
73
74
75def inspect(root, path):
76 read_json(path)
77 contained_path(root, path)
78 day_path("20260606")
79"""
80
81SUBMODULE_IMPORT = """\
82# SPDX-License-Identifier: AGPL-3.0-only
83# Copyright (c) 2026 sol pbc
84from solstone.think.journal_io.atomic import write_json
85
86
87def persist(path):
88 write_json(path, {"ok": True})
89"""
90
91NPZ_SUBMODULE_IMPORT = """\
92# SPDX-License-Identifier: AGPL-3.0-only
93# Copyright (c) 2026 sol pbc
94from solstone.think.journal_io.npz import save_npz, update_npz
95
96
97def persist(path, arrays, transform):
98 save_npz(path, arrays, expected_keys=("data",))
99 update_npz(path, transform, expected_keys=("data",))
100"""
101
102
103WRITE_NPZ_SUBMODULE_IMPORT = """\
104# SPDX-License-Identifier: AGPL-3.0-only
105# Copyright (c) 2026 sol pbc
106from solstone.think.journal_io.npz import write_npz
107
108
109def persist(path, arrays):
110 write_npz(path, arrays, expected_keys=("data",))
111"""
112
113
114def _write_file(root: Path, rel: str, content: str) -> None:
115 path = root / rel
116 path.parent.mkdir(parents=True, exist_ok=True)
117 path.write_text(content, encoding="utf-8")
118
119
120@pytest.fixture
121def bad_root(tmp_path) -> Path:
122 root = tmp_path / "bad"
123 _write_file(root, "solstone/apps/badapp/routes.py", BAD_IMPORT)
124 return root
125
126
127@pytest.fixture
128def local_root(tmp_path) -> Path:
129 root = tmp_path / "local"
130 _write_file(root, "solstone/apps/localapp/routes.py", LOCAL_DISCRIMINATION)
131 return root
132
133
134@pytest.fixture
135def owner_root(tmp_path) -> Path:
136 root = tmp_path / "owner"
137 _write_file(root, "solstone/think/entities/saving.py", OWNER_IMPORT)
138 return root
139
140
141@pytest.fixture
142def non_gated_root(tmp_path) -> Path:
143 root = tmp_path / "non-gated"
144 _write_file(root, "solstone/apps/reader/routes.py", NON_GATED_IMPORTS)
145 return root
146
147
148def _run(root: Path) -> subprocess.CompletedProcess:
149 return subprocess.run(
150 [sys.executable, str(SCRIPT), "--root", str(root)],
151 capture_output=True,
152 text=True,
153 )
154
155
156def test_bad_import_exits_one_and_names_file_and_primitive(bad_root):
157 result = _run(bad_root)
158 assert result.returncode == 1, result.stdout + result.stderr
159 assert "solstone/apps/badapp/routes.py" in result.stderr
160 assert "write_json" in result.stderr
161
162
163def test_local_same_name_and_path_write_text_are_not_violations(local_root):
164 result = _run(local_root)
165 assert result.returncode == 0, result.stdout + result.stderr
166 assert "journal-io-access: pass" in result.stdout
167
168
169def test_owner_path_is_exempt(owner_root):
170 result = _run(owner_root)
171 assert result.returncode == 0, result.stdout + result.stderr
172 assert "journal-io-access: pass" in result.stdout
173
174
175def test_non_gated_imports_are_not_violations(non_gated_root):
176 result = _run(non_gated_root)
177 assert result.returncode == 0, result.stdout + result.stderr
178 assert "journal-io-access: pass" in result.stdout
179
180
181def test_submodule_import_is_flagged():
182 findings = cja.scan_source(SUBMODULE_IMPORT)
183 assert findings == [(7, "write_json", "write_json")]
184
185
186def test_npz_submodule_write_imports_are_flagged():
187 findings = cja.scan_source(NPZ_SUBMODULE_IMPORT)
188 assert [(primitive, bound_name) for _lineno, primitive, bound_name in findings] == [
189 ("save_npz", "save_npz"),
190 ("update_npz", "update_npz"),
191 ]
192
193
194def test_write_npz_submodule_import_is_flagged():
195 findings = cja.scan_source(WRITE_NPZ_SUBMODULE_IMPORT)
196 assert ("write_npz", "write_npz") in [
197 (primitive, bound_name) for _lineno, primitive, bound_name in findings
198 ]
199
200
201def test_ratchet_by_file_kind_count(bad_root):
202 # No allowlist -> every violation is new.
203 new, tracked = cja.evaluate(bad_root, {})
204 assert new
205 assert tracked == []
206
207 # Allowlist the exact counts -> green, all tracked.
208 counts = cja.count_violations(bad_root)
209 new_exact, tracked_exact = cja.evaluate(bad_root, counts)
210 assert new_exact == []
211 assert tracked_exact
212
213 # Lower a single (file, kind) count below its actual occurrences -> fails.
214 key = next(iter(counts))
215 ratcheted = dict(counts)
216 ratcheted[key] = counts[key] - 1
217 new_over, _ = cja.evaluate(bad_root, ratcheted)
218 assert new_over