interface:
  display_name: "SARIF Parsing"
  short_description: "Parse and process SARIF static analysis results"
  icon_small: "assets/trail-of-bits-mark.svg"
  icon_large: "assets/trail-of-bits-mark.svg"
  brand_color: "#D83A34"

# SARIF jq Query Reference

Ready-to-use jq queries for common SARIF parsing tasks.

## Severity resolution (read this first)

`result.level` is optional in SARIF 2.1.0. CodeQL omits it on every result and stores
severity on the rule instead, as `defaultConfiguration.level`; the result inherits it.
So `select(.level == "error")` matches nothing on a CodeQL run no matter how many
errors it found, and a CI gate written that way exits 0 on a failing repo.

Resolve it instead: `result.level` when present, otherwise the matched rule's
`defaultConfiguration.level`, otherwise `"warning"` (the SARIF default). Join the rule
on `ruleIndex` when the tool populates it, on `ruleId` when it does not.

```bash
# Paste once per shell. `jq "$LEVEL_FN"'<query>'` concatenates the two into one filter.
LEVEL_FN='
  def rule($run):
    . as $r
    | ($run.tool.driver.rules // []) as $rules
    | (if ($r.ruleIndex | type) == "number" and $r.ruleIndex >= 0
       then $rules[$r.ruleIndex] else null end)
      // first($rules[] | select(.id == $r.ruleId))
      // null;
  def level($run):
    . as $r
    | if ($r.kind // "fail") != "fail" then "none"
      else ($r.level // rule($run).defaultConfiguration.level // "warning") end;
'

# Severity of every result, whichever way the tool recorded it
jq -r "$LEVEL_FN"'.runs[] as $run | $run.results[] | "\(level($run))\t\(.ruleId)"' results.sarif

# The CI gate: count of error-level findings, however the tool recorded severity
jq "$LEVEL_FN"'[.runs[] as $run | $run.results[] | select(level($run) == "error")] | length' results.sarif
```

`kind` other than `"fail"` marks a pass/notApplicable record rather than a finding, and
resolves to `"none"`. Without that clause a compliance tool's passing checks inherit
their rule's `error` and fail the build. The `>= 0` guard matters as much: SARIF writes
`ruleIndex: -1` for "no rule", and `$rules[-1]` in jq is the *last* rule, so dropping it
labels those results with whatever severity the final rule in the array happens to have.

Two fixtures under `fixtures/` exercise both shapes: `codeql-no-level.sarif` (severity on
the rules only) and `levels-on-results.sarif` (severity on the results). Each holds exactly
one error, so any query below can be checked against a known answer.

## Basic Exploration

```bash
# Pretty print
jq '.' results.sarif

# Get SARIF version
jq '.version' results.sarif

# List tool names from all runs
jq '.runs[].tool.driver.name' results.sarif

# Count runs
jq '.runs | length' results.sarif
```

## Result Queries

```bash
# Total result count
jq '[.runs[].results[]] | length' results.sarif

# Count by severity level
jq "$LEVEL_FN"'reduce (.runs[] as $run | $run.results[] | level($run)) as $l ({}; .[$l] += 1)' results.sarif

# List unique rule IDs
jq '[.runs[].results[].ruleId] | unique | sort' results.sarif

# Count per rule
jq '[.runs[].results[]] | group_by(.ruleId) | map({rule: .[0].ruleId, count: length}) | sort_by(-.count)' results.sarif
```

## Filtering Results

```bash
# Only errors
jq "$LEVEL_FN"'.runs[] as $run | $run.results[] | select(level($run) == "error")' results.sarif

# Only warnings
jq "$LEVEL_FN"'.runs[] as $run | $run.results[] | select(level($run) == "warning")' results.sarif

# By specific rule ID
jq --arg rule "SQL_INJECTION" '.runs[].results[] | select(.ruleId == $rule)' results.sarif

# By file path (contains)
jq --arg file "auth" '.runs[].results[] | select(.locations[].physicalLocation.artifactLocation.uri | contains($file))' results.sarif

# By file extension
jq '.runs[].results[] | select(.locations[].physicalLocation.artifactLocation.uri | test("\\.py$"))' results.sarif

# Multiple conditions
jq "$LEVEL_FN"'.runs[] as $run | $run.results[] | select(level($run) == "error" and (.ruleId | startswith("SEC")))' results.sarif
```

## Extracting Locations

