Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 123 additions & 7 deletions src/agentops/cli/report_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@

import typer

from agentops.cli._planned import _planned_command
from agentops.utils.logging import get_logger

log = get_logger(__name__)
Expand Down Expand Up @@ -72,12 +71,129 @@ def cmd_report(


@report_app.command("show")
def cmd_report_show() -> None:
"""View reports in table format (planned)."""
_planned_command("agentops report show")
def cmd_report_show(
results_in: Annotated[
Path | None,
typer.Option(
"--in",
help=(
"Path to results.json. "
"If omitted, uses .agentops/results/latest/results.json"
),
),
] = None,
report_format: Annotated[
str,
typer.Option("--format", "-f", help="Report format to display: md or html."),
] = "md",
) -> None:
"""Display an existing evaluation report."""
import webbrowser

resolved_in = (results_in or DEFAULT_REPORT_INPUT).resolve()
results_dir = resolved_in.parent

if report_format == "html":
html_path = results_dir / "report.html"
if not html_path.exists():
typer.echo(
f"Error: report.html not found in {results_dir}. "
"Regenerate with: agentops report --format html",
err=True,
)
raise typer.Exit(code=1)
typer.echo(f"Opening {html_path} in browser...")
webbrowser.open(html_path.as_uri())
else:
md_path = results_dir / "report.md"
if not md_path.exists():
typer.echo(
f"Error: report.md not found in {results_dir}. "
"Regenerate with: agentops report",
err=True,
)
raise typer.Exit(code=1)
typer.echo(md_path.read_text(encoding="utf-8"))


@report_app.command("export")
def cmd_report_export() -> None:
"""Export reports in JSON/Markdown/CSV formats (planned)."""
_planned_command("agentops report export")
def cmd_report_export(
results_in: Annotated[
Path | None,
typer.Option(
"--in",
help=(
"Path to results.json. "
"If omitted, uses .agentops/results/latest/results.json"
),
),
] = None,
output: Annotated[
Path | None,
typer.Option("--out", "-o", help="Output file path."),
] = None,
export_format: Annotated[
str,
typer.Option("--format", "-f", help="Export format: json, csv, or md."),
] = "json",
) -> None:
"""Export evaluation results to JSON, CSV, or Markdown."""
import csv
import io
import json

resolved_in = (results_in or DEFAULT_REPORT_INPUT).resolve()

if not resolved_in.exists():
typer.echo(f"Error: results.json not found: {resolved_in}", err=True)
raise typer.Exit(code=1)

data = json.loads(resolved_in.read_text(encoding="utf-8"))

if export_format == "json":
content = json.dumps(data, indent=2)
elif export_format == "csv":
metrics = data.get("metrics", [])
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["metric", "value"])
for m in metrics:
writer.writerow([m.get("name", ""), m.get("value", "")])

# Add row-level metrics if available
row_metrics = data.get("row_metrics", [])
if row_metrics:
writer.writerow([])
# Collect all metric names across rows
metric_names: list[str] = []
for row in row_metrics:
for m in row.get("metrics", []):
if m["name"] not in metric_names:
metric_names.append(m["name"])
writer.writerow(["row_index"] + metric_names)
for row in row_metrics:
values = {m["name"]: m["value"] for m in row.get("metrics", [])}
writer.writerow(
[row.get("row_index", "")]
+ [values.get(n, "") for n in metric_names]
)
content = buf.getvalue()
elif export_format == "md":
md_path = resolved_in.parent / "report.md"
if not md_path.exists():
typer.echo(
"Error: report.md not found. Regenerate with: agentops report",
err=True,
)
raise typer.Exit(code=1)
content = md_path.read_text(encoding="utf-8")
else:
typer.echo("Error: --format must be json, csv, or md.", err=True)
raise typer.Exit(code=1)

if output:
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(content, encoding="utf-8")
typer.echo(f"Exported to {output}")
else:
typer.echo(content)
250 changes: 250 additions & 0 deletions tests/unit/test_report_commands.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,250 @@
"""Tests for report show and report export commands."""

from __future__ import annotations

import csv
import io
import json
from pathlib import Path

from typer.testing import CliRunner

from agentops.cli.app import app

runner = CliRunner()


def _create_results_dir(tmp_path: Path, *, with_html: bool = False) -> Path:
"""Create a results directory with results.json and report.md."""
results_dir = tmp_path / ".agentops" / "results" / "latest"
results_dir.mkdir(parents=True)

