382 lines
14 KiB
Python
Executable file
382 lines
14 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Run source metamorphic relations and compare canonical finding multisets."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fold_eval.metamorphic import compare_metamorphic_findings
|
|
from fold_eval.sarif import canonical_findings
|
|
from fold_eval.schema import (
|
|
artifact_record,
|
|
build_run_artifact,
|
|
executable_version,
|
|
sha256_file,
|
|
utc_now,
|
|
validate_benchmark_case,
|
|
validate_run_artifact,
|
|
write_json,
|
|
)
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
RUNNER_COMMON_PATH = REPO_ROOT / "scripts" / "benchmarks" / "runner_common.py"
|
|
DEFAULT_SPEC = (
|
|
REPO_ROOT / "tests" / "benchmarks" / "metamorphic" / "division_metamorphic.json"
|
|
)
|
|
_spec = importlib.util.spec_from_file_location("runner_common", RUNNER_COMMON_PATH)
|
|
if _spec is None or _spec.loader is None:
|
|
raise ImportError(f"failed to load {RUNNER_COMMON_PATH}")
|
|
common = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(common)
|
|
|
|
|
|
def _expected_integer(expect: Mapping[str, Any], name: str, default: int) -> int:
|
|
value = expect.get(name, default)
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise ValueError(f"expect.{name} must be an integer")
|
|
return value
|
|
|
|
|
|
def _clone_family_failures(findings: list[dict[str, Any]]) -> list[str]:
|
|
failures: list[str] = []
|
|
claimed_members: set[str] = set()
|
|
classifications = {
|
|
"configuration",
|
|
"mapping",
|
|
"generated",
|
|
"macro",
|
|
"template",
|
|
"product-logic",
|
|
"unclassified",
|
|
}
|
|
for finding in findings:
|
|
if finding.get("rule_id") != "fold.clone":
|
|
continue
|
|
identity = str(finding.get("finding_id", "<missing>"))
|
|
members = finding.get("quality_member_ids", [])
|
|
if not isinstance(members, list) or len(members) < 2:
|
|
failures.append(f"clone {identity} lacks a two-member family")
|
|
continue
|
|
if members != sorted(set(members)):
|
|
failures.append(f"clone {identity} members are not sorted and unique")
|
|
representative = finding.get("quality_representative_id")
|
|
if representative != members[0]:
|
|
failures.append(
|
|
f"clone {identity} representative is not the first stable member"
|
|
)
|
|
metrics = finding.get("quality_metrics", {})
|
|
family_size = metrics.get("family_size") if isinstance(metrics, dict) else None
|
|
if family_size != len(members):
|
|
failures.append(f"clone {identity} family_size does not match its members")
|
|
classification = finding.get("quality_classification")
|
|
if classification not in classifications:
|
|
failures.append(
|
|
f"clone {identity} has unknown classification {classification!r}"
|
|
)
|
|
overlap = claimed_members.intersection(members)
|
|
if overlap:
|
|
failures.append(
|
|
f"clone {identity} overlaps prior families at {sorted(overlap)}"
|
|
)
|
|
claimed_members.update(str(member) for member in members)
|
|
if finding.get("quality_incomplete") and not finding.get(
|
|
"quality_uncertainty_reasons"
|
|
):
|
|
failures.append(
|
|
f"clone {identity} is incomplete without an uncertainty reason"
|
|
)
|
|
return failures
|
|
|
|
|
|
def _sources(variant: Mapping[str, Any], spec_path: Path) -> list[Path]:
|
|
raw = variant.get("sources", [])
|
|
if not isinstance(raw, list) or not raw:
|
|
raise ValueError("metamorphic variants require a non-empty sources array")
|
|
sources = [common.resolve_path(str(item), spec_path.parent) for item in raw]
|
|
missing = [str(path) for path in sources if not path.is_file()]
|
|
if missing:
|
|
raise FileNotFoundError(", ".join(missing))
|
|
return sources
|
|
|
|
|
|
def _write_compile_db(sources: list[Path], output_dir: Path, standard: str) -> Path:
|
|
rows = [
|
|
{
|
|
"directory": str(REPO_ROOT),
|
|
"arguments": ["clang++", f"-std={standard}", "-c", str(source)],
|
|
"file": str(source),
|
|
}
|
|
for source in sources
|
|
]
|
|
path = output_dir / "compile_commands.json"
|
|
path.write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def run_variant(
|
|
*,
|
|
fold: Path,
|
|
spec: Mapping[str, Any],
|
|
spec_path: Path,
|
|
variant: Mapping[str, Any],
|
|
variant_name: str,
|
|
output_root: Path,
|
|
) -> tuple[dict[str, Any], list[str]]:
|
|
output_dir = output_root / variant_name
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
sources = _sources(variant, spec_path)
|
|
database = _write_compile_db(
|
|
sources, output_dir, str(spec.get("standard", "c++20"))
|
|
)
|
|
report_path = output_dir / "report.sarif.json"
|
|
log_path = output_dir / "scan.log"
|
|
artifact_path = output_dir / "run.json"
|
|
command = [
|
|
str(fold),
|
|
"--quiet",
|
|
"scan",
|
|
"--mode",
|
|
"source",
|
|
"--compile-db",
|
|
str(database),
|
|
"--detectors",
|
|
str(spec.get("detectors", "security")),
|
|
"--format",
|
|
"json",
|
|
"--output",
|
|
str(report_path),
|
|
"--processes",
|
|
str(variant.get("processes", 1)),
|
|
]
|
|
if "min_body_stmts" in spec:
|
|
command.extend(["--min-body-stmts", str(spec["min_body_stmts"])])
|
|
started_at = utc_now()
|
|
proc = common.run_measured(command, REPO_ROOT, timeout_seconds=60.0)
|
|
ended_at = utc_now()
|
|
log_path.write_text(proc.output, encoding="utf-8")
|
|
failures: list[str] = []
|
|
findings: list[dict[str, Any]] = []
|
|
missing_location = 0
|
|
missing_backend = 0
|
|
if proc.returncode not in (0, 1):
|
|
failures.append(f"scan exited {proc.returncode}")
|
|
elif not report_path.is_file():
|
|
failures.append("scan 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, REPO_ROOT
|
|
)
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
failures.append(f"invalid report: {exc}")
|
|
if missing_location:
|
|
failures.append(f"report has {missing_location} locationless findings")
|
|
if missing_backend:
|
|
failures.append(f"report has {missing_backend} findings without backend")
|
|
expect = spec.get("expect", {})
|
|
if isinstance(expect, dict):
|
|
summary_count = sum(bool(item.get("summary_backed")) for item in findings)
|
|
cross_tu_count = sum(
|
|
bool(item.get("cross_tu_corroboration")) for item in findings
|
|
)
|
|
if summary_count > _expected_integer(
|
|
expect, "max_summary_backed", summary_count
|
|
):
|
|
failures.append(
|
|
f"summary-backed findings {summary_count} exceed expected maximum"
|
|
)
|
|
if cross_tu_count > _expected_integer(
|
|
expect, "max_cross_tu_corroboration", cross_tu_count
|
|
):
|
|
failures.append(
|
|
f"cross-TU-corroborated findings {cross_tu_count} exceed expected maximum"
|
|
)
|
|
minimum = _expected_integer(expect, "min_findings", 0)
|
|
if len(findings) < minimum:
|
|
failures.append(
|
|
f"findings {len(findings)} below expected minimum {minimum}"
|
|
)
|
|
rule_counts: dict[str, int] = {}
|
|
for finding in findings:
|
|
rule = str(finding.get("rule_id", ""))
|
|
rule_counts[rule] = rule_counts.get(rule, 0) + 1
|
|
for expectation_name, comparison in (
|
|
("min_rule_findings", lambda actual, expected: actual < expected),
|
|
("max_rule_findings", lambda actual, expected: actual > expected),
|
|
):
|
|
expected_counts = expect.get(expectation_name, {})
|
|
if not isinstance(expected_counts, dict):
|
|
raise ValueError(f"expect.{expectation_name} must be an object")
|
|
for rule, expected_count in expected_counts.items():
|
|
if isinstance(expected_count, bool) or not isinstance(
|
|
expected_count, int
|
|
):
|
|
raise ValueError(
|
|
f"expect.{expectation_name}.{rule} must be an integer"
|
|
)
|
|
actual = rule_counts.get(str(rule), 0)
|
|
if comparison(actual, expected_count):
|
|
failures.append(
|
|
f"{rule} findings {actual} violate {expectation_name} {expected_count}"
|
|
)
|
|
if spec.get("validate_clone_families", False):
|
|
failures.extend(_clone_family_failures(findings))
|
|
if spec.get("require_unique_finding_ids", False):
|
|
identities = [str(item.get("finding_id", "")) for item in findings]
|
|
missing_identities = sum(not identity for identity in identities)
|
|
duplicate_identities = len(identities) - len(set(identities))
|
|
if missing_identities:
|
|
failures.append(
|
|
f"report has {missing_identities} findings without identities"
|
|
)
|
|
if duplicate_identities:
|
|
failures.append(
|
|
f"report has {duplicate_identities} duplicate finding identities"
|
|
)
|
|
|
|
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"metamorphic:{spec['case_id']}:{variant_name}",
|
|
case_id=str(spec["case_id"]),
|
|
tool={
|
|
"name": "fold",
|
|
"path": str(fold),
|
|
"version": executable_version(fold),
|
|
"sha256": sha256_file(fold),
|
|
},
|
|
target={
|
|
"sources": [str(source) for source in sources],
|
|
"compile_db": str(database),
|
|
"input_fingerprint": sha256_file(database),
|
|
},
|
|
command=command,
|
|
started_at=started_at,
|
|
ended_at=ended_at,
|
|
exit_status=proc.returncode,
|
|
budgets={"wall_seconds": 60.0, "processes": variant.get("processes", 1)},
|
|
artifacts=artifacts,
|
|
measurements=proc.metrics(),
|
|
observations={
|
|
"findings": findings,
|
|
"missing_location": missing_location,
|
|
"missing_backend": missing_backend,
|
|
},
|
|
)
|
|
failures.extend(
|
|
f"run artifact: {error}" for error in validate_run_artifact(run_artifact)
|
|
)
|
|
write_json(artifact_path, run_artifact)
|
|
return {
|
|
"name": variant_name,
|
|
"findings": findings,
|
|
"finding_count": len(findings),
|
|
"report": str(report_path),
|
|
"run_artifact": str(artifact_path),
|
|
"measurements": proc.metrics(),
|
|
}, failures
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--build", type=Path, default=Path("build/release"))
|
|
parser.add_argument("--fold", type=Path)
|
|
parser.add_argument("--spec", type=Path, default=DEFAULT_SPEC)
|
|
parser.add_argument(
|
|
"--output-dir", type=Path, default=Path("/tmp/fold-source-metamorphic")
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
build = args.build if args.build.is_absolute() else REPO_ROOT / args.build
|
|
fold = common.find_fold_binary(REPO_ROOT, build, args.fold)
|
|
if fold is None:
|
|
common.print_missing_fold(args.fold or build / "src" / "cli" / "fold")
|
|
return 2
|
|
spec_path = args.spec.resolve()
|
|
try:
|
|
spec = common.load_json(spec_path)
|
|
validation = validate_benchmark_case(
|
|
spec, allow_legacy=False, source_path=str(spec_path)
|
|
)
|
|
if not validation["valid"]:
|
|
raise ValueError("; ".join(validation["errors"]))
|
|
relation = spec.get("relation", {})
|
|
comparisons = relation.get("comparisons", [])
|
|
if not isinstance(comparisons, list) or not comparisons:
|
|
raise ValueError("relation.comparisons must be a non-empty array")
|
|
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(json.dumps({"status": "failed", "reason": str(exc)}, indent=2))
|
|
return 2
|
|
|
|
args.output_dir.mkdir(parents=True, exist_ok=True)
|
|
results: list[dict[str, Any]] = []
|
|
failures: list[str] = []
|
|
for index, comparison in enumerate(comparisons, start=1):
|
|
if not isinstance(comparison, dict):
|
|
failures.append(f"comparison {index} is not an object")
|
|
continue
|
|
name = str(comparison.get("name", f"comparison-{index}"))
|
|
try:
|
|
baseline, baseline_failures = run_variant(
|
|
fold=fold,
|
|
spec=spec,
|
|
spec_path=spec_path,
|
|
variant=comparison.get("baseline", {}),
|
|
variant_name=f"{index}-{name}-baseline",
|
|
output_root=args.output_dir,
|
|
)
|
|
transformed, transformed_failures = run_variant(
|
|
fold=fold,
|
|
spec=spec,
|
|
spec_path=spec_path,
|
|
variant=comparison.get("transformed", {}),
|
|
variant_name=f"{index}-{name}-transformed",
|
|
output_root=args.output_dir,
|
|
)
|
|
comparison_result = compare_metamorphic_findings(
|
|
baseline["findings"],
|
|
transformed["findings"],
|
|
comparison.get("mapping", {}),
|
|
)
|
|
row_failures = baseline_failures + transformed_failures
|
|
if not comparison_result["exact"]:
|
|
row_failures.append("canonical finding multiset drifted")
|
|
failures.extend(f"{name}: {failure}" for failure in row_failures)
|
|
results.append(
|
|
{
|
|
"name": name,
|
|
"status": "failed" if row_failures else "ok",
|
|
"comparison": comparison_result,
|
|
"baseline": baseline,
|
|
"transformed": transformed,
|
|
"failures": row_failures,
|
|
}
|
|
)
|
|
except (OSError, ValueError, TypeError) as exc:
|
|
failures.append(f"{name}: {exc}")
|
|
results.append({"name": name, "status": "failed", "reason": str(exc)})
|
|
|
|
summary = {
|
|
"schema_version": 1,
|
|
"suite": "source_metamorphic",
|
|
"case_id": spec["case_id"],
|
|
"status": "failed" if failures else "ok",
|
|
"results": results,
|
|
"failures": failures,
|
|
}
|
|
write_json(args.output_dir / "summary.json", summary)
|
|
print(json.dumps(summary, indent=2, sort_keys=True))
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|