| 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 |
|
|
