troupe-standup/wizard-work/beam/test_tree_root.py

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