```bash
# File and line for each result
jq '.runs[].results[] | {
  rule: .ruleId,
  file: .locations[0].physicalLocation.artifactLocation.uri,
  line: .locations[0].physicalLocation.region.startLine
}' results.sarif

# Unique affected files
jq '[.runs[].results[].locations[].physicalLocation.artifactLocation.uri] | unique | sort' results.sarif

# Results grouped by file
jq '[.runs[].results[] | {file: .locations[0].physicalLocation.artifactLocation.uri, result: .}] | group_by(.file) | map({file: .[0].file, count: length})' results.sarif
```

## Rule Information

```bash
# List all rules with severity
jq '.runs[].tool.driver.rules[] | {id: .id, name: .name, level: .defaultConfiguration.level}' results.sarif

# Get rule description by ID
jq --arg id "RULE001" '.runs[].tool.driver.rules[] | select(.id == $id)' results.sarif

# Rules with help URLs
jq '.runs[].tool.driver.rules[] | select(.helpUri) | {id: .id, help: .helpUri}' results.sarif
```

## Fingerprints

```bash
# Results with fingerprints
jq '.runs[].results[] | select(.fingerprints or .partialFingerprints) | {rule: .ruleId, fp: (.fingerprints // .partialFingerprints)}' results.sarif

# Extract all partial fingerprints
jq '[.runs[].results[].partialFingerprints] | add' results.sarif
```

## Aggregation and Reporting

```bash
# Summary by severity and rule
jq "$LEVEL_FN"'[.runs[] as $run | $run.results[] | {level: level($run), ruleId}] | group_by(.level) | map({level: .[0].level, rules: (group_by(.ruleId) | map({rule: .[0].ruleId, count: length}))})' results.sarif

# Top 10 most frequent rules
jq '[.runs[].results[]] | group_by(.ruleId) | map({rule: .[0].ruleId, count: length}) | sort_by(-.count) | .[0:10]' results.sarif

# Files with most issues
jq '[.runs[].results[] | .locations[0].physicalLocation.artifactLocation.uri] | group_by(.) | map({file: .[0], count: length}) | sort_by(-.count) | .[0:10]' results.sarif
```

## Output Formatting

```bash
# CSV-like output
jq -r "$LEVEL_FN"'.runs[] as $run | $run.results[] | [.ruleId, level($run), .locations[0].physicalLocation.artifactLocation.uri, .locations[0].physicalLocation.region.startLine, .message.text] | @csv' results.sarif

# Tab-separated
jq -r "$LEVEL_FN"'.runs[] as $run | $run.results[] | [.ruleId, level($run), .locations[0].physicalLocation.artifactLocation.uri // "N/A"] | @tsv' results.sarif

# Markdown table
echo "| Rule | Level | File | Line |"
echo "|------|-------|------|------|"
jq -r "$LEVEL_FN"'.runs[] as $run | $run.results[] | "| \(.ruleId) | \(level($run)) | \(.locations[0].physicalLocation.artifactLocation.uri // "N/A") | \(.locations[0].physicalLocation.region.startLine // "N/A") |"' results.sarif
```

## Comparison and Diff

```bash
# Find rules in file1 not in file2
comm -23 <(jq -r '[.runs[].results[].ruleId] | unique | sort[]' file1.sarif) <(jq -r '[.runs[].results[].ruleId] | unique | sort[]' file2.sarif)

# Compare result counts
echo "File 1: $(jq '[.runs[].results[]] | length' file1.sarif)"
echo "File 2: $(jq '[.runs[].results[]] | length' file2.sarif)"
```

## Transformation

```bash
# Extract minimal SARIF (results only)
jq '{version: .version, runs: [.runs[] | {tool: {driver: {name: .tool.driver.name}}, results: .results}]}' results.sarif

# Filter and create new SARIF with only errors, per run: the rules a result
# inherits from live in its own run, so runs must not be flattened together
jq "$LEVEL_FN"'.runs |= map(. as $run | .results = [.results[] | select(level($run) == "error")])' results.sarif > errors-only.sarif

# Merge multiple SARIF files
jq -s '{version: "2.1.0", runs: [.[].runs[]]}' file1.sarif file2.sarif > merged.sarif
```

## Validation Checks

```bash
# Check if version is 2.1.0
jq -e '.version == "2.1.0"' results.sarif && echo "Valid version" || echo "Invalid version"

# Check for empty results
jq -e '[.runs[].results[]] | length > 0' results.sarif && echo "Has results" || echo "No results"

# Verify all results have locations
jq '[.runs[].results[] | select(.locations | length == 0)] | length' results.sarif
```

