diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 1c995e4..b339293 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,7 +1,7 @@ { - "modules": 120, + "modules": 122, "counts": { - "load_bearing": 111, + "load_bearing": 113, "standalone_capability": 9 }, "orphaned": [], @@ -401,6 +401,7 @@ "model.specs", "model.taint", "plane.budget", + "plane.tasks.repo_facts", "preprocess", "push.assistance", "push.cache", @@ -528,6 +529,7 @@ "importers": [ "model.partitions", "model.scan", + "plane.tasks.repo_facts", "preprocess", "push.runner" ] @@ -661,6 +663,18 @@ "reason": "ai-service-plane Req 2.3/2.4: metered chat client with usd and calls ceilings; task integration in tasks 3, 6", "importers": [] }, + { + "module": "plane.tasks", + "classification": "load_bearing", + "reason": "ai-service-plane Req 4.1: the task entrypoint package the runner imports by name; runner integration pending (task 3)", + "importers": [] + }, + { + "module": "plane.tasks.repo_facts", + "classification": "load_bearing", + "reason": "ai-service-plane Req 5.1/5.2: source-only functions and callers per file; runner integration pending", + "importers": [] + }, { "module": "policy", "classification": "load_bearing", @@ -707,6 +721,7 @@ "model.partitions", "model.scan", "pairs", + "plane.tasks.repo_facts", "push.runner", "rank", "regress.candidate", @@ -1337,6 +1352,5 @@ "cli" ] } - ], - "note": "Group 8: runner reaches recomputed versioned eligibility and its frozen scorer." + ] } diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 99132c1..c80492d 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -56,6 +56,8 @@ SPEC_OWNED = { "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion", "plane": "ai-service-plane Req 1-4: the service plane package; runner and reconciler integration in tasks 3-4", "plane.budget": "ai-service-plane Req 2.3/2.4: metered chat client with usd and calls ceilings; task integration in tasks 3, 6", + "plane.tasks": "ai-service-plane Req 4.1: the task entrypoint package the runner imports by name; runner integration pending (task 3)", + "plane.tasks.repo_facts": "ai-service-plane Req 5.1/5.2: source-only functions and callers per file; runner integration pending", } STANDALONE = { diff --git a/src/openultrasast/plane/tasks/__init__.py b/src/openultrasast/plane/tasks/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/openultrasast/plane/tasks/repo_facts.py b/src/openultrasast/plane/tasks/repo_facts.py new file mode 100644 index 0000000..74fe397 --- /dev/null +++ b/src/openultrasast/plane/tasks/repo_facts.py @@ -0,0 +1,364 @@ +"""`repo-facts`: the memory layer of the service plane (ai-service-plane, Requirement 5.1 / 5.2). + +Source text only, no model, no git. For every product file: the functions it defines; and for every candidate +function, the call sites in *other* product files (path, 1-based line, enclosing function), capped at 40 and +sorted. The unit of work is one file (Requirement 2.2): a row goes to `units.jsonl` as each file finishes and +a restart skips the rows already there. `facts.json` is assembled from the rows and is byte-identical across +runs over the same tree; `summary.json` is written last and says `done` only when every unit finished. + +The declaration patterns follow `benchmarks/independent/sink_candidates.py`, extended to Java, Go and Ruby; +the product filter follows `model_sinks.py` plus the `[[layout]]` facts (`model.layout`) for the languages that +declare them. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import sys +import traceback +from collections.abc import Iterable, Mapping, Sequence +from pathlib import Path +from typing import Any + +from ...model.layout import is_test_path, is_vendored, layout_facts +from ...preprocess import IGNORED_DIRS, detect_language + +CALLERS_CAP = 40 +MAX_FILE_BYTES = 1_000_000 +LANGUAGES = frozenset({"python", "php", "javascript", "typescript", "java", "go", "ruby"}) +GLOBAL = "" + +_EXCLUDED_DIRS = frozenset( + IGNORED_DIRS | {"vendor", "node_modules", "bower_components", "third_party", "third-party", "dist", "build", "out", "target"} +) +_NOT_PRODUCT = re.compile( + r"(^|/)(tests?|__tests__|__mocks__|spec|specs|docs?|examples?|fixtures?|e2e|cypress|storybook|testdata)/" + r"|\.(test|spec|stories|mock|e2e)\.[cm]?[jt]sx?$" + r"|_test\.(py|go|rb)$|(^|/)test_[^/]*\.(py|rb)$|Test\.java$|Tests\.java$|_spec\.rb$" + r"|\.min\.(js|css)$|\.bundle\.js$" +) + +_JS_MODIFIERS = r"(?:(?:export|default|public|private|protected|static|async|readonly|override)\s+)*" +_JAVA_MODIFIERS = r"(?:(?:public|private|protected|static|final|abstract|synchronized|native|default|strictfp)\s+)*" +DECLARATION: dict[str, re.Pattern[str]] = { + "python": re.compile(r"^\s*(?:async\s+)?def\s+(\w+)\s*\("), + "php": re.compile(r"^\s*(?:(?:public|private|protected|static|final|abstract)\s+)*function\s+&?(\w+)\s*\("), + "javascript": re.compile( + r"^\s*" + _JS_MODIFIERS + r"function\s*\*?\s*(\w+)\s*\(" + r"|^\s*" + _JS_MODIFIERS + r"(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$" + r"|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function\b|\([^)]*\)\s*(?::[^=]+)?=>|\w+\s*=>)" + ), + "java": re.compile( + r"^\s*" + _JAVA_MODIFIERS + r"(?:<[^>]*>\s+)?(?!return\b|new\b|throw\b)[\w<>\[\],.?]+\s+(\w+)\s*\([^;]*\)\s*" + r"(?:throws\s+[\w.,\s]+)?\{?\s*$" + ), + "go": re.compile(r"^\s*func\s+(?:\([^)]*\)\s*)?(\w+)\s*\("), + "ruby": re.compile(r"^\s*def\s+(?:self\.)?(\w+[?!=]?)"), +} +DECLARATION["typescript"] = DECLARATION["javascript"] +KEYWORDS = frozenset( + { + "if", + "for", + "while", + "switch", + "catch", + "return", + "function", + "else", + "do", + "try", + "with", + "elif", + "except", + "new", + "throw", + "super", + "constructor", + "foreach", + "match", + "case", + "unless", + "until", + } +) +_COMMENT_PREFIXES = ("#", "//", "/*", "*", "*/", "--") + + +class TaskFailed(RuntimeError): + """A crash inside `run` that has already been recorded in `summary.json` (status `failed`, traceback).""" + + +def product_files(workspace: Path) -> list[Path]: + """Product source files of the supported languages, sorted; tests, vendored trees, build output, minified + bundles and dot-directories are not product (the classification of `model_sinks.py` and the layout facts).""" + facts = layout_facts() + files: list[Path] = [] + for path in sorted(workspace.rglob("*")): + if path.is_symlink() or not path.is_file(): + continue + relative = path.relative_to(workspace) + parts = relative.parts + if any(part.startswith(".") or part in _EXCLUDED_DIRS for part in parts[:-1]) or parts[-1].startswith("."): + continue + posix = relative.as_posix() + if _NOT_PRODUCT.search(posix) or is_vendored(posix, facts) or is_test_path(posix, facts): + continue + if detect_language(path) not in LANGUAGES or path.stat().st_size > MAX_FILE_BYTES: + continue + files.append(path) + return files + + +def declared_functions(lines: Sequence[str], language: str) -> list[str]: + """The function names a file declares, in source order, each once.""" + pattern = DECLARATION.get(language) + if pattern is None: + return [] + names: list[str] = [] + for line in lines: + name = _declared_name(pattern, line) + if name and name not in names: + names.append(name) + return names + + +def enclosing(lines: Sequence[str], index: int, language: str) -> str: + """The nearest declaration at or above line ``index`` (0-based), `` when there is none.""" + pattern = DECLARATION.get(language) + if pattern is None: + return GLOBAL + indent = _indent(lines[index]) if language == "python" else None + for back in range(index, -1, -1): + name = _declared_name(pattern, lines[back]) + if not name: + continue + if indent is None or back == index or _indent(lines[back]) < indent: + return name + indent = min(indent, _indent(lines[back])) # a sibling or deeper def: keep looking for the one that encloses + return GLOBAL + + +def _indent(line: str) -> int: + return len(line) - len(line.lstrip(" \t")) + + +def call_sites(lines: Sequence[str], language: str, names: Iterable[str]) -> list[dict[str, Any]]: + """Calls to any of ``names`` in one file: `{"function", "line", "enclosing"}`, one per name and line, in + line order. A declaration line is not a call of the function it declares; comment lines are skipped.""" + wanted = sorted(set(names), key=lambda n: (-len(n), n)) + if not wanted: + return [] + token = re.compile(r"(? str: + match = pattern.match(line) + if not match: + return "" + name = next((g for g in match.groups() if g), "") + return "" if name in KEYWORDS else name + + +def _read_lines(path: Path) -> list[str]: + return path.read_text(encoding="utf-8", errors="replace").splitlines() + + +def _candidate_names(candidates: Sequence[Mapping[str, Any]]) -> list[str]: + return sorted({str(c["function"]) for c in candidates if c.get("function")}) + + +def _digest(names: Sequence[str]) -> str: + return hashlib.sha256("\n".join(names).encode("utf-8")).hexdigest()[:16] + + +def _load_units(units_path: Path, digest: str) -> dict[str, dict[str, Any]]: + """Rows already in `units.jsonl` that were computed for the same candidate set (Requirement 2.2).""" + rows: dict[str, dict[str, Any]] = {} + if not units_path.is_file(): + return rows + for raw in units_path.read_text(encoding="utf-8").splitlines(): + if not raw.strip(): + continue + try: + row = json.loads(raw) + except json.JSONDecodeError: + continue + if isinstance(row, dict) and row.get("digest") == digest and isinstance(row.get("path"), str): + rows[row["path"]] = row + return rows + + +def _write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def write_summary(output_dir: Path, *, status: str, units_done: int, units_total: int, **extra: Any) -> dict[str, Any]: + """The task contract's `summary.json`: `done` is refused unless every unit finished.""" + if status == "done" and units_done != units_total: + status = "failed" + extra.setdefault("reason", f"{units_done} of {units_total} units finished") + summary: dict[str, Any] = { + "status": status, + "units_done": units_done, + "units_total": units_total, + "usd": None, + "calls": 0, + "usage": {}, + "model": None, + **extra, + } + _write_json(output_dir / "summary.json", summary) + return summary + + +def assemble_facts( + rows: Mapping[str, Mapping[str, Any]], candidates: Sequence[Mapping[str, Any]], cap: int = CALLERS_CAP +) -> dict[str, Any]: + """`facts.json` from the per-file rows: functions per file, and per candidate the callers in other files.""" + files = {path: {"functions": list(row.get("functions", ()))} for path, row in sorted(rows.items())} + by_name: dict[str, list[dict[str, Any]]] = {} + for path, row in sorted(rows.items()): + for call in row.get("calls", ()): + by_name.setdefault(str(call["function"]), []).append( + {"path": path, "line": int(call["line"]), "enclosing": str(call.get("enclosing", GLOBAL))} + ) + callers: dict[str, list[dict[str, Any]]] = {} + for candidate in sorted(candidates, key=lambda c: (str(c["path"]), str(c["function"]))): + path, name = str(candidate["path"]), str(candidate["function"]) + sites = sorted( + (site for site in by_name.get(name, ()) if site["path"] != path), key=lambda s: (s["path"], s["line"], s["enclosing"]) + ) + callers[f"{path}::{name}"] = sites[:cap] + return { + "files": files, + "callers": callers, + "counts": { + "files": len(files), + "functions": sum(len(f["functions"]) for f in files.values()), + "candidates": len(callers), + "callers": sum(len(v) for v in callers.values()), + }, + } + + +def run(workspace: Path, output_dir: Path, candidates: list[dict[str, Any]] | None = None) -> dict[str, Any]: + """Compute the facts for ``workspace`` into ``output_dir``; returns the written summary. + + ``candidates`` rows are ``{"path", "function"}`` (workspace-relative POSIX path). ``None`` means every + function every product file defines. A `units.jsonl` row from an earlier run over the same candidate set + is reused, not recomputed. + """ + output_dir.mkdir(parents=True, exist_ok=True) + units_path = output_dir / "units.jsonl" + files = product_files(workspace) + languages = {p: detect_language(p) for p in files} + wanted: list[dict[str, Any]] + if candidates is None: + wanted = [ + {"path": p.relative_to(workspace).as_posix(), "function": name} + for p in files + for name in declared_functions(_read_lines(p), languages[p]) + ] + else: + wanted = [{"path": str(c["path"]), "function": str(c["function"])} for c in candidates] + names = _candidate_names(wanted) + digest = _digest(names) + present = {p.relative_to(workspace).as_posix() for p in files} + rows = {path: row for path, row in _load_units(units_path, digest).items() if path in present} + try: + with units_path.open("a", encoding="utf-8") as sink: + for path in files: + relative = path.relative_to(workspace).as_posix() + if relative in rows: + continue + lines = _read_lines(path) + row: dict[str, Any] = { + "path": relative, + "language": languages[path], + "digest": digest, + "functions": declared_functions(lines, languages[path]), + "calls": call_sites(lines, languages[path], names), + } + sink.write(json.dumps(row, sort_keys=True) + "\n") + sink.flush() + rows[relative] = row + except Exception as exc: + write_summary( + output_dir, status="failed", units_done=len(rows), units_total=len(files), reason=str(exc), traceback=traceback.format_exc() + ) + raise TaskFailed(str(exc)) from exc + facts = assemble_facts(rows, wanted) + _write_json(output_dir / "facts.json", facts) + return write_summary(output_dir, status="done", units_done=len(rows), units_total=len(files)) + + +def _load_candidates(path: str | None) -> list[dict[str, Any]] | None: + if not path: + return None + payload = json.loads(Path(path).read_text(encoding="utf-8")) + rows = payload.get("candidates", []) if isinstance(payload, dict) else payload + if not isinstance(rows, list): + raise ValueError(f"{path}: candidates must be a list of {{path, function}} rows") + return [{"path": str(row["path"]), "function": str(row["function"])} for row in rows] + + +def main(environ: Mapping[str, str] | None = None) -> int: + """Entrypoint under the task contract: `OUSAST_WORKSPACE_DIR`, `OUSAST_OUTPUT_DIR`, optional + `OUSAST_INPUT_CANDIDATES` (a JSON file). Exit 0 when done, 2 when failed; a crash still leaves a + `summary.json` with the traceback.""" + env = environ if environ is not None else os.environ + output_dir = Path(env.get("OUSAST_OUTPUT_DIR") or "") + try: + if not env.get("OUSAST_OUTPUT_DIR"): + raise ValueError("OUSAST_OUTPUT_DIR is not set") + workspace = Path(env.get("OUSAST_WORKSPACE_DIR") or "") + if not env.get("OUSAST_WORKSPACE_DIR") or not workspace.is_dir(): + raise ValueError(f"OUSAST_WORKSPACE_DIR is not a directory: {workspace}") + summary = run(workspace, output_dir, _load_candidates(env.get("OUSAST_INPUT_CANDIDATES"))) + except TaskFailed as exc: + print(f"repo-facts failed: {exc}", file=sys.stderr) + return 2 + except Exception as exc: + if env.get("OUSAST_OUTPUT_DIR"): + output_dir.mkdir(parents=True, exist_ok=True) + write_summary(output_dir, status="failed", units_done=0, units_total=0, reason=str(exc), traceback=traceback.format_exc()) + print(f"repo-facts failed: {exc}", file=sys.stderr) + return 2 + return 0 if summary["status"] == "done" else 2 + + +if __name__ == "__main__": + sys.exit(main()) + + +__all__ = [ + "CALLERS_CAP", + "DECLARATION", + "TaskFailed", + "assemble_facts", + "call_sites", + "declared_functions", + "enclosing", + "main", + "product_files", + "run", + "write_summary", +] diff --git a/tests/test_plane_repo_facts.py b/tests/test_plane_repo_facts.py new file mode 100644 index 0000000..981dbc4 --- /dev/null +++ b/tests/test_plane_repo_facts.py @@ -0,0 +1,274 @@ +"""`repo-facts` (ai-service-plane task 5): cross-file callers, product filter, determinism, per-unit resume.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.plane.tasks import repo_facts + +HANDLERS = """\ +from .db import run_query +from .util import render + +def handler(request): + rows = run_query(request.args["q"]) + return render(rows) + +def unrelated(): + return 1 + +render(None) +""" +DB = """\ +import logging + +def run_query(sql): + # run_query(sql) is documented here, not called + log("query") + return sql + +def helper(): + return log("x") +""" +UTIL = """\ +def log(message): + print(message) + +def render(rows): + return str(rows) + +class Probe: + def check(self): + return run_query("select 1") +""" +TEST_FILE = "from app.db import run_query\n\ndef test_run_query():\n assert run_query('x')\n" +VENDOR_FILE = "def run_query(x):\n return x\n\nrun_query(1)\n" + + +@pytest.fixture +def tree(tmp_path: Path) -> Path: + workspace = tmp_path / "ws" + (workspace / "app").mkdir(parents=True) + (workspace / "app" / "handlers.py").write_text(HANDLERS) + (workspace / "app" / "db.py").write_text(DB) + (workspace / "app" / "util.py").write_text(UTIL) + (workspace / "tests").mkdir() + (workspace / "tests" / "test_db.py").write_text(TEST_FILE) + (workspace / "vendor" / "lib").mkdir(parents=True) + (workspace / "vendor" / "lib" / "shim.py").write_text(VENDOR_FILE) + (workspace / "node_modules" / "pkg").mkdir(parents=True) + (workspace / "node_modules" / "pkg" / "index.js").write_text("function run_query() {}\nrun_query();\n") + (workspace / "app" / "bundle.min.js").write_text("function run_query(){}run_query();") + (workspace / ".hidden").mkdir() + (workspace / ".hidden" / "x.py").write_text("run_query(2)\n") + return workspace + + +def _run(workspace: Path, out: Path, candidates: list[dict[str, str]] | None = None) -> dict[str, object]: + repo_facts.run(workspace, out, candidates) + facts = json.loads((out / "facts.json").read_text()) + assert isinstance(facts, dict) + return facts + + +def test_product_files_exclude_tests_vendor_bundles_and_dotdirs(tree: Path) -> None: + files = [p.relative_to(tree).as_posix() for p in repo_facts.product_files(tree)] + assert files == ["app/db.py", "app/handlers.py", "app/util.py"] + + +def test_cross_file_callers_have_line_and_enclosing_function(tree: Path, tmp_path: Path) -> None: + out = tmp_path / "out" + facts = _run(tree, out, [{"path": "app/db.py", "function": "run_query"}, {"path": "app/util.py", "function": "render"}]) + assert facts["files"] == { + "app/db.py": {"functions": ["run_query", "helper"]}, + "app/handlers.py": {"functions": ["handler", "unrelated"]}, + "app/util.py": {"functions": ["log", "render", "check"]}, + } + assert facts["callers"]["app/db.py::run_query"] == [ + {"path": "app/handlers.py", "line": 5, "enclosing": "handler"}, + {"path": "app/util.py", "line": 9, "enclosing": "check"}, + ] + assert facts["callers"]["app/util.py::render"] == [ + {"path": "app/handlers.py", "line": 6, "enclosing": "handler"}, + {"path": "app/handlers.py", "line": 11, "enclosing": ""}, + ] + assert facts["counts"] == {"files": 3, "functions": 7, "candidates": 2, "callers": 4} + summary = json.loads((out / "summary.json").read_text()) + assert summary == {"status": "done", "units_done": 3, "units_total": 3, "usd": None, "calls": 0, "usage": {}, "model": None} + + +def test_same_file_calls_and_declaration_lines_are_not_callers(tree: Path, tmp_path: Path) -> None: + facts = _run(tree, tmp_path / "out", [{"path": "app/util.py", "function": "log"}]) + # db.py calls log twice (a call and a comment mentioning it); util.py defines it, so its own lines never count. + assert facts["callers"]["app/util.py::log"] == [ + {"path": "app/db.py", "line": 5, "enclosing": "run_query"}, + {"path": "app/db.py", "line": 9, "enclosing": "helper"}, + ] + + +def test_every_function_is_a_candidate_when_none_are_given(tree: Path, tmp_path: Path) -> None: + facts = _run(tree, tmp_path / "out") + assert sorted(facts["callers"]) == [ + "app/db.py::helper", + "app/db.py::run_query", + "app/handlers.py::handler", + "app/handlers.py::unrelated", + "app/util.py::check", + "app/util.py::log", + "app/util.py::render", + ] + assert facts["callers"]["app/handlers.py::handler"] == [] + + +def test_output_is_byte_identical_on_a_second_run(tree: Path, tmp_path: Path) -> None: + candidates = [{"path": "app/db.py", "function": "run_query"}] + _run(tree, tmp_path / "a", candidates) + _run(tree, tmp_path / "b", candidates) + for name in ("facts.json", "units.jsonl", "summary.json"): + assert (tmp_path / "a" / name).read_bytes() == (tmp_path / "b" / name).read_bytes(), name + + +def test_callers_are_capped_at_forty_per_function(tmp_path: Path) -> None: + workspace = tmp_path / "ws" + workspace.mkdir() + (workspace / "lib.py").write_text("def target():\n pass\n") + (workspace / "many.py").write_text("".join(f"target({i})\n" for i in range(60))) + facts = _run(workspace, tmp_path / "out", [{"path": "lib.py", "function": "target"}]) + sites = facts["callers"]["lib.py::target"] + assert len(sites) == repo_facts.CALLERS_CAP == 40 + assert [s["line"] for s in sites] == list(range(1, 41)) + + +def test_resume_skips_files_already_in_units(tree: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + out = tmp_path / "out" + candidates = [{"path": "app/db.py", "function": "run_query"}] + _run(tree, out, candidates) + first = (out / "facts.json").read_bytes() + rows_before = (out / "units.jsonl").read_text().splitlines() + (out / "facts.json").unlink() + (out / "summary.json").unlink() + calls: list[str] = [] + real = repo_facts.call_sites + + def counting(lines: list[str], language: str, names: list[str]) -> list[dict[str, object]]: + calls.append(language) + return real(lines, language, names) + + monkeypatch.setattr(repo_facts, "call_sites", counting) + summary = repo_facts.run(tree, out, candidates) + assert calls == [], "a file already in units.jsonl was recomputed" + assert summary["status"] == "done" + assert (out / "facts.json").read_bytes() == first + assert (out / "units.jsonl").read_text().splitlines() == rows_before + + +def test_resume_recomputes_only_the_missing_unit(tree: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + out = tmp_path / "out" + candidates = [{"path": "app/db.py", "function": "run_query"}] + _run(tree, out, candidates) + rows = (out / "units.jsonl").read_text().splitlines() + kept = [r for r in rows if json.loads(r)["path"] != "app/util.py"] + (out / "units.jsonl").write_text("\n".join(kept) + "\n") + processed: list[str] = [] + real = repo_facts.call_sites + + def counting(lines: list[str], language: str, names: list[str]) -> list[dict[str, object]]: + processed.append(lines[0]) + return real(lines, language, names) + + monkeypatch.setattr(repo_facts, "call_sites", counting) + _run(tree, out, candidates) + assert processed == ["def log(message):"] + facts = json.loads((out / "facts.json").read_text()) + assert [s["path"] for s in facts["callers"]["app/db.py::run_query"]] == ["app/handlers.py", "app/util.py"] + + +def test_rows_from_another_candidate_set_are_not_reused(tree: Path, tmp_path: Path) -> None: + out = tmp_path / "out" + _run(tree, out, [{"path": "app/db.py", "function": "run_query"}]) + facts = _run(tree, out, [{"path": "app/util.py", "function": "render"}]) + assert [s["line"] for s in facts["callers"]["app/util.py::render"]] == [6, 11] + summary = json.loads((out / "summary.json").read_text()) + assert summary["status"] == "done" and summary["units_done"] == 3 + + +def test_summary_is_never_done_when_a_unit_is_missing(tree: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + out = tmp_path / "out" + real = repo_facts.call_sites + + def crash_on_util(lines: list[str], language: str, names: list[str]) -> list[dict[str, object]]: + if lines[0].startswith("def log"): + raise OSError("disk went away") + return real(lines, language, names) + + monkeypatch.setattr(repo_facts, "call_sites", crash_on_util) + env = {"OUSAST_WORKSPACE_DIR": str(tree), "OUSAST_OUTPUT_DIR": str(out)} + assert repo_facts.main(env) == 2 + summary = json.loads((out / "summary.json").read_text()) + assert summary["status"] == "failed" + assert (summary["units_done"], summary["units_total"]) == (2, 3) + assert "disk went away" in summary["traceback"] + assert not (out / "facts.json").exists() + # the contract itself refuses `done` with a unit missing, whatever the caller claims + (tmp_path / "forced").mkdir() + forced = repo_facts.write_summary(tmp_path / "forced", status="done", units_done=2, units_total=3) + assert forced["status"] == "failed" + # and a restart finishes only the missing unit + monkeypatch.setattr(repo_facts, "call_sites", real) + assert repo_facts.main(env) == 0 + summary = json.loads((out / "summary.json").read_text()) + assert summary["status"] == "done" and summary["units_done"] == 3 + assert len((out / "units.jsonl").read_text().splitlines()) == 3 + + +def test_main_reads_candidates_from_the_input_env(tree: Path, tmp_path: Path) -> None: + out = tmp_path / "out" + candidates = tmp_path / "candidates.json" + candidates.write_text(json.dumps({"candidates": [{"path": "app/db.py", "function": "run_query"}]})) + env = {"OUSAST_WORKSPACE_DIR": str(tree), "OUSAST_OUTPUT_DIR": str(out), "OUSAST_INPUT_CANDIDATES": str(candidates)} + assert repo_facts.main(env) == 0 + facts = json.loads((out / "facts.json").read_text()) + assert list(facts["callers"]) == ["app/db.py::run_query"] + + +def test_main_without_a_workspace_fails_with_a_summary(tmp_path: Path) -> None: + out = tmp_path / "out" + assert repo_facts.main({"OUSAST_WORKSPACE_DIR": str(tmp_path / "missing"), "OUSAST_OUTPUT_DIR": str(out)}) == 2 + summary = json.loads((out / "summary.json").read_text()) + assert summary["status"] == "failed" and "OUSAST_WORKSPACE_DIR" in summary["reason"] + + +@pytest.mark.parametrize( + ("language", "source", "expected"), + [ + ("php", " {\n};\n" + "class K {\n save(data) {\n }\n}\nif (x) {\n}\n", + ["fetchAll", "handler", "save"], + ), + ("typescript", "export const route = (req: Req): void => {\n};\nfunction* gen() {}\n", ["route", "gen"]), + ( + "java", + "public class A {\n public static List names(int n) throws IOException {\n }\n" + " private void go() {\n return foo(1);\n }\n else if (x) {\n }\n}\n", + ["names", "go"], + ), + ("go", "func (r *Repo) Get(id int) error {\n}\nfunc main() {\n}\n", ["Get", "main"]), + ("ruby", "class A\n def self.build(x)\n end\n def valid?\n end\nend\n", ["build", "valid?"]), + ], +) +def test_declaration_patterns_per_language(language: str, source: str, expected: list[str]) -> None: + assert repo_facts.declared_functions(source.splitlines(), language) == expected + + +def test_enclosing_function_in_python_respects_indentation() -> None: + lines = ["def a():", " x()", "", "def b():", " def inner():", " y()", " z()", "", "w()"] + assert repo_facts.enclosing(lines, 1, "python") == "a" + assert repo_facts.enclosing(lines, 5, "python") == "inner" + assert repo_facts.enclosing(lines, 6, "python") == "b" + assert repo_facts.enclosing(lines, 8, "python") == ""