Spaces:
Running
Running
| 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" | |