beam: tree_root helper — portable seed path + import-time I/O guard (Q86 pattern)
This commit is contained in:
1 parent
4ba24d47ec
commit
29e0133a29
2 files changed
+104
No files matched your search
@@ -0,0 +1,68 @@
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
from tree_root import find_tree_root, assert_inside_tree, portable_seed
|
||||
|
||||
|
||||
def test_find_root_walks_up():
|
||||
d = tempfile.mkdtemp()
|
||||
root = os.path.join(d, "repo")
|
||||
os.makedirs(os.path.join(root, "inbox"))
|
||||
open(os.path.join(root, "WORK-QUEUE.md"), "w").write("x")
|
||||
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 pytest.raises(FileNotFoundError):
|
||||
find_tree_root(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("{}")
|
||||
assert assert_inside_tree(p, root) == os.path.realpath(p)
|
||||
|
||||
|
||||
def test_assert_inside_escape_raises():
|
||||
root = tempfile.mkdtemp()
|
||||
with pytest.raises(ValueError):
|
||||
assert_inside_tree("/etc/passwd", root)
|
||||
|
||||
|
||||
def test_portable_seed_inside():
|
||||
root = tempfile.mkdtemp()
|
||||
os.makedirs(os.path.join(root, "inbox"))
|
||||
p = portable_seed(root)
|
||||
assert p.startswith(os.path.realpath(root) + os.sep)
|
||||
|
||||
|
||||
def test_import_no_io():
|
||||
src = open(os.path.join(os.path.dirname(__file__), "tree_root.py")).read()
|
||||
import builtins
|
||||
orig = builtins.open
|
||||
calls = []
|
||||
def guard(*a, **k):
|
||||
calls.append(a)
|
||||
raise RuntimeError("import-time I/O detected")
|
||||
ns = {}
|
||||
builtins.open = guard
|
||||
try:
|
||||
try:
|
||||
exec(compile(src, "tree_root.py", "exec"), ns)
|
||||
assert not calls, "import-time open calls: %r" % calls
|
||||
except RuntimeError as e:
|
||||
if "import-time I/O" in str(e):
|
||||
pytest.fail("module performed import-time I/O")
|
||||
raise
|
||||
finally:
|
||||
builtins.open = orig
|
||||
@@ -0,0 +1,36 @@
|
||||
"""tree_root.py - derive a repo root from __file__ instead of a host-absolute path."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
|
||||
def find_tree_root(start, marker="WORK-QUEUE.md"):
|
||||
cur = os.path.abspath(start)
|
||||
if os.path.isfile(cur):
|
||||
cur = os.path.dirname(cur)
|
||||
while True:
|
||||
if os.path.isfile(os.path.join(cur, marker)):
|
||||
return cur
|
||||
parent = os.path.dirname(cur)
|
||||
if parent == cur:
|
||||
raise FileNotFoundError("marker %r not found in any ancestor of %s" % (marker, start))
|
||||
cur = parent
|
||||
|
||||
|
||||
def assert_inside_tree(path, root):
|
||||
rp = os.path.realpath(path)
|
||||
rr = os.path.realpath(root)
|
||||
if rp == rr or rp.startswith(rr + os.sep):
|
||||
return rp
|
||||
raise ValueError("path %s escapes tree root %s" % (rp, rr))
|
||||
|
||||
|
||||
def portable_seed(root, rel="inbox/seed-2026-09-26.json"):
|
||||
return assert_inside_tree(os.path.join(root, rel), root)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
r = find_tree_root(__file__)
|
||||
p = portable_seed(r)
|
||||
print("root=", r)
|
||||
print("seed=", p)
|
||||
Reference in new issue
Block a user