"""
SARIF Parsing Helper Functions

Reusable utilities for working with SARIF files.
No external dependencies beyond standard library.
"""

import hashlib
import json
from collections import defaultdict
from collections.abc import Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import unquote

# What a result's severity is when neither the result nor its rule states one.
SARIF_DEFAULT_LEVEL = "warning"


@dataclass
class Finding:
    """Structured representation of a SARIF result."""

    rule_id: str
    level: str  # resolved by resolve_level(), not read from result.level
    message: str
    file_path: str | None = None
    start_line: int | None = None
    end_line: int | None = None
    start_column: int | None = None
    end_column: int | None = None
    fingerprint: str | None = None
    tool_name: str | None = None
    rule_name: str | None = None
    raw: dict = field(default_factory=dict, repr=False)


def load_sarif(path: str | Path) -> dict:
    """Load and parse a SARIF file."""
    with open(path) as f:
        return json.load(f)


def save_sarif(sarif: dict, path: str | Path, indent: int = 2) -> None:
    """Save SARIF data to file."""
    with open(path, "w") as f:
        json.dump(sarif, f, indent=indent)


def validate_version(sarif: dict) -> bool:
    """Check if SARIF version is 2.1.0."""
    return sarif.get("version") == "2.1.0"


def normalize_path(uri: str, base_path: str = "") -> str:
    """Normalize SARIF artifact URI to consistent path."""
    if not uri:
        return ""

    # Remove file:// prefix
    if uri.startswith("file://"):
        uri = uri[7:]

    # URL decode
    uri = unquote(uri)

    # Handle relative paths
    if base_path and not Path(uri).is_absolute():
        uri = str(Path(base_path) / uri)

    return str(Path(uri))


def safe_get(data: dict, *keys, default: Any = None) -> Any:
    """Safely navigate nested dict structure."""
    for key in keys:
        if isinstance(data, dict):
            data = data.get(key, {})
        elif isinstance(data, list) and isinstance(key, int):
            data = data[key] if 0 <= key < len(data) else {}
        else:
            return default
    return data if data != {} else default


def find_rule(result: dict, run: dict) -> dict | None:
    """Find the rule definition a result was produced by.

    Joins on `ruleIndex` first (the cheap, unambiguous key CodeQL populates) and falls
    back to matching `ruleId` against `runs[].tool.driver.rules[].id`, which is all
    Semgrep and most other tools give you.
    """
    rules = safe_get(run, "tool", "driver", "rules", default=[]) or []
    index = result.get("ruleIndex")
    if isinstance(index, int) and not isinstance(index, bool) and 0 <= index < len(rules):
        return rules[index]

    rule_id = result.get("ruleId")
    if rule_id:
        for rule in rules:
            if rule.get("id") == rule_id:
                return rule
    return None


def resolve_level(result: dict, run: dict) -> str:
    """Resolve a result's effective severity, per SARIF 2.1.0 section 3.27.10.

    `result.level` is optional, and CodeQL routinely omits it: severity lives on the
    rule as `defaultConfiguration.level`, and the result inherits it. Reading
    `result.level` directly therefore scores an entire CodeQL run as clean, which is
    why a severity gate written that way exits 0 on a repo full of errors.

    Resolution order:
      1. `kind` other than "fail" (a pass/informational/notApplicable record) is "none"
      2. `result.level` when present
      3. the matched rule's `defaultConfiguration.level`
      4. "warning", the SARIF default

    `invocations[].ruleConfigurationOverrides` can outrank the rule default; no tool in
    common use emits it, so this does not read it.
    """
    if (result.get("kind") or "fail") != "fail":
        return "none"

    level = result.get("level")
    if level:
        return level

    rule = find_rule(result, run)
    if rule:
        rule_level = safe_get(rule, "defaultConfiguration", "level")
        if rule_level:
            return rule_level

    return SARIF_DEFAULT_LEVEL


def extract_location(result: dict) -> tuple[str | None, int | None, int | None]:
    """Extract file path, start line, and end line from result."""
    loc = safe_get(result, "locations", 0, default={})
    phys = loc.get("physicalLocation", {})
    region = phys.get("region", {})

    file_path = safe_get(phys, "artifactLocation", "uri")
    start_line = region.get("startLine")
    end_line = region.get("endLine")

    return file_path, start_line, end_line