results = {
"version": 1,
"status": "completed",
"bundle": {"name": "test_bundle", "path": "bundles/test.yaml"},
"dataset": {"name": "test_dataset", "path": "datasets/test.yaml"},
"execution": {
"backend": "foundry",
"command": "test",
"started_at": "2026-04-07T10:00:00Z",
"finished_at": "2026-04-07T10:01:00Z",
"duration_seconds": 60.0,
"exit_code": 0,
},
"metrics": [
{"name": "CoherenceEvaluator", "value": 4.5},
{"name": "RelevanceEvaluator", "value": 5.0},
{"name": "samples_evaluated", "value": 3.0},
],
"row_metrics": [
{
"row_index": 1,
"metrics": [
{"name": "CoherenceEvaluator", "value": 4.0},
{"name": "RelevanceEvaluator", "value": 5.0},
],
},
{
"row_index": 2,
"metrics": [
{"name": "CoherenceEvaluator", "value": 5.0},
{"name": "RelevanceEvaluator", "value": 5.0},
],
},
],
"item_evaluations": [],
"thresholds": [],
"summary": {
"metrics_count": 3,
"thresholds_count": 0,
"thresholds_passed": 0,
"thresholds_failed": 0,
"overall_passed": True,
},
}
(results_dir / "results.json").write_text(
json.dumps(results, indent=2), encoding="utf-8"
)
(results_dir / "report.md").write_text(
"# Test Report\n\nAll passed.\n", encoding="utf-8"
)
if with_html:
(results_dir / "report.html").write_text(
"<html><body>Report</body></html>", encoding="utf-8"
)
return results_dir


# ---------------------------------------------------------------------------
# report show
# ---------------------------------------------------------------------------


class TestReportShow:
def test_show_md(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
["report", "show", "--in", str(results_dir / "results.json")],
)
assert result.exit_code == 0
assert "# Test Report" in result.stdout
assert "All passed." in result.stdout

def test_show_md_default_format(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
["report", "show", "--in", str(results_dir / "results.json")],
)
assert result.exit_code == 0
assert "# Test Report" in result.stdout

def test_show_md_missing(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
(results_dir / "report.md").unlink()
result = runner.invoke(
app,
["report", "show", "--in", str(results_dir / "results.json")],
)
assert result.exit_code == 1

def test_show_html_missing(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
[
"report",
"show",
"--in",
str(results_dir / "results.json"),
"--format",
"html",
],
)
assert result.exit_code == 1
assert "report.html not found" in (result.stdout + result.stderr)


# ---------------------------------------------------------------------------
# report export
# ---------------------------------------------------------------------------


class TestReportExport:
def test_export_json_stdout(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--format",
"json",
],
)
assert result.exit_code == 0
data = json.loads(result.stdout)
assert data["status"] == "completed"
assert len(data["metrics"]) == 3

def test_export_json_to_file(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
out_file = tmp_path / "export.json"
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--out",
str(out_file),
],
)
assert result.exit_code == 0
assert out_file.exists()
data = json.loads(out_file.read_text(encoding="utf-8"))
assert data["status"] == "completed"

def test_export_csv_stdout(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--format",
"csv",
],
)
assert result.exit_code == 0
reader = csv.reader(io.StringIO(result.stdout))
rows = list(reader)
# Header + 3 metrics + blank + row header + 2 row metrics
assert rows[0] == ["metric", "value"]
assert rows[1][0] == "CoherenceEvaluator"

def test_export_csv_to_file(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
out_file = tmp_path / "export.csv"
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--format",
"csv",
"--out",
str(out_file),
],
)
assert result.exit_code == 0
assert out_file.exists()
content = out_file.read_text(encoding="utf-8")
assert "CoherenceEvaluator" in content

def test_export_md_stdout(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--format",
"md",
],
)
assert result.exit_code == 0
assert "# Test Report" in result.stdout

def test_export_invalid_format(self, tmp_path: Path) -> None:
results_dir = _create_results_dir(tmp_path)
result = runner.invoke(
app,
[
"report",
"export",
"--in",
str(results_dir / "results.json"),
"--format",
"xml",
],
)
assert result.exit_code == 1

def test_export_missing_results(self, tmp_path: Path) -> None:
result = runner.invoke(
app,
["report", "export", "--in", str(tmp_path / "nonexistent.json")],
)
assert result.exit_code == 1