139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
"""Shared Fold source-analysis lane for differential and corpus benchmarks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import runner_common as common
|
|
|
|
from .sarif import canonical_findings
|
|
from .schema import (
|
|
artifact_record,
|
|
build_run_artifact,
|
|
executable_version,
|
|
sha256_file,
|
|
utc_now,
|
|
validate_run_artifact,
|
|
write_json,
|
|
)
|
|
|
|
|
|
def write_compile_db(source: Path, output_dir: Path, repo_root: Path) -> Path:
|
|
database = output_dir / "compile_commands.json"
|
|
database.write_text(
|
|
json.dumps(
|
|
[
|
|
{
|
|
"directory": str(repo_root),
|
|
"arguments": ["clang++", "-std=c++20", "-c", str(source)],
|
|
"file": str(source),
|
|
}
|
|
],
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return database
|
|
|
|
|
|
def run_fold_source(
|
|
*,
|
|
fold: Path,
|
|
source: Path,
|
|
detector: str,
|
|
case_id: str,
|
|
label: str,
|
|
output_dir: Path,
|
|
repo_root: Path,
|
|
artifact_base: Path | None = None,
|
|
timeout_seconds: float = 60.0,
|
|
) -> tuple[dict[str, Any], list[str]]:
|
|
"""Run one detector and retain its command, SARIF, log, and run contract."""
|
|
|
|
lane_dir = output_dir / label / "fold"
|
|
lane_dir.mkdir(parents=True, exist_ok=True)
|
|
database = write_compile_db(source, lane_dir, repo_root)
|
|
report_path = lane_dir / "report.sarif.json"
|
|
log_path = lane_dir / "scan.log"
|
|
run_path = lane_dir / "run.json"
|
|
command = [
|
|
str(fold),
|
|
"--quiet",
|
|
"scan",
|
|
"--mode",
|
|
"source",
|
|
"--compile-db",
|
|
str(database),
|
|
"--detectors",
|
|
detector,
|
|
"--format",
|
|
"json",
|
|
"--output",
|
|
str(report_path),
|
|
"--processes",
|
|
"1",
|
|
]
|
|
started_at = utc_now()
|
|
proc = common.run_measured(command, repo_root, timeout_seconds=timeout_seconds)
|
|
ended_at = utc_now()
|
|
log_path.write_text(proc.output, encoding="utf-8")
|
|
failures: list[str] = []
|
|
findings: list[dict[str, Any]] = []
|
|
if proc.returncode not in (0, 1):
|
|
failures.append(f"Fold exited {proc.returncode}")
|
|
elif not report_path.is_file():
|
|
failures.append("Fold did not write a report")
|
|
else:
|
|
try:
|
|
report = json.loads(report_path.read_text(encoding="utf-8"))
|
|
findings, missing_location, missing_backend = canonical_findings(
|
|
report, repo_root, artifact_base or repo_root
|
|
)
|
|
if missing_location:
|
|
failures.append(
|
|
f"Fold emitted {missing_location} locationless findings"
|
|
)
|
|
if missing_backend:
|
|
failures.append(
|
|
f"Fold emitted {missing_backend} findings without backend"
|
|
)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
failures.append(f"invalid Fold report: {exc}")
|
|
|
|
artifacts = [artifact_record("log", log_path, "text/plain")]
|
|
if report_path.is_file():
|
|
artifacts.append(
|
|
artifact_record("sarif-report", report_path, "application/sarif+json")
|
|
)
|
|
run_artifact = build_run_artifact(
|
|
experiment_id=f"source-lane:{case_id}:{label}:fold",
|
|
case_id=case_id,
|
|
tool={
|
|
"name": "fold",
|
|
"path": str(fold),
|
|
"version": executable_version(fold),
|
|
"sha256": sha256_file(fold),
|
|
},
|
|
target={"source": str(source), "input_fingerprint": sha256_file(source)},
|
|
command=command,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
exit_status=proc.returncode,
|
|
budgets={"wall_seconds": timeout_seconds, "processes": 1},
|
|
artifacts=artifacts,
|
|
measurements=proc.metrics(),
|
|
observations={"findings": findings},
|
|
)
|
|
failures.extend(
|
|
f"run artifact: {error}" for error in validate_run_artifact(run_artifact)
|
|
)
|
|
write_json(run_path, run_artifact)
|
|
return {
|
|
"findings": findings,
|
|
"report": str(report_path),
|
|
"run_artifact": str(run_path),
|
|
"measurements": proc.metrics(),
|
|
}, failures
|