def iter_results(sarif: dict) -> Iterator[tuple[dict, dict]]:
    """Iterate over all results with their run context."""
    for run in sarif.get("runs", []):
        for result in run.get("results", []):
            yield result, run


def extract_findings(sarif: dict) -> list[Finding]:
    """Extract all findings as structured objects."""
    findings = []

    for result, run in iter_results(sarif):
        tool_name = safe_get(run, "tool", "driver", "name")
        file_path, start_line, end_line = extract_location(result)

        loc = safe_get(result, "locations", 0, default={})
        phys = loc.get("physicalLocation", {})
        region = phys.get("region", {})
        rule = find_rule(result, run)

        # Get fingerprint
        fp = None
        if result.get("partialFingerprints"):
            fp = next(iter(result["partialFingerprints"].values()), None)
        elif result.get("fingerprints"):
            fp = next(iter(result["fingerprints"].values()), None)

        findings.append(
            Finding(
                rule_id=result.get("ruleId", "unknown"),
                level=resolve_level(result, run),
                message=safe_get(result, "message", "text", default=""),
                file_path=file_path,
                start_line=start_line,
                end_line=end_line,
                start_column=region.get("startColumn"),
                end_column=region.get("endColumn"),
                fingerprint=fp,
                tool_name=tool_name,
                rule_name=rule.get("name") if rule else None,
                raw=result,
            )
        )

    return findings


def filter_by_level(findings: list[Finding], *levels: str) -> list[Finding]:
    """Filter findings by severity level(s)."""
    return [f for f in findings if f.level in levels]


def filter_by_file(findings: list[Finding], pattern: str) -> list[Finding]:
    """Filter findings by file path pattern (substring match)."""
    return [f for f in findings if f.file_path and pattern in f.file_path]


def filter_by_rule(findings: list[Finding], *rule_ids: str) -> list[Finding]:
    """Filter findings by rule ID(s)."""
    return [f for f in findings if f.rule_id in rule_ids]


def sort_by_severity(findings: list[Finding], reverse: bool = False) -> list[Finding]:
    """Sort findings by severity (error > warning > note > none)."""
    severity_order = {"error": 0, "warning": 1, "note": 2, "none": 3}
    return sorted(findings, key=lambda f: severity_order.get(f.level, 99), reverse=reverse)


def group_by_file(findings: list[Finding]) -> dict[str, list[Finding]]:
    """Group findings by file path."""
    grouped = defaultdict(list)
    for f in findings:
        key = f.file_path or "unknown"
        grouped[key].append(f)
    return dict(grouped)


def group_by_rule(findings: list[Finding]) -> dict[str, list[Finding]]:
    """Group findings by rule ID."""
    grouped = defaultdict(list)
    for f in findings:
        grouped[f.rule_id].append(f)
    return dict(grouped)


def count_by_level(findings: list[Finding]) -> dict[str, int]:
    """Count findings by severity level."""
    counts = defaultdict(int)
    for f in findings:
        counts[f.level] += 1
    return dict(counts)


def count_by_rule(findings: list[Finding]) -> dict[str, int]:
    """Count findings by rule ID."""
    counts = defaultdict(int)
    for f in findings:
        counts[f.rule_id] += 1
    return dict(counts)


def compute_fingerprint(result: dict, include_message: bool = True) -> str:
    """Compute stable fingerprint from result data.

    The whole normalized path goes into the hash, directory included. Hashing the
    basename alone gave `src/auth/login.py:42` and `src/admin/login.py:42` one
    fingerprint under the same rule, so `deduplicate()` and `diff_findings()` threw
    away the second finding and called the file fixed.

    The cost is that runs reporting different absolute prefixes for the same file
    (`/github/workspace/src/a.py` vs `/builds/proj/src/a.py`) no longer match. Make the
    URIs repo-relative before fingerprinting when comparing across environments; a
    collision that hides a finding is the worse failure.
    """
    components = [result.get("ruleId", "")]

    file_path, start_line, _ = extract_location(result)
    if file_path:
        # POSIX separators so a fingerprint computed on Windows matches one from CI.
        components.append(Path(normalize_path(file_path)).as_posix())
    if start_line:
        components.append(str(start_line))
    if include_message:
        msg = safe_get(result, "message", "text", default="")
        # First 50 chars of message for stability
        components.append(msg[:50])

    return hashlib.sha256("|".join(components).encode()).hexdigest()[:16]


