"""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)) )