mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
merge(plane): repo-facts memory task (task 5)
Audit registry conflict resolved by keeping every plane row; manifest regenerated from the merged tree. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"modules": 121,
|
||||
"modules": 123,
|
||||
"counts": {
|
||||
"load_bearing": 112,
|
||||
"load_bearing": 114,
|
||||
"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"
|
||||
]
|
||||
@@ -667,6 +669,18 @@
|
||||
"reason": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
|
||||
"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",
|
||||
@@ -713,6 +727,7 @@
|
||||
"model.partitions",
|
||||
"model.scan",
|
||||
"pairs",
|
||||
"plane.tasks.repo_facts",
|
||||
"push.runner",
|
||||
"rank",
|
||||
"regress.candidate",
|
||||
@@ -1343,6 +1358,5 @@
|
||||
"cli"
|
||||
]
|
||||
}
|
||||
],
|
||||
"note": "Group 8: runner reaches recomputed versioned eligibility and its frozen scorer."
|
||||
]
|
||||
}
|
||||
|
||||
@@ -57,6 +57,8 @@ SPEC_OWNED = {
|
||||
"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.manifests": "ai-service-plane Req 1.1-1.4: ax-shaped Task/Workspace/Model and the Run kind, validated before anything runs",
|
||||
"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 = {
|
||||
|
||||
@@ -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 = "<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), `<global>` 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"(?<![A-Za-z0-9_$])(" + "|".join(re.escape(n) for n in wanted) + r")\s*\(")
|
||||
pattern = DECLARATION.get(language)
|
||||
found: list[dict[str, Any]] = []
|
||||
for index, line in enumerate(lines):
|
||||
if line.lstrip().startswith(_COMMENT_PREFIXES):
|
||||
continue
|
||||
declared = _declared_name(pattern, line) if pattern is not None else ""
|
||||
seen: set[str] = set()
|
||||
for match in token.finditer(line):
|
||||
name = match.group(1)
|
||||
if name == declared or name in seen:
|
||||
continue
|
||||
seen.add(name)
|
||||
found.append({"function": name, "line": index + 1, "enclosing": enclosing(lines, index, language)})
|
||||
return found
|
||||
|
||||
|
||||
def _declared_name(pattern: re.Pattern[str], line: str) -> 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",
|
||||
]
|
||||
@@ -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": "<global>"},
|
||||
]
|
||||
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", "<?php\nclass A {\n public static function &load($id) {\n }\n}\nfunction top() {}\n", ["load", "top"]),
|
||||
(
|
||||
"javascript",
|
||||
"export async function fetchAll() {}\nconst handler = async (req, res) => {\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<String> 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") == "<global>"
|
||||
Reference in New Issue
Block a user