File size: 5,646 Bytes
8a87cbb | 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 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | from __future__ import annotations
import argparse
import json
import pytest
from aegis.cli.app import main
from aegis.cli.commands.run import run_command
from aegis.runtime import AegisRuntime, EvaluationFailureError, RunOutcome
def _args(**kwargs: object) -> argparse.Namespace:
defaults = {
"task": "Review this repository",
"agent": None,
"workflow": None,
"output": "text",
"max_steps": None,
"verbose": False,
}
defaults.update(kwargs)
return argparse.Namespace(**defaults)
class SuccessfulRuntime:
def __init__(self) -> None:
self.calls: list[dict[str, object]] = []
def run(
self,
*,
task: str,
agent_name: str | None = None,
workflow_name: str | None = None,
max_steps: int | None = None,
run_id=None,
) -> RunOutcome:
self.calls.append(
{
"task": task,
"agent_name": agent_name,
"workflow_name": workflow_name,
"max_steps": max_steps,
"run_id": run_id,
}
)
return RunOutcome(
run_id="run_123",
status="completed",
agent=agent_name or "default-task",
workflow=workflow_name or "default",
result={"summary": task},
evaluation={"passed": True, "score": 100},
error=None,
)
class RaisingRuntime:
def __init__(self, exc: BaseException) -> None:
self._exc = exc
def run(self, **kwargs: object) -> RunOutcome:
raise self._exc
@pytest.mark.parametrize(
"argv, expected",
[
(["run", "--help"], 0),
(["run", "Review this repository"], 0),
],
)
def test_main_supports_help_and_execution(
argv: list[str], expected: int, capsys: pytest.CaptureFixture[str]
) -> None:
code = main(argv)
captured = capsys.readouterr()
assert code == expected
if argv[-1] == "--help":
assert "usage:" in captured.out.lower()
else:
assert "Run ID:" in captured.out
def test_run_command_success_with_text_output(capsys: pytest.CaptureFixture[str]) -> None:
runtime = SuccessfulRuntime()
code = run_command(_args(), runtime=runtime)
captured = capsys.readouterr()
assert code == 0
assert "Run ID: run_123" in captured.out
assert "Status: completed" in captured.out
assert "Result:" in captured.out
assert captured.err == ""
assert runtime.calls[0]["run_id"] is not None
assert str(runtime.calls[0]["run_id"]).startswith("run_") is False
def test_run_command_rejects_empty_task(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(task=" "), runtime=AegisRuntime())
captured = capsys.readouterr()
assert code == 2
assert "Task cannot be empty" in captured.err
assert "Traceback" not in captured.err
def test_run_command_unknown_agent(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(agent="missing"), runtime=AegisRuntime())
captured = capsys.readouterr()
assert code == 3
assert "Unknown agent: missing" in captured.err
assert "Traceback" not in captured.err
def test_run_command_unknown_workflow(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(workflow="missing"), runtime=AegisRuntime())
captured = capsys.readouterr()
assert code == 3
assert "Unknown workflow: missing" in captured.err
assert "Traceback" not in captured.err
def test_run_command_runtime_exception(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(), runtime=RaisingRuntime(RuntimeError("boom")))
captured = capsys.readouterr()
assert code == 1
assert "Unexpected internal error." in captured.err
assert "Traceback" not in captured.err
def test_run_command_evaluator_failure(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(), runtime=RaisingRuntime(EvaluationFailureError("evaluation failed")))
captured = capsys.readouterr()
assert code == 6
assert "evaluation failed" in captured.err
def test_run_command_json_output_success(capsys: pytest.CaptureFixture[str]) -> None:
runtime = SuccessfulRuntime()
code = run_command(_args(output="json"), runtime=runtime)
captured = capsys.readouterr()
payload = json.loads(captured.out)
assert code == 0
assert payload["status"] == "completed"
assert payload["agent"] == "default-task"
assert payload["workflow"] == "default"
assert payload["error"] is None
def test_run_command_json_output_failure(capsys: pytest.CaptureFixture[str]) -> None:
code = run_command(_args(output="json", agent="missing"), runtime=AegisRuntime())
captured = capsys.readouterr()
payload = json.loads(captured.out)
assert code == 3
assert payload["status"] == "failed"
assert payload["error"] == "Unknown agent: missing"
assert payload["result"] is None
def test_run_command_max_steps_forwarded(capsys: pytest.CaptureFixture[str]) -> None:
runtime = SuccessfulRuntime()
code = run_command(_args(max_steps=10), runtime=runtime)
capsys.readouterr()
assert code == 0
assert runtime.calls[0]["max_steps"] == 10
def test_run_command_verbose_failure_has_no_stack_trace_for_known_error(
capsys: pytest.CaptureFixture[str]
) -> None:
code = run_command(_args(verbose=True, agent="missing"), runtime=AegisRuntime())
captured = capsys.readouterr()
assert code == 3
assert "Traceback" not in captured.err
assert "Unknown agent: missing" in captured.err
|