ledger
232c8af304
marker + .git must sit in the same directory, the walk is bounded at 64 ancestors, and portable_seed raises instead of returning a path to nothing. Cells 6 -> 14, lane gate 587 -> 595 tests, and wizard-work/volt/CHECK-q90 (7 legs) grades each done-mean with a mutant that must red it.
208 lines
6.7 KiB
Python
208 lines
6.7 KiB
Python
"""Cells for tree_root.py -- root boundary, seed existence, import-time reads.
|
|
|
|
Q90, 2026-10-30. Three of these cells exist because the module could not
|
|
fail that way before: a stray ancestor marker was accepted as a root, a
|
|
missing seed was returned as a path, and the import-time guard patched
|
|
``builtins.open`` only, so ``Path(...).read_text()``, ``os.open`` and
|
|
``io.open`` all read at import while it printed a clean verdict.
|
|
|
|
The guard below models six read entry points by name and arms all six. It is
|
|
self-checking in both directions: a clean module must record zero calls, and
|
|
each modelled spelling executed at module scope must be recorded under its
|
|
own label -- so deleting an arm, or narrowing the model, reds this file
|
|
instead of silently passing it.
|
|
"""
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from beam_assert import fail, raises
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
from tree_root import find_tree_root, assert_inside_tree, portable_seed
|
|
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
# The read entry points this lane models. Asserted against _SPELLINGS so the
|
|
# model cannot be narrowed without a red cell.
|
|
_READ_MODEL = [
|
|
"builtins.open",
|
|
"io.open",
|
|
"os.open",
|
|
"Path.open",
|
|
"Path.read_text",
|
|
"Path.read_bytes",
|
|
]
|
|
|
|
# One import-time statement per modelled entry point. Each must be recorded
|
|
# under exactly its own label.
|
|
_SPELLINGS = {
|
|
"builtins.open": "open(__file__)\n",
|
|
"io.open": "import io\nio.open(__file__)\n",
|
|
"os.open": "import os\nos.open(__file__, os.O_RDONLY)\n",
|
|
"Path.open": "from pathlib import Path\nPath(__file__).open()\n",
|
|
"Path.read_text": "from pathlib import Path\nPath(__file__).read_text()\n",
|
|
"Path.read_bytes": "from pathlib import Path\nPath(__file__).read_bytes()\n",
|
|
}
|
|
|
|
|
|
def _guarded_import(src, name="guarded_module.py"):
|
|
"""Exec ``src`` as a module with every modelled read entry point armed.
|
|
|
|
Returns the labels of the reads attempted during the exec. The spy raises
|
|
RuntimeError with a marker string instead of touching the disk, so an
|
|
attempted read is recorded and stopped rather than performed.
|
|
"""
|
|
import builtins
|
|
import io
|
|
from pathlib import Path
|
|
|
|
calls = []
|
|
armed = []
|
|
|
|
def arm(obj, attr, label):
|
|
orig = getattr(obj, attr)
|
|
armed.append((obj, attr, orig))
|
|
|
|
def spy(*a, **k):
|
|
calls.append(label)
|
|
raise RuntimeError("import-time I/O detected: %s" % label)
|
|
|
|
setattr(obj, attr, spy)
|
|
|
|
arm(builtins, "open", "builtins.open")
|
|
arm(io, "open", "io.open")
|
|
arm(os, "open", "os.open")
|
|
for attr in ("open", "read_text", "read_bytes"):
|
|
arm(Path, attr, "Path." + attr)
|
|
try:
|
|
try:
|
|
exec(compile(src, name, "exec"),
|
|
{"__name__": "guarded_" + os.path.basename(name),
|
|
"__file__": name})
|
|
except RuntimeError as e:
|
|
if "import-time I/O detected" not in str(e):
|
|
raise
|
|
finally:
|
|
for obj, attr, orig in reversed(armed):
|
|
setattr(obj, attr, orig)
|
|
return calls
|
|
|
|
|
|
def _tree(marker=True, git=True):
|
|
"""A temp root with the marker and, optionally, a .git entry."""
|
|
d = tempfile.mkdtemp()
|
|
if marker:
|
|
open(os.path.join(d, "WORK-QUEUE.md"), "w").write("x")
|
|
if git:
|
|
os.makedirs(os.path.join(d, ".git"))
|
|
return d
|
|
|
|
|
|
def test_find_root_walks_up():
|
|
root = _tree()
|
|
deep = os.path.join(root, "a", "b", "c")
|
|
os.makedirs(deep)
|
|
f = os.path.join(deep, "m.py")
|
|
open(f, "w").write("pass")
|
|
assert find_tree_root(f) == os.path.realpath(root)
|
|
|
|
|
|
def test_find_root_missing_raises():
|
|
d = tempfile.mkdtemp()
|
|
with raises(FileNotFoundError):
|
|
find_tree_root(d)
|
|
|
|
|
|
def test_find_root_ignores_decoy_marker():
|
|
"""A marker without .git is not a root, even when it is nearer than one."""
|
|
root = _tree()
|
|
decoy = os.path.join(root, "vendor", "drop")
|
|
os.makedirs(os.path.join(decoy, "n", "x"))
|
|
open(os.path.join(decoy, "WORK-QUEUE.md"), "w").write("x")
|
|
start = os.path.join(decoy, "n", "x", "m.py")
|
|
open(start, "w").write("pass")
|
|
assert find_tree_root(start) == os.path.realpath(root)
|
|
|
|
|
|
def test_find_root_rejects_lone_marker():
|
|
"""With no real root above it, a lone marker raises instead of winning."""
|
|
d = _tree(marker=True, git=False)
|
|
with raises(FileNotFoundError):
|
|
find_tree_root(d)
|
|
|
|
|
|
def test_find_root_respects_depth_bound():
|
|
root = _tree()
|
|
deep = root
|
|
for i in range(6):
|
|
deep = os.path.join(deep, "d%d" % i)
|
|
os.makedirs(deep)
|
|
assert find_tree_root(deep) == os.path.realpath(root)
|
|
with raises(FileNotFoundError):
|
|
find_tree_root(deep, max_depth=3)
|
|
|
|
|
|
def test_find_root_require_git_opt_out():
|
|
"""A marker-only tree is reachable, but only when the caller asks for it."""
|
|
d = _tree(marker=True, git=False)
|
|
assert find_tree_root(d, require_git=False) == os.path.realpath(d)
|
|
|
|
|
|
def test_assert_inside_ok():
|
|
root = tempfile.mkdtemp()
|
|
p = os.path.join(root, "inbox", "seed.json")
|
|
os.makedirs(os.path.dirname(p), exist_ok=True)
|
|
open(p, "w").write("{}\n")
|
|
assert assert_inside_tree(p, root) == os.path.realpath(p)
|
|
|
|
|
|
def test_assert_inside_escape_raises():
|
|
root = tempfile.mkdtemp()
|
|
with raises(ValueError):
|
|
assert_inside_tree("/etc/passwd", root)
|
|
|
|
|
|
def test_portable_seed_inside():
|
|
root = tempfile.mkdtemp()
|
|
rel = os.path.join("inbox", "seed-2026-09-26.json")
|
|
os.makedirs(os.path.dirname(os.path.join(root, rel)))
|
|
open(os.path.join(root, rel), "w").write("{}\n")
|
|
p = portable_seed(root)
|
|
assert p == os.path.realpath(os.path.join(root, rel))
|
|
|
|
|
|
def test_portable_seed_missing_raises():
|
|
"""The seed that is not there must raise, not be returned as a path."""
|
|
root = tempfile.mkdtemp()
|
|
os.makedirs(os.path.join(root, "inbox"))
|
|
with raises(FileNotFoundError):
|
|
portable_seed(root)
|
|
|
|
|
|
def test_portable_seed_escape_raises():
|
|
root = tempfile.mkdtemp()
|
|
with raises(ValueError):
|
|
portable_seed(root, rel="../outside.json")
|
|
|
|
|
|
def test_import_no_io():
|
|
src = open(os.path.join(_HERE, "tree_root.py")).read()
|
|
calls = _guarded_import(src, "tree_root.py")
|
|
assert calls == [], "import-time reads in tree_root.py: %r" % (calls,)
|
|
|
|
|
|
def test_import_io_model_catches_each_spelling():
|
|
"""Every modelled spelling must be caught, under its own label."""
|
|
misses = []
|
|
for label, src in _SPELLINGS.items():
|
|
calls = _guarded_import(src, "spelling.py")
|
|
if calls != [label]:
|
|
misses.append("%s -> %r" % (label, calls))
|
|
assert not misses, "read spelling(s) survived the guard: %s" % ("; ".join(misses))
|
|
|
|
|
|
def test_import_io_model_is_exactly_the_listed_set():
|
|
assert sorted(_SPELLINGS) == sorted(_READ_MODEL), (
|
|
"guard model drifted: %r vs %r" % (sorted(_SPELLINGS), sorted(_READ_MODEL))
|
|
)
|