def deduplicate(findings: list[Finding]) -> list[Finding]:
    """Remove duplicate findings based on fingerprints."""
    seen = set()
    unique = []

    for f in findings:
        key = f.fingerprint or compute_fingerprint(f.raw)
        if key not in seen:
            seen.add(key)
            unique.append(f)

    return unique


def merge_sarif_files(*paths: str | Path) -> dict:
    """Merge multiple SARIF files into one."""
    merged = {
        "version": "2.1.0",
        "$schema": "https://json.schemastore.org/sarif-2.1.0.json",
        "runs": [],
    }

    for path in paths:
        sarif = load_sarif(path)
        merged["runs"].extend(sarif.get("runs", []))

    return merged


def diff_findings(
    baseline: list[Finding], current: list[Finding]
) -> tuple[list[Finding], list[Finding], list[Finding]]:
    """
    Compare two sets of findings.

    Returns:
        - new: findings in current but not baseline
        - fixed: findings in baseline but not current
        - unchanged: findings in both
    """
    baseline_fps = {f.fingerprint or compute_fingerprint(f.raw) for f in baseline}
    current_fps = {f.fingerprint or compute_fingerprint(f.raw) for f in current}

    new = [f for f in current if (f.fingerprint or compute_fingerprint(f.raw)) not in baseline_fps]
    fixed = [
        f for f in baseline if (f.fingerprint or compute_fingerprint(f.raw)) not in current_fps
    ]
    unchanged = [
        f for f in current if (f.fingerprint or compute_fingerprint(f.raw)) in baseline_fps
    ]

    return new, fixed, unchanged


def get_rules(sarif: dict) -> dict[str, dict]:
    """Extract rule definitions from SARIF file."""
    rules = {}
    for run in sarif.get("runs", []):
        for rule in safe_get(run, "tool", "driver", "rules", default=[]):
            rules[rule.get("id", "")] = rule
    return rules


def to_csv_rows(findings: list[Finding]) -> list[list[str]]:
    """Convert findings to CSV-ready rows."""
    rows = [["rule_id", "level", "file", "line", "message"]]
    for f in findings:
        rows.append(
            [
                f.rule_id,
                f.level,
                f.file_path or "",
                str(f.start_line or ""),
                f.message.replace("\n", " ")[:200],
            ]
        )
    return rows


def summary(findings: list[Finding]) -> dict:
    """Generate summary statistics for findings."""
    return {
        "total": len(findings),
        "by_level": count_by_level(findings),
        "by_rule": count_by_rule(findings),
        "files_affected": len(set(f.file_path for f in findings if f.file_path)),
        "rules_triggered": len(set(f.rule_id for f in findings)),
    }


# Example usage
if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        print("Usage: uv run --no-project sarif_helpers.py <sarif_file>")
        sys.exit(1)

    sarif = load_sarif(sys.argv[1])

    if not validate_version(sarif):
        print("Warning: SARIF version is not 2.1.0")

    findings = extract_findings(sarif)
    findings = sort_by_severity(findings)

    print("\nSummary:")
    stats = summary(findings)
    print(f"  Total findings: {stats['total']}")
    print(f"  Files affected: {stats['files_affected']}")
    print(f"  Rules triggered: {stats['rules_triggered']}")
    print("\nBy severity:")
    for level, count in stats["by_level"].items():
        print(f"  {level}: {count}")

    print("\nTop 5 rules:")
    for rule, count in sorted(stats["by_rule"].items(), key=lambda x: -x[1])[:5]:
        print(f"  {rule}: {count}")

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest>=8"]
# ///
"""Tests for sarif_helpers.py, with the weight on severity resolution.

`result.level` is optional in SARIF 2.1.0, and CodeQL never emits it: severity lives on
the rule as `defaultConfiguration.level` and the result inherits it. Every helper here
used to read `result.level` directly, so `filter_by_level(findings, "error")` returned
nothing for a CodeQL run and the documented CI gate exited 0 on a repo full of errors.

fixtures/codeql-no-level.sarif is that shape, and test_naive_level_read_finds_nothing
pins it: if someone "fixes" the fixture by adding levels to its results, that test fails
rather than the suite quietly losing the only case it exists to cover.
fixtures/levels-on-results.sarif is the ordinary shape, and it is here so a resolver
cannot pass by ignoring `result.level` altogether.

The jq tests run the documented commands themselves, extracted from the markdown. A
severity gate that is correct in sarif_helpers.py and wrong in the docs is still a
severity gate that passes on a failing repo.
"""

