Spaces:
Running
Running
File size: 3,360 Bytes
978db0e d771e01 978db0e d771e01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | from __future__ import annotations
import json
from pathlib import Path
import pytest
from trace2mcp.cli import main
def _write_trace(path: Path) -> Path:
path.write_text(
json.dumps(
{
"version": "1",
"inputs": {"query": "open source agents"},
"events": [
{
"id": "search",
"tool": "search",
"arguments": {"query": "open source agents"},
"result": {"title": "Trace2MCP"},
"status": "success",
"side_effect": "read",
"duration_ms": 1,
},
{
"id": "format",
"tool": "format",
"arguments": {"title": "Trace2MCP"},
"result": "Trace2MCP",
"status": "success",
"side_effect": "read",
"duration_ms": 1,
},
],
}
),
encoding="utf-8",
)
return path
def test_inspect_json(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
trace = _write_trace(tmp_path / "trace.json")
assert main(["inspect", str(trace), "--json"]) == 0
output = json.loads(capsys.readouterr().out)
assert output["events"] == 2
assert output["side_effects"] == {"read": 2}
assert output["tools"] == {"search": 1, "format": 1}
def test_compile_generates_safe_project(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
trace = _write_trace(tmp_path / "trace.json")
destination = tmp_path / "generated"
assert main(["compile", str(trace), "--out", str(destination)]) == 0
workflow = json.loads((destination / "workflow.json").read_text(encoding="utf-8"))
assert workflow["name"] == "compiled_workflow"
assert [node["id"] for node in workflow["nodes"]] == ["search", "format"]
assert (destination / "server.py").is_file()
assert (destination / "handlers.py").is_file()
assert "Next: trace2mcp serve" in capsys.readouterr().out
def test_benchmark_json_reports_measured_times(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
trace = _write_trace(tmp_path / "trace.json")
assert main(["benchmark", str(trace), "--repetitions", "2", "--json"]) == 0
result = json.loads(capsys.readouterr().out)
assert result["repetitions"] == 2
assert result["baseline_calls"] == 2
assert result["calls"] == 2
assert len(result["sequential_wall_ms"]) == 2
assert len(result["dag_wall_ms"]) == 2
assert result["recorded_work_ms"] == pytest.approx(2.0)
def test_demo_runs_full_offline_pipeline(capsys: pytest.CaptureFixture[str]) -> None:
assert main(["demo", "--json"]) == 0
report = json.loads(capsys.readouterr().out)
assert report["pipeline"] == ["TEACH", "COMPILE", "VERIFY", "RUN"]
assert report["compilation"]["workflow"] == "company_report"
assert report["compilation"]["source_steps"] == 6
assert report["compilation"]["deduplicated_reads"] == 1
assert report["verification"]["valid"] is True
assert report["frozen_replay"]["mode"] == "frozen"
assert report["safety_gate"]["decision"] == "approval"
|