diff --git a/autohands/check_dataset_allowlist.py b/autohands/check_dataset_allowlist.py index 5b5f8e7..a727923 100644 --- a/autohands/check_dataset_allowlist.py +++ b/autohands/check_dataset_allowlist.py @@ -17,8 +17,30 @@ (Group B, PyAutoBuild#126) — the guard skips it with a notice rather than failing, until that repo opts in. +Leg 2 — the mirror-image failure. The check above asserts that +nothing *generated* got committed. This one asserts that nothing *committed* gets +deleted: `should_simulate()` ends in `shutil.rmtree`, so a script that reaches an +allowlisted dataset directory while `PYAUTO_SMALL_DATASETS=1` is still in force +destroys data the allowlist exists to protect. The library-side stamp guard +(`_is_capped_at_the_current_cap`) cannot cover this in general — it reads +`/data.fits`, so any JSON-only dataset is invisible to it. + +A script opts out of the capped regime with an `__Env__` declaration whose tokens +release `PYAUTO_SMALL_DATASETS` (`full_datasets`, or the superset `real_output`). +The releasing set is derived from `env_config.ENV_DECLARATION_TOKENS`, never +hardcoded, so a future token that releases the var is picked up automatically. + +Matching a call site to a directory is deliberately EXACT, not fuzzy: the +argument is resolved through a restricted AST evaluator (string literals, simple +module-level names, `Path(...) / "..."`, `os.path.join(...)`, plain f-strings). +Anything it cannot resolve is REPORTED AND SKIPPED, never guessed — this gate +runs in `pre_build`, where a false positive blocks a release. Under-reporting is +acceptable and visible; over-reporting is not. The skipped count is always +printed, so a silent partial sweep cannot read as full coverage. + Run from a workspace root. Exit 0 = clean/skipped, 1 = violation. """ +import ast import re import subprocess import sys @@ -49,6 +71,241 @@ def tracked_dataset_files(): return [f for f in out.splitlines() if f and not f.endswith("dataset/.gitignore")] +UNRESOLVED = None + + +def _releasing_tokens(): + """Tokens whose declaration unsets ``PYAUTO_SMALL_DATASETS``. + + Derived from the token map rather than hardcoded as ``{"full_datasets", + "real_output"}``: a token added later that also releases the var must start + protecting scripts without an edit here. Falls back to the known pair only if + the import is unavailable (the guard must never hard-fail on an env_config + refactor -- it would block a release). + """ + try: + from autohands.env_config import ENV_DECLARATION_TOKENS + except Exception: + return {"full_datasets", "real_output"} + + return { + tok + for tok, vars_ in ENV_DECLARATION_TOKENS.items() + if "PYAUTO_SMALL_DATASETS" in vars_ + } + + +def _resolve(node, names): + """Resolve an AST node to a relative path string, or ``UNRESOLVED``. + + A deliberately small grammar. Every unhandled node type returns + ``UNRESOLVED``, which the caller reports and skips -- this feeds a + release-blocking gate, so guessing is worse than not knowing. + """ + if isinstance(node, ast.Constant): + return node.value if isinstance(node.value, str) else UNRESOLVED + + if isinstance(node, ast.Name): + return names.get(node.id, UNRESOLVED) + + # Path("dataset") / "point_source" / name + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div): + left = _resolve(node.left, names) + right = _resolve(node.right, names) + if left is UNRESOLVED or right is UNRESOLVED: + return UNRESOLVED + return f"{left}/{right}" + + if isinstance(node, ast.JoinedStr): + parts = [] + for value in node.values: + piece = _resolve( + value.value if isinstance(value, ast.FormattedValue) else value, names + ) + if piece is UNRESOLVED: + return UNRESOLVED + parts.append(piece) + return "".join(parts) + + if isinstance(node, ast.Call): + func = node.func + name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + + # Path("dataset", "multi_galaxy", dataset_name) -- Path joins ALL its + # arguments, and the multi-argument form is the dominant idiom in the + # workspaces. Handling only the single-argument case left ~31% of call + # sites unresolved. + if name in ("Path", "PosixPath") and node.args: + parts = [] + for arg in node.args: + piece = _resolve(arg, names) + if piece is UNRESOLVED: + return UNRESOLVED + parts.append(piece) + return "/".join(parts) + + # os.path.join(a, b, ...) / path.join(...) / Path.joinpath(...) + if name in ("join", "joinpath"): + parts = [] + if name == "joinpath" and isinstance(func, ast.Attribute): + base = _resolve(func.value, names) + if base is UNRESOLVED: + return UNRESOLVED + parts.append(base) + for arg in node.args: + piece = _resolve(arg, names) + if piece is UNRESOLVED: + return UNRESOLVED + parts.append(piece) + return "/".join(parts) + + # str(x) around an already-resolvable path is common at the call site. + if name == "str" and len(node.args) == 1: + return _resolve(node.args[0], names) + + return UNRESOLVED + + +def _module_assignments(tree): + """Ordered ``(lineno, name, value_node)`` for TOP-LEVEL assignments only. + + Top-level only, because these workspace scripts are straight-line modules: + a binding inside a function, loop or conditional is not knowable from + position alone, so including it would mean guessing. + + Order matters. Reassignment (``dataset_name = "simple"`` then later + ``dataset_name = "simple__no_lens_light"``) is the norm in these scripts, so + a name is resolved against the assignments that precede the call site rather + than being dropped as ambiguous. + """ + out = [] + for node in tree.body: + if not isinstance(node, ast.Assign): + continue + for target in node.targets: + if isinstance(target, ast.Name): + out.append((node.lineno, target.id, node.value)) + return out + + +def _names_before(assignments, lineno): + """Bindings in force at ``lineno``, evaluated in source order.""" + names = {} + for assign_line, name, value_node in assignments: + if assign_line >= lineno: + break + resolved = _resolve(value_node, names) + if resolved is UNRESOLVED: + names.pop(name, None) + else: + names[name] = resolved + return names + + +def _normalise(path_str: str) -> str: + """Collapse a resolved argument to a repo-relative, slash-joined path.""" + cleaned = path_str.replace("\\", "/").strip().strip("/") + parts = [seg for seg in cleaned.split("/") if seg not in ("", ".")] + return "/".join(parts) + + +def should_simulate_sites(py_files): + """Yield ``(file, lineno, resolved_path_or_None)`` per ``should_simulate`` call.""" + for f in py_files: + try: + tree = ast.parse(Path(f).read_text(encoding="utf-8", errors="replace")) + except (SyntaxError, OSError): + continue + assignments = _module_assignments(tree) + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + fname = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + if fname != "should_simulate" or not node.args: + continue + resolved = _resolve(node.args[0], _names_before(assignments, node.lineno)) + yield f, node.lineno, ( + _normalise(resolved) if resolved is not UNRESOLVED else UNRESOLVED + ) + + +def tracked_python_files(): + out = subprocess.run( + ["git", "ls-files", "*.py"], capture_output=True, text=True + ).stdout + return [f for f in out.splitlines() if f] + + +def check_capped_deletion(prefixes, tracked) -> int: + """Leg 2 — allowlisted datasets reachable from a capped ``should_simulate``. + + ``prefixes``/``tracked`` are leg 1's already-computed allowlist and tracked + file list, so this adds no extra git calls beyond the Python file listing. + """ + from autohands.env_config import read_env_declaration + + # The invariant is NOT "the path sits under an allowlist prefix" -- it is + # "rmtree(path) would delete committed files". Those differ, and the prefix + # form over-reports. The shape that exposed it: a workspace commits a handful + # of documentation images directly in `dataset/
/`, while its scripts + # regenerate sibling subdirectories `dataset/
//` that hold + # nothing tracked. Deleting those destroys nothing, but every one of them sits + # under the allowlist prefix, so prefix matching reported them all as + # release-blocking failures. + def deletes_tracked(resolved: str) -> bool: + return any(f == resolved or f.startswith(resolved + "/") for f in tracked) + + releasing = _releasing_tokens() + violations, skipped = [], [] + + for f, lineno, resolved in should_simulate_sites(tracked_python_files()): + try: + tokens = set(read_env_declaration(Path(f)) or []) + except Exception: + tokens = set() + if tokens & releasing: + continue # releases the cap -> can never enter the small regime + if resolved is UNRESOLVED: + skipped.append(f"{f}:{lineno}") + continue + if deletes_tracked(resolved): + violations.append((f, lineno, resolved, sorted(tokens))) + + if skipped: + print( + f"[capped-deletion] {len(skipped)} call site(s) skipped — argument not " + f"statically resolvable, so NOT covered by this check:" + ) + for s in skipped: + print(f" {s}") + + if violations: + print( + f"[capped-deletion] FAIL — should_simulate() would delete COMMITTED " + f"files without releasing PYAUTO_SMALL_DATASETS ({len(violations)}):", + file=sys.stderr, + ) + for f, lineno, resolved, tokens in violations: + declared = " ".join(tokens) if tokens else "" + print(f" {f}:{lineno} -> {resolved} [ENV: {declared}]", file=sys.stderr) + print( + "\nshould_simulate() ends in shutil.rmtree, and each path above holds " + "git-tracked files kept by design (they carry an `!dataset/...` allowlist " + "line). Under PYAUTO_SMALL_DATASETS=1 those files are deleted and replaced " + "with capped-simulator output. Add a releasing token to the script's " + "`__Env__` section (`ENV: full_datasets ...`).", + file=sys.stderr, + ) + return 1 + + print( + f"[capped-deletion] OK — no capped should_simulate() call site would delete " + f"any of the {len(tracked)} committed dataset file(s)." + ) + return 0 + + def main() -> int: prefixes, has_dataset_ignore = allowlist_prefixes(Path(".gitignore")) tracked = tracked_dataset_files() @@ -91,7 +348,10 @@ def allowed(f: str) -> bool: f"[dataset-allowlist] OK — {len(tracked)} tracked dataset files, all within " f"the allowlist ({len(prefixes)} patterns)." ) - return 0 + + # Leg 2 runs only once leg 1 is clean: if committed data is already outside the + # allowlist, "which allowlisted dirs are at risk" is the wrong question to ask. + return check_capped_deletion(prefixes, tracked) if __name__ == "__main__": diff --git a/autohands/env_config.py b/autohands/env_config.py index 5b87d79..5207995 100644 --- a/autohands/env_config.py +++ b/autohands/env_config.py @@ -56,6 +56,13 @@ # exactly the semantics of today's profile `unset:` lists, which is what makes # the later profile->declaration migration a provable no-op (empty resolved-env # diff). The token map (a token may release more than one var): +# NOTE `real_output` is a SUPERSET token -- it releases all four managed vars at +# once. When auditing "which scripts still run capped / on NumPy / in test mode", +# a script declaring `real_output` is already released on every one of them. +# Treating the map as four single-var tokens under-counts the released set and +# reports such a script as an offender; that mis-read cost real time during a +# cross-workspace audit. Derive membership from this map rather than hardcoding +# a token list -- see `check_dataset_allowlist._releasing_tokens`. ENV_DECLARATION_TOKENS: Dict[str, tuple] = { "jax": ("PYAUTO_DISABLE_JAX",), "full_datasets": ("PYAUTO_SMALL_DATASETS",), diff --git a/tests/test_dataset_allowlist_capped_deletion.py b/tests/test_dataset_allowlist_capped_deletion.py new file mode 100644 index 0000000..7c71bb4 --- /dev/null +++ b/tests/test_dataset_allowlist_capped_deletion.py @@ -0,0 +1,221 @@ +"""Regression tests for leg 2 of the dataset-allowlist guard. + +Leg 1 asserts nothing *generated* got committed. Leg 2 asserts nothing +*committed* gets deleted: ``should_simulate`` ends in ``shutil.rmtree``, so a +script that reaches a committed dataset while ``PYAUTO_SMALL_DATASETS=1`` is +still in force destroys data the allowlist exists to protect. + +Two properties matter more than coverage, and both are locked in here: + +- **No false positives.** This runs in ``pre_build``; a spurious failure blocks a + release. The predicate is "rmtree would delete tracked files", NOT "the path + sits under an allowlist prefix" — those differ, and the prefix form flagged six + safe call sites in a real workspace (see the regression test below). +- **No silent under-reporting.** An argument the resolver cannot evaluate is + reported and skipped, never guessed. +""" + +import ast + +from autohands.check_dataset_allowlist import ( + UNRESOLVED, + _module_assignments, + _names_before, + _releasing_tokens, + _resolve, +) + + +def _resolve_call_arg(src: str): + """Resolve the first ``should_simulate`` argument in ``src``.""" + tree = ast.parse(src) + assignments = _module_assignments(tree) + for node in ast.walk(tree): + if isinstance(node, ast.Call): + func = node.func + name = func.attr if isinstance(func, ast.Attribute) else getattr(func, "id", "") + if name == "should_simulate" and node.args: + return _resolve(node.args[0], _names_before(assignments, node.lineno)) + raise AssertionError("no should_simulate call in source") + + +# --- resolver --------------------------------------------------------------- + + +def test_resolves_multi_argument_path(): + """`Path("dataset", "multi_galaxy", name)` is the dominant workspace idiom. + + Handling only the single-argument form left ~31% of the call sites in the + largest workspace unresolved. + """ + src = ( + 'from pathlib import Path\n' + 'dataset_name = "simple"\n' + 'dataset_path = Path("dataset", "multi_galaxy", dataset_name)\n' + 'should_simulate(str(dataset_path))\n' + ) + assert _resolve_call_arg(src) == "dataset/multi_galaxy/simple" + + +def test_resolves_truediv_chain_and_os_path_join(): + div = ( + 'from pathlib import Path\n' + 'p = Path("dataset") / "point_source" / "simple"\n' + 'should_simulate(str(p))\n' + ) + join = ( + 'from os import path\n' + 'name = "simple"\n' + 'p = path.join("dataset", "point_source", name)\n' + 'should_simulate(p)\n' + ) + assert _resolve_call_arg(div) == "dataset/point_source/simple" + assert _resolve_call_arg(join) == "dataset/point_source/simple" + + +def test_reassignment_resolves_to_the_binding_in_force_at_the_call_site(): + """These scripts reassign `dataset_name` between sections; each call site + must see the value above it, not the file's last one.""" + src = ( + 'from pathlib import Path\n' + 'dataset_name = "first"\n' + 'should_simulate(str(Path("dataset", dataset_name)))\n' + 'dataset_name = "second"\n' + ) + assert _resolve_call_arg(src) == "dataset/first" + + +def test_unknown_expression_is_unresolved_not_guessed(): + """A name the resolver cannot evaluate must yield UNRESOLVED — the caller + reports and skips it rather than matching on a partial path.""" + src = ( + 'from pathlib import Path\n' + 'dataset_name = compute_name()\n' + 'should_simulate(str(Path("dataset", dataset_name)))\n' + ) + assert _resolve_call_arg(src) is UNRESOLVED + + +def test_assignment_inside_a_function_is_not_treated_as_module_scope(): + """Only top-level bindings are positionally knowable.""" + src = ( + 'from pathlib import Path\n' + 'def f():\n' + ' dataset_name = "inner"\n' + 'should_simulate(str(Path("dataset", dataset_name)))\n' + ) + assert _resolve_call_arg(src) is UNRESOLVED + + +# --- releasing-token derivation --------------------------------------------- + + +def test_releasing_tokens_derived_from_the_token_map_not_hardcoded(): + """Must include the superset token `real_output`, which releases all four + managed vars — miscounting it as non-releasing is the easy error.""" + from autohands.env_config import ENV_DECLARATION_TOKENS + + releasing = _releasing_tokens() + + assert "full_datasets" in releasing + assert "real_output" in releasing + assert "real_plots" not in releasing + assert releasing == { + tok + for tok, vars_ in ENV_DECLARATION_TOKENS.items() + if "PYAUTO_SMALL_DATASETS" in vars_ + } + + +# --- the containment predicate (end-to-end) --------------------------------- + + +def _run_check(tmp_path, monkeypatch, script_src, tracked, capsys): + """Drive check_capped_deletion over one synthetic script.""" + from autohands import check_dataset_allowlist as guard + + script = tmp_path / "script.py" + script.write_text(script_src) + monkeypatch.setattr(guard, "tracked_python_files", lambda: [str(script)]) + monkeypatch.chdir(tmp_path) + + code = guard.check_capped_deletion(["dataset/overview"], tracked) + return code, capsys.readouterr() + + +def test_sibling_dir_holding_no_tracked_files_is_not_a_violation( + tmp_path, monkeypatch, capsys +): + """The real shape that exposed the bad predicate: a workspace commits doc + images directly in `dataset/
/`, while its scripts regenerate a + sibling `dataset/
//` that holds nothing tracked. + + Deleting that destroys nothing. Prefix matching against the allowlist called + all six such call sites release-blocking failures; containment does not. + """ + src = ( + 'from pathlib import Path\n' + 'p = Path("dataset", "overview", "imaging_ci", "uniform")\n' + 'should_simulate(str(p))\n' + ) + tracked = ["dataset/overview/ccd.gif", "dataset/overview/what_is_cti.png"] + + code, captured = _run_check(tmp_path, monkeypatch, src, tracked, capsys) + + assert code == 0 + assert "FAIL" not in captured.out + captured.err + + +def test_path_holding_tracked_files_without_a_releasing_token_fails( + tmp_path, monkeypatch, capsys +): + """The originating bug's shape: the resolved path itself holds tracked files.""" + src = ( + '"""\n__Env__\n\nENV: real_plots\n"""\n' + 'from pathlib import Path\n' + 'p = Path("dataset", "overview")\n' + 'should_simulate(str(p))\n' + ) + tracked = ["dataset/overview/ccd.gif"] + + code, captured = _run_check(tmp_path, monkeypatch, src, tracked, capsys) + + assert code == 1 + assert "FAIL" in captured.err + assert "dataset/overview" in captured.err + + +def test_releasing_token_exempts_an_otherwise_failing_call_site( + tmp_path, monkeypatch, capsys +): + """Same script, plus `full_datasets` — the shape of the shipped fix.""" + src = ( + '"""\n__Env__\n\nENV: full_datasets real_plots\n"""\n' + 'from pathlib import Path\n' + 'p = Path("dataset", "overview")\n' + 'should_simulate(str(p))\n' + ) + tracked = ["dataset/overview/ccd.gif"] + + code, captured = _run_check(tmp_path, monkeypatch, src, tracked, capsys) + + assert code == 0 + assert "FAIL" not in captured.out + captured.err + + +def test_unresolvable_call_site_is_reported_and_skipped( + tmp_path, monkeypatch, capsys +): + """Under-reporting is acceptable; a SILENT partial sweep is not.""" + src = ( + 'from pathlib import Path\n' + 'p = Path("dataset", compute())\n' + 'should_simulate(str(p))\n' + ) + tracked = ["dataset/overview/ccd.gif"] + + code, captured = _run_check(tmp_path, monkeypatch, src, tracked, capsys) + + assert code == 0 + assert "skipped" in captured.out + assert "script.py:3" in captured.out