from __future__ import annotations

import json
import re
import shutil
import subprocess
from pathlib import Path

import pytest
from sarif_helpers import (
    compute_fingerprint,
    deduplicate,
    extract_findings,
    filter_by_level,
    find_rule,
    load_sarif,
    resolve_level,
)

RESOURCES = Path(__file__).resolve().parent
SKILL = RESOURCES.parent / "SKILL.md"
JQ_QUERIES = RESOURCES / "jq-queries.md"
NO_LEVEL = RESOURCES / "fixtures/codeql-no-level.sarif"
WITH_LEVEL = RESOURCES / "fixtures/levels-on-results.sarif"

# The gate every CI pipeline built from this skill runs, in its jq form.
ERROR_GATE = '[.runs[] as $run | $run.results[] | select(level($run) == "error")] | length'
NAIVE_GATE = '[.runs[].results[] | select(.level == "error")] | length'

FENCE = re.compile(r"^```[a-z]*\n(.*?)^```", re.MULTILINE | re.DOTALL)
LEVEL_FN = re.compile(r"LEVEL_FN='\n(.*?)'\n", re.DOTALL)
# The GitHub Actions gate, which inlines its own copy of the resolver.
CI_GATE = re.compile(r"HIGH_COUNT=\$\(jq '\n(.*?)\n\s*' results\.sarif\)", re.DOTALL)


def results_of(sarif: dict) -> list[dict]:
    return [r for run in sarif["runs"] for r in run["results"]]


def jq(program: str, path: Path) -> str:
    """Run jq, failing rather than skipping when it is missing.

    A skip reads as a clean run while the only check that can catch a broken documented
    gate did not execute. jq is this skill's primary tool and CI installs it.
    """
    if not shutil.which("jq"):
        raise AssertionError(
            "jq is not installed, so the documented severity gate went unverified. "
            "Install it (brew install jq); this suite must not pass without it."
        )
    out = subprocess.run(["jq", program, str(path)], capture_output=True, text=True, check=True)
    return out.stdout.strip()


def documented_level_fn(path: Path) -> str:
    """The LEVEL_FN jq definition as the docs publish it."""
    match = LEVEL_FN.search(path.read_text())
    assert match, f"{path.name} no longer defines LEVEL_FN; the jq tests below test nothing"
    return match.group(1)


# --- the fixtures themselves, so the suite cannot go vacuous ------------------------


def test_codeql_fixture_omits_every_result_level():
    """The bug only exists on results with no level of their own."""
    results = results_of(load_sarif(NO_LEVEL))
    assert results, "fixture has no results"
    assert all("level" not in r for r in results)
    rules = load_sarif(NO_LEVEL)["runs"][0]["tool"]["driver"]["rules"]
    assert any(r.get("defaultConfiguration", {}).get("level") == "error" for r in rules)


def test_codeql_fixture_covers_both_join_directions():
    """One result joins by ruleIndex, another only by ruleId."""
    results = results_of(load_sarif(NO_LEVEL))
    assert any("ruleIndex" in r for r in results)
    assert any("ruleIndex" not in r for r in results)


def test_naive_level_read_finds_nothing_on_codeql_output():
    """The bug, stated: reading result.level scores a CodeQL run as clean."""
    results = results_of(load_sarif(NO_LEVEL))
    assert [r for r in results if r.get("level") == "error"] == []
    assert len(filter_by_level(extract_findings(load_sarif(NO_LEVEL)), "error")) == 1


# --- resolution ---------------------------------------------------------------------


def test_severity_inherited_from_rule_defaults():
    levels = {f.rule_id: f.level for f in extract_findings(load_sarif(NO_LEVEL))}
    assert levels == {
        "py/sql-injection": "error",  # defaultConfiguration.level, via ruleIndex
        "py/clear-text-logging-sensitive-data": "warning",  # via ruleId, no ruleIndex
        "py/unused-import": "warning",  # rule has no defaultConfiguration at all
    }


def test_result_level_outranks_rule_default():
    """Both fixtures hold exactly one error; a resolver that ignored result.level would
    report zero here, which is how you notice the fix was an inversion."""
    findings = extract_findings(load_sarif(WITH_LEVEL))
    assert {f.rule_id: f.level for f in findings} == {
        "python.lang.security.audit.dangerous-subprocess-use": "error",
        "python.django.security.audit.avoid-mark-safe": "warning",
    }
    rules = load_sarif(WITH_LEVEL)["runs"][0]["tool"]["driver"]["rules"]
    assert all(r["defaultConfiguration"]["level"] == "warning" for r in rules)
    assert len(filter_by_level(findings, "error")) == 1


