fold/scripts/benchmarks/compare_findings.py

#!/usr/bin/env python3
"""Compare Fold JSON findings against a golden benchmark spec.

The comparator is intentionally small and dependency-free so it can run in local
pre-commit flows and CI. It checks report-shape invariants first (location and
backend provenance), then applies benchmark-specific must-report/must-not-report
rules.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from collections import Counter
from pathlib import Path
from typing import Any


def load_json(path: Path) -> Any:
    try:
        with path.open("r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"invalid JSON file {path}: {exc}") from exc


def nonnegative_int(value: Any) -> int | None:
    return (
        value
        if isinstance(value, int) and not isinstance(value, bool) and value >= 0
        else None
    )


def findings(report: dict[str, Any]) -> list[dict[str, Any]]:
    runs = report.get("runs")
    if not isinstance(runs, list) or not runs:
        return []
    flattened: list[dict[str, Any]] = []
    for run in runs:
        if not isinstance(run, dict):
            continue
        results = run.get("results", [])
        if not isinstance(results, list):
            continue
        flattened.extend(result for result in results if isinstance(result, dict))
    return flattened


def message(finding: dict[str, Any]) -> str:
    msg = finding.get("message", {})
    if isinstance(msg, dict):
        return str(msg.get("text", ""))
    return str(msg)


def physical_location(finding: dict[str, Any]) -> dict[str, Any] | None:
    locs = finding.get("locations", [])
    if not isinstance(locs, list) or not locs:
        return None
    phys = locs[0].get("physicalLocation", {})
    return phys if isinstance(phys, dict) else None


def has_physical_location(finding: dict[str, Any]) -> bool:
    phys = physical_location(finding)
    if not phys:
        return False
    artifact = phys.get("artifactLocation", {})
    uri = artifact.get("uri") if isinstance(artifact, dict) else None
    region = phys.get("region", {})
    line = region.get("startLine") if isinstance(region, dict) else None
    return bool(uri) and isinstance(line, int) and line > 0


def backend_of(finding: dict[str, Any]) -> str:
    props = finding.get("properties", {})
    if isinstance(props, dict):
        return str(props.get("backend", ""))
    return ""


def artifact_uri(finding: dict[str, Any]) -> str:
    phys = physical_location(finding)
    if not phys:
        return ""
    artifact = phys.get("artifactLocation", {})
    return str(artifact.get("uri", "")) if isinstance(artifact, dict) else ""


def logical_function(finding: dict[str, Any]) -> str:
    locations = finding.get("locations", [])
    if not isinstance(locations, list) or not locations:
        return ""
    logical = locations[0].get("logicalLocations", [])
    if not isinstance(logical, list) or not logical:
        return ""
    first = logical[0]
    return str(first.get("fullyQualifiedName", "")) if isinstance(first, dict) else ""


def coverage_labels(rule: dict[str, Any]) -> set[str]:
    raw = rule.get("coverage", [])
    labels = [raw] if isinstance(raw, str) else raw
    return (
        {label for label in labels if isinstance(label, str)}
        if isinstance(labels, list)
        else set()
    )


def fixture_coverage_failures(expect: dict[str, Any]) -> list[str]:
    required = expect.get("coverage_requirements", {})
    if not isinstance(required, dict):
        return ["coverage_requirements must be an object"] if required else []

    failures: list[str] = []
    positive = expect.get("must_report", [])
    negative = expect.get("must_not_report", [])
    if not isinstance(positive, list) or not isinstance(negative, list):
        return ["coverage_requirements requires must_report and must_not_report lists"]
    for rule_id, labels in sorted(required.items()):
        if not isinstance(rule_id, str) or not isinstance(labels, list):
            failures.append("coverage_requirements must map rule IDs to label lists")
            continue
        required_labels = {label for label in labels if isinstance(label, str)}
        if required_labels != set(labels) or not required_labels:
            failures.append(
                f"coverage requirements for {rule_id} must be non-empty strings"
            )
            continue
        for label in required_labels:
            entries = (
                positive
                if label in {"positive", "regression"}
                else negative
                if label == "negative"
                else []
            )
            if not entries:
                failures.append(f"coverage {rule_id} has unsupported label: {label}")
                continue
            if not any(
                isinstance(entry, dict)
                and entry.get("ruleId") == rule_id
                and label in coverage_labels(entry)
                for entry in entries
            ):
                failures.append(f"coverage {rule_id} missing {label} fixture assertion")
    return failures


def matches_rule(finding: dict[str, Any], rule: dict[str, Any]) -> bool:
    rule_id = rule.get("ruleId")
    if rule_id is not None and finding.get("ruleId") != rule_id:
        return False
    backend = rule.get("backend")
    if backend is not None and backend_of(finding) != backend:
        return False
    fields = (
        ("messageRegex", message(finding)),
        ("uriRegex", artifact_uri(finding)),
        ("functionRegex", logical_function(finding)),
    )
    return all(
        rule.get(key) is None
        or re.search(str(rule[key]), value, re.MULTILINE) is not None
        for key, value in fields
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--report", required=True, type=Path, help="Fold JSON report")
    parser.add_argument(
        "--spec", required=True, type=Path, help="Golden comparison spec"
    )
    args = parser.parse_args()

    report = load_json(args.report)
    spec = load_json(args.spec)
    fs = findings(report)
    expect = spec.get("expect", {}) if isinstance(spec.get("expect", {}), dict) else {}

    failures: list[str] = fixture_coverage_failures(expect)
    warnings: list[str] = []

    if expect.get("require_physical_location", True):
        bad = [f for f in fs if not has_physical_location(f)]
        if bad:
            failures.append(f"{len(bad)} findings lack physical locations")

    if expect.get("require_backend", True):
        bad = [f for f in fs if not backend_of(f)]
        if bad:
            failures.append(f"{len(bad)} findings lack backend provenance")

    present_backends = {backend_of(f) for f in fs if backend_of(f)}
    for required in expect.get("require_backends", []):
        if str(required) not in present_backends:
            failures.append(f"required backend missing: {required}")

    counts = Counter(str(f.get("ruleId", "")) for f in fs)
    for rule_id, max_count in expect.get("max_by_rule", {}).items():
        maximum = nonnegative_int(max_count)
        if maximum is None:
            failures.append(f"rule {rule_id} has invalid max count {max_count}")
        elif counts[rule_id] > maximum:
            failures.append(
                f"rule {rule_id} count {counts[rule_id]} exceeds max {max_count}"
            )

    for rule in expect.get("must_not_report", []):
        hits = [f for f in fs if matches_rule(f, rule)]
        if hits:
            failures.append(
                f"must-not-report matched {len(hits)} finding(s): {json.dumps(rule, sort_keys=True)}"
            )

    for rule in expect.get("must_report", []):
        hits = [f for f in fs if matches_rule(f, rule)]
        if not hits:
            failures.append(f"must-report missing: {json.dumps(rule, sort_keys=True)}")

    max_total = expect.get("max_total")
    maximum = nonnegative_int(max_total)
    if max_total is not None and maximum is None:
        failures.append(f"invalid max_total {max_total}")
    elif maximum is not None and len(fs) > maximum:
        failures.append(f"total findings {len(fs)} exceeds max_total {max_total}")

    min_total = expect.get("min_total")
    minimum = nonnegative_int(min_total)
    if min_total is not None and minimum is None:
        failures.append(f"invalid min_total {min_total}")
    elif minimum is not None and len(fs) < minimum:
        failures.append(f"total findings {len(fs)} below min_total {min_total}")

    summary = {
        "name": spec.get("name", args.spec.stem),
        "report": str(args.report),
        "total": len(fs),
        "by_rule": dict(sorted(counts.items())),
        "failures": failures,
        "warnings": warnings,
    }
    print(json.dumps(summary, indent=2, sort_keys=True))
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())