test/eval/results/test_result.py
Rama main · réplica de lectura del repositorio original.
# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0
"""Public-API tests for :class:`RunResult` — aggregation math, summary, save."""
import json
from pathlib import Path
import pytest
from ag2.eval import (
Aggregates,
Feedback,
RunResult,
ScoreStats,
Suite,
Task,
TaskResult,
Trace,
)
def _empty_trace(duration_ms: int = 0) -> Trace:
return Trace(events=[], exception=None, duration_ms=duration_ms)
def _task_result(
task_id: str,
feedback: tuple[Feedback, ...],
*,
budget_violation: bool = False,
exception: BaseException | None = None,
tags: tuple[str, ...] = (),
) -> TaskResult:
return TaskResult(
task=Task(task_id=task_id, inputs={"input": "?"}, tags=tags),
trace=Trace(events=[], exception=exception, duration_ms=0),
feedback=feedback,
budget_violation=budget_violation,
)
def _result(*task_results: TaskResult, run_id: str = "test-run") -> RunResult:
suite = Suite.from_list([{"task_id": tr.task.task_id, "inputs": tr.task.inputs} for tr in task_results])
return RunResult(
run_id=run_id,
tasks=tuple(task_results),
suite=suite,
target_path="test_module:test_factory",
concurrency=1,
duration_ms=42,
created_at="2026-05-11T00:00:00+00:00",
)
class TestPassRate:
def test_all_pass(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="check", score=True),)),
_task_result("t2", (Feedback(key="check", score=True),)),
)
assert result.pass_rate("check") == 1.0
def test_partial_pass(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="check", score=True),)),
_task_result("t2", (Feedback(key="check", score=False),)),
_task_result("t3", (Feedback(key="check", score=True),)),
_task_result("t4", (Feedback(key="check", score=False),)),
)
assert result.pass_rate("check") == 0.5
def test_zero_pass(self) -> None:
result = _result(_task_result("t1", (Feedback(key="check", score=False),)))
assert result.pass_rate("check") == 0.0
def test_unknown_key_returns_zero(self) -> None:
result = _result(_task_result("t1", (Feedback(key="check", score=True),)))
assert result.pass_rate("does-not-exist") == 0.0
def test_numeric_scores_do_not_contribute_to_pass_rate(self) -> None:
"""A scorer that returns 5 is not 'pass=True'. Pass rate stays empty for that key."""
result = _result(_task_result("t1", (Feedback(key="counter", score=5),)))
assert result.pass_rate("counter") == 0.0
class TestScoreStats:
def test_basic_stats(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="extra", score=0),)),
_task_result("t2", (Feedback(key="extra", score=1),)),
_task_result("t3", (Feedback(key="extra", score=2),)),
)
stats = result.score_stats("extra")
assert stats.n == 3
assert stats.mean == 1.0
assert stats.p50 == 1.0
def test_float_scores_supported(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="ratio", score=0.25),)),
_task_result("t2", (Feedback(key="ratio", score=0.75),)),
)
stats = result.score_stats("ratio")
assert stats.n == 2
assert stats.mean == pytest.approx(0.5)
def test_unknown_key_returns_empty_stats(self) -> None:
result = _result(_task_result("t1", (Feedback(key="x", score=1),)))
stats = result.score_stats("not-there")
assert stats == ScoreStats(mean=0.0, p50=0.0, p95=0.0, n=0)
def test_bool_feedback_does_not_enter_score_stats(self) -> None:
"""bool is a subclass of int but should not become numeric score stats."""
result = _result(_task_result("t1", (Feedback(key="b", score=True),)))
assert result.score_stats("b").n == 0
class TestValueCounts:
def test_counts_labels(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="reason", value="completed"),)),
_task_result("t2", (Feedback(key="reason", value="completed"),)),
_task_result("t3", (Feedback(key="reason", value="halted"),)),
)
assert result.value_counts("reason") == {"completed": 2, "halted": 1}
def test_unknown_key_returns_empty_dict(self) -> None:
result = _result(_task_result("t1", (Feedback(key="x", value="a"),)))
assert result.value_counts("not-there") == {}
class TestAggregates:
def test_aggregates_property_exposes_everything(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="ok", score=True), Feedback(key="why", value="completed"))),
_task_result("t2", (Feedback(key="ok", score=False), Feedback(key="why", value="halted"))),
_task_result("t3", (Feedback(key="ok", score=True), Feedback(key="why", value="completed"))),
)
aggs = result.aggregates
assert isinstance(aggs, Aggregates)
assert aggs.pass_rate == {"ok": pytest.approx(2 / 3)}
assert aggs.value_counts == {"why": {"completed": 2, "halted": 1}}
assert aggs.errors == 0
assert aggs.budget_violations == 0
def test_errors_counted(self) -> None:
result = _result(
_task_result("t1", (), exception=RuntimeError("boom")),
_task_result("t2", (Feedback(key="ok", score=True),)),
)
assert result.aggregates.errors == 1
def test_budget_violations_counted(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="ok", score=True),), budget_violation=True),
_task_result("t2", (Feedback(key="ok", score=True),), budget_violation=False),
)
assert result.aggregates.budget_violations == 1
class TestTagSlicing:
def test_pass_rate_by_tag(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="check", score=True),), tags=("easy",)),
_task_result("t2", (Feedback(key="check", score=True),), tags=("easy",)),
_task_result("t3", (Feedback(key="check", score=False),), tags=("hard",)),
_task_result("t4", (Feedback(key="check", score=False),), tags=("hard",)),
)
assert result.pass_rate("check") == 0.5 # whole run
assert result.pass_rate("check", tag="easy") == 1.0
assert result.pass_rate("check", tag="hard") == 0.0
def test_value_counts_by_tag(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="reason", value="completed"),), tags=("easy",)),
_task_result("t2", (Feedback(key="reason", value="error"),), tags=("hard",)),
)
assert result.value_counts("reason", tag="easy") == {"completed": 1}
assert result.value_counts("reason", tag="hard") == {"error": 1}
def test_tags_lists_all_present(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="check", score=True),), tags=("easy", "smoke")),
_task_result("t2", (Feedback(key="check", score=True),), tags=("hard",)),
)
assert result.tags == frozenset({"easy", "smoke", "hard"})
def test_unknown_tag_is_an_empty_slice(self) -> None:
result = _result(_task_result("t1", (Feedback(key="check", score=True),), tags=("easy",)))
assert result.pass_rate("check", tag="nope") == 0.0
class TestSummary:
def test_includes_run_id_duration_and_pass_rate(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="ok", score=True),)),
_task_result("t2", (Feedback(key="ok", score=False),)),
run_id="abc123",
)
summary = result.summary()
assert "abc123" in summary
assert "42ms" in summary
assert "ok" in summary
assert "50.0% (1/2)" in summary # pass-rate shows the (passed/total) denominator
assert "Runs:" in summary # number of runs executed is surfaced
def test_renders_score_stats_and_value_counts(self) -> None:
result = _result(
_task_result("t1", (Feedback(key="count", score=2), Feedback(key="end", value="completed"))),
_task_result("t2", (Feedback(key="count", score=4), Feedback(key="end", value="completed"))),
)
summary = result.summary()
assert "count" in summary
assert "mean=3.00" in summary
assert "completed=2" in summary
class TestSave:
def test_save_to_directory(self, tmp_path: Path) -> None:
result = _result(
_task_result("t1", (Feedback(key="ok", score=True),)),
run_id="my-run-id",
)
written = result.save(tmp_path)
assert written == tmp_path / "my-run-id.json"
assert written.exists()
data = json.loads(written.read_text(encoding="utf-8"))
assert data["run_id"] == "my-run-id"
assert data["schema_version"] == "0.2"
def test_save_to_explicit_json_path(self, tmp_path: Path) -> None:
result = _result(_task_result("t1", (Feedback(key="ok", score=True),)))
target = tmp_path / "subdir" / "custom.json"
written = result.save(target)
assert written == target
assert target.exists()
def test_save_with_store_dir_set_at_construction(self, tmp_path: Path) -> None:
suite = Suite.from_list([{"task_id": "t1", "inputs": {"input": "?"}}])
result = RunResult(
run_id="cfg-run",
tasks=(_task_result("t1", (Feedback(key="ok", score=True),)),),
suite=suite,
target_path="x:y",
concurrency=1,
duration_ms=1,
created_at="2026-05-11T00:00:00+00:00",
store_dir=tmp_path,
)
written = result.save() # no path argument
assert written == tmp_path / "cfg-run.json"
assert written.exists()
def test_save_without_path_or_store_dir_raises(self) -> None:
result = _result(_task_result("t1", (Feedback(key="ok", score=True),)))
with pytest.raises(ValueError, match="no path given and no store_dir"):
result.save()
class TestRunResultProperties:
def test_run_id_and_schema(self) -> None:
result = _result(_task_result("t1", ()), run_id="r1")
assert result.run_id == "r1"
assert result.schema_version == "0.2"
def test_tasks_round_trip(self) -> None:
tr = _task_result("t1", (Feedback(key="ok", score=True),))
result = _result(tr)
assert result.tasks == (tr,)