@pytest.mark.parametrize("kind", ["pass", "notApplicable", "informational"])
def test_kind_other_than_fail_is_none(kind):
    """A passing compliance check must not inherit its rule's error level."""
    run = {"tool": {"driver": {"rules": [{"id": "r", "defaultConfiguration": {"level": "error"}}]}}}
    assert resolve_level({"ruleId": "r", "kind": kind}, run) == "none"
    assert resolve_level({"ruleId": "r", "kind": "fail"}, run) == "error"


@pytest.mark.parametrize("kind", [None, "fail"])
def test_null_or_fail_kind_resolves_from_the_rule(kind):
    """SARIF's `kind` defaults to "fail", and the jq gate coalesces null with `// "fail"`.
    `.get("kind", "fail")` would instead read an explicit null and hide the error as
    "none", so the resolver must coalesce null to "fail" to agree with the jq gate."""
    run = {"tool": {"driver": {"rules": [{"id": "r", "defaultConfiguration": {"level": "error"}}]}}}
    assert resolve_level({"ruleId": "r", "kind": kind}, run) == "error"


def test_unmatched_rule_falls_back_to_sarif_default():
    run = {"tool": {"driver": {"name": "tool-with-no-rule-metadata"}}}
    assert resolve_level({"ruleId": "r"}, run) == "warning"
    assert find_rule({"ruleId": "r"}, run) is None


@pytest.mark.parametrize("index", [9, -1])
def test_rule_index_outside_the_array_falls_back_to_rule_id(index):
    """-1 is SARIF's "no rule", and it is the dangerous one: `$rules[-1]` in jq is the
    last rule, so an unguarded index labels the result with that rule's severity."""
    run = {
        "tool": {
            "driver": {
                "rules": [
                    {"id": "r", "defaultConfiguration": {"level": "note"}},
                    {"id": "last", "defaultConfiguration": {"level": "error"}},
                ]
            }
        }
    }
    assert resolve_level({"ruleId": "r", "ruleIndex": index}, run) == "note"


# --- the documented jq gate ---------------------------------------------------------


def test_documented_jq_gate_reports_errors_from_rule_defaults():
    """Before and after, against the same bytes: the naive gate misses the error the
    documented one finds, and neither over-reports on ordinary SARIF."""
    level_fn = documented_level_fn(JQ_QUERIES)
    assert jq(NAIVE_GATE, NO_LEVEL) == "0"
    assert jq(level_fn + ERROR_GATE, NO_LEVEL) == "1"
    assert jq(NAIVE_GATE, WITH_LEVEL) == "1"
    assert jq(level_fn + ERROR_GATE, WITH_LEVEL) == "1"


def test_jq_and_python_resolution_agree():
    level_fn = documented_level_fn(JQ_QUERIES)
    program = level_fn + r'[.runs[] as $run | $run.results[] | level($run)] | join(",")'
    for fixture in (NO_LEVEL, WITH_LEVEL):
        from_jq = json.loads(jq(program, fixture)).split(",")
        assert from_jq == [f.level for f in extract_findings(load_sarif(fixture))]


def test_skill_and_reference_publish_the_same_resolution():
    assert documented_level_fn(SKILL) == documented_level_fn(JQ_QUERIES)


def test_the_github_actions_gate_runs_and_counts_the_inherited_error():
    """The gate from issue #262, run as published. It inlines its own copy of the
    resolver because a workflow step has no shell variable to paste into, so comparing
    the LEVEL_FN blocks does not cover it."""
    match = CI_GATE.search(SKILL.read_text())
    assert match, "the GitHub Actions gate is no longer where this test reads it"
    program = match.group(1)
    assert jq(program, NO_LEVEL) == "1"
    assert jq(program, WITH_LEVEL) == "1"


