pragent/tests/pilot/test_eval_scores.py

"""Tests for the deterministic review scorers."""
import os
import sys

import pytest

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))

import eval_scores as es  # noqa: E402


def f(sev, path="a.py", line=1):
    return {"severity": sev, "path": path, "line": line, "problem": "p", "fix": ""}


# --- finding_rate ---------------------------------------------------------

def test_finding_rate_counts_findings():
    assert es.finding_rate([f("high"), f("low")]) == 2.0


def test_finding_rate_zero_for_silent_review():
    assert es.finding_rate([]) == 0.0
    assert es.finding_rate(None) == 0.0


# --- severity_info_ratio --------------------------------------------------

def test_info_ratio_all_advisory():
    assert es.severity_info_ratio([f("info"), f("trivial")]) == 1.0


def test_info_ratio_mixed():
    assert es.severity_info_ratio([f("info"), f("high")]) == 0.5


def test_info_ratio_none_when_no_findings():
    # Undefined, not zero — zero would read as perfectly calibrated.
    assert es.severity_info_ratio([]) is None


def test_info_ratio_unknown_severity_treated_as_medium():
    # Matches _normalize_finding's fallback, so an odd severity is not
    # silently counted as advisory.
    assert es.severity_info_ratio([f("bogus")]) == 0.0


# --- severity_max ---------------------------------------------------------

def test_severity_max_picks_highest():
    assert es.severity_max([f("info"), f("critical"), f("low")]) == "critical"


def test_severity_max_none_when_silent():
    assert es.severity_max([]) == "none"


def test_severity_max_case_insensitive():
    assert es.severity_max([f("HIGH")]) == "high"


# --- dropped_findings -----------------------------------------------------

def test_dropped_findings_delta():
    assert es.dropped_findings(5, 2) == 3.0


def test_dropped_findings_never_negative():
    assert es.dropped_findings(1, 3) == 0.0


def test_dropped_findings_none_when_unknown():
    assert es.dropped_findings(None, 2) is None


# --- cost_per_finding -----------------------------------------------------

def test_cost_per_finding_divides():
    assert es.cost_per_finding(1.0, [f("high"), f("low")]) == 0.5


def test_cost_per_finding_silent_review_divides_by_one():
    # The run still cost money; attributing all of it to "found nothing" is
    # the honest reading, and it avoids a division by zero.
    assert es.cost_per_finding(0.25, []) == 0.25


def test_cost_per_finding_none_when_unpriced():
    assert es.cost_per_finding(None, [f("high")]) is None


def test_cost_per_finding_none_on_garbage():
    assert es.cost_per_finding("abc", [f("high")]) is None


# --- build_scores ---------------------------------------------------------

def _by_name(events):
    return {e["body"]["name"]: e["body"] for e in events}


def test_build_scores_emits_expected_set():
    events = es.build_scores(
        trace_id="t1", findings=[f("high"), f("info")], environment="claude",
        cost_usd=0.5, dropped_count=2, timestamp="2026-01-01T00:00:00Z",
    )
    names = _by_name(events)
    assert set(names) == {
        es.FINDING_RATE, es.SEVERITY_INFO_RATIO, es.SEVERITY_MAX,
        es.DROPPED_FINDINGS, es.COST_PER_FINDING,
    }
    assert names[es.FINDING_RATE]["value"] == 2.0
    assert names[es.SEVERITY_MAX]["value"] == "high"
    assert names[es.DROPPED_FINDINGS]["value"] == 2.0
    assert names[es.COST_PER_FINDING]["value"] == 0.25


def test_build_scores_all_events_are_score_create_on_the_trace():
    events = es.build_scores(
        trace_id="t9", findings=[f("low")], environment="ollama", cost_usd=1.0,
    )
    assert all(e["type"] == "score-create" for e in events)
    assert all(e["body"]["traceId"] == "t9" for e in events)
    assert all(e["body"]["environment"] == "ollama" for e in events)


def test_build_scores_omits_undefined_scores():
    # No cost and no drop count measured -> those scores are absent, not zero.
    events = es.build_scores(trace_id="t2", findings=[], environment="ollama")
    names = set(_by_name(events))
    assert es.COST_PER_FINDING not in names
    assert es.DROPPED_FINDINGS not in names
    assert es.SEVERITY_INFO_RATIO not in names
    assert names == {es.FINDING_RATE, es.SEVERITY_MAX}


def test_build_scores_categorical_value_is_string():
    events = es.build_scores(trace_id="t3", findings=[f("high")], environment="claude")
    sev = _by_name(events)[es.SEVERITY_MAX]
    assert sev["dataType"] == "CATEGORICAL"
    assert isinstance(sev["value"], str)


def test_build_scores_numeric_values_are_floats():
    events = es.build_scores(
        trace_id="t4", findings=[f("high")], environment="claude", cost_usd=1,
    )
    for name, body in _by_name(events).items():
        if body["dataType"] == "NUMERIC":
            assert isinstance(body["value"], float), name


def test_build_scores_comment_propagates():
    events = es.build_scores(
        trace_id="t5", findings=[f("high")], environment="claude",
        cost_usd=1.0, comment="cost basis: equivalent:claude-sonnet-5",
    )
    assert all("equivalent" in e["body"]["comment"] for e in events)


# --- score configs --------------------------------------------------------

def test_every_emitted_score_has_a_config():
    configured = {c["name"] for c in es.SCORE_CONFIGS}
    events = es.build_scores(
        trace_id="t6", findings=[f("high")], environment="claude",
        cost_usd=1.0, dropped_count=0,
    )
    assert set(_by_name(events)) <= configured


def test_severity_max_config_covers_every_severity_it_can_emit():
    labels = {c["label"] for c in
              next(c for c in es.SCORE_CONFIGS if c["name"] == es.SEVERITY_MAX)["categories"]}
    assert set(es.SEVERITY_RANK) | {"none"} == labels


# --- ingestion envelope ---------------------------------------------------

def test_every_event_carries_a_timestamp():
    # Ingestion rejects events without one, and reports the rejection as a
    # per-event 400 inside an HTTP 207 that reads as success.
    events = es.build_scores(
        trace_id="t7", findings=[f("high")], environment="claude", cost_usd=1.0,
    )
    assert events
    assert all(e.get("timestamp") for e in events)


def test_timestamp_defaults_when_caller_omits_it():
    events = es.build_scores(trace_id="t8", findings=[f("low")], environment="claude")
    assert all(isinstance(e["timestamp"], str) and e["timestamp"].endswith("Z") for e in events)


def test_explicit_timestamp_is_used():
    events = es.build_scores(
        trace_id="t9", findings=[f("low")], environment="claude",
        timestamp="2026-01-02T03:04:05Z",
    )
    assert all(e["timestamp"] == "2026-01-02T03:04:05Z" for e in events)