def test_documented_jq_does_not_read_the_last_rule_for_rule_index_minus_one(tmp_path):
    """`$rules[-1]` is the last element in jq, so the documented guard has to reject it."""
    sarif = {
        "version": "2.1.0",
        "runs": [
            {
                "tool": {
                    "driver": {
                        "name": "t",
                        "rules": [
                            {"id": "r", "defaultConfiguration": {"level": "note"}},
                            {"id": "last", "defaultConfiguration": {"level": "error"}},
                        ],
                    }
                },
                "results": [{"ruleId": "r", "ruleIndex": -1, "message": {"text": "m"}}],
            }
        ],
    }
    path = tmp_path / "minus-one.sarif"
    path.write_text(json.dumps(sarif))
    program = documented_level_fn(JQ_QUERIES) + ".runs[] as $run | $run.results[] | level($run)"
    assert json.loads(jq(program, path)) == "note"
    assert jq(documented_level_fn(JQ_QUERIES) + ERROR_GATE, path) == "0"


def unresolved_severity(markdown: str) -> tuple[list[str], int]:
    """Documented jq lines that read a result's `.level` without resolving it, and the
    number of jq blocks scanned.

    Line by line, not block by block: one query reverted to `select(.level == "error")`
    hides inside a block whose other queries still resolve, which is how five of these
    survived review the first time.

    `.defaultConfiguration.level` is the rule's own field, read on purpose by the queries
    that list rule metadata, and `$r.level` is the resolver's own first branch.
    """
    offenders, blocks = [], 0
    for block in FENCE.findall(markdown):
        if "jq " not in block:
            continue
        blocks += 1
        for raw in block.splitlines():
            line = raw.strip()
            if line.startswith("#"):
                continue
            for benign in ("defaultConfiguration.level", "$r.level"):
                line = line.replace(benign, "")
            if ".level" in line and "level($run)" not in line:
                offenders.append(raw.strip())
    return offenders, blocks


def test_no_documented_jq_command_filters_on_bare_result_level():
    """No unresolved read anywhere, and the resolved queries did not vanish either.

    A doc with every severity query deleted would satisfy the first assertion, so the
    second one counts the resolutions that must still be there.
    """
    scanned, resolutions = 0, 0
    for doc in (SKILL, JQ_QUERIES):
        text = doc.read_text()
        offenders, blocks = unresolved_severity(text)
        assert offenders == [], f"{doc.name}: unresolved severity in {offenders}"
        scanned += blocks
        resolutions += text.count("level($run)")
    assert scanned >= 10, f"only {scanned} jq blocks found; block discovery is broken"
    assert resolutions >= 15, f"only {resolutions} resolved queries left in the docs"


def test_the_severity_guard_still_detects_a_regression():
    """The guard above reports nothing by design, so prove it can still report."""
    regressed = """```bash
# select(.level == "error") is what this used to say
jq '.runs[].results[] | select(.level == "error")' results.sarif
jq '.runs[].tool.driver.rules[] | {id: .id, level: .defaultConfiguration.level}' results.sarif
jq "$LEVEL_FN"'.runs[] as $run | $run.results[] | select(level($run) == "error")' out.sarif
```
"""
    offenders, blocks = unresolved_severity(regressed)
    assert blocks == 1
    assert offenders == ["""jq '.runs[].results[] | select(.level == "error")' results.sarif"""]


# --- fingerprints -------------------------------------------------------------------


def result_at(uri: str) -> dict:
    return {
        "ruleId": "py/sql-injection",
        "message": {"text": "This query depends on a user-provided value."},
        "locations": [
            {"physicalLocation": {"artifactLocation": {"uri": uri}, "region": {"startLine": 42}}}
        ],
    }


def test_fingerprint_separates_identical_findings_in_different_directories():
    """Hashing the basename alone made these one finding, so the second was discarded."""
    assert compute_fingerprint(result_at("app/auth/login.py")) != compute_fingerprint(
        result_at("app/admin/login.py")
    )


def test_deduplicate_keeps_a_finding_from_each_directory():
    sarif = {
        "version": "2.1.0",
        "runs": [
            {
                "tool": {"driver": {"name": "CodeQL"}},
                "results": [result_at("app/auth/login.py"), result_at("app/admin/login.py")],
            }
        ],
    }
    assert len(deduplicate(extract_findings(sarif))) == 2


def test_fingerprint_ignores_uri_scheme_and_percent_encoding():
    assert compute_fingerprint(result_at("file:///src/a%20b.py")) == compute_fingerprint(
        result_at("/src/a b.py")
    )


def test_fingerprint_still_matches_the_same_finding_twice():
    assert compute_fingerprint(result_at("app/auth/login.py")) == compute_fingerprint(
        result_at("app/auth/login.py")
    )