mirror of
https://github.com/codeflash-ai/codeflash-agent.git
synced 2026-05-04 18:25:19 +00:00
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""Tests for the run_agent CI runner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from github_app.agents import run_agent
|
|
|
|
PATCH_TARGET = "github_app.agents.asyncio.create_subprocess_exec"
|
|
|
|
|
|
def _mock_proc(stdout: bytes = b"output", stderr: bytes = b"", rc: int = 0):
|
|
proc = AsyncMock()
|
|
proc.communicate.return_value = (stdout, stderr)
|
|
proc.returncode = rc
|
|
return proc
|
|
|
|
|
|
async def test_run_agent_success(mock_config):
|
|
"""run_agent returns decoded stdout on success."""
|
|
proc = _mock_proc(stdout=b"Agent completed successfully.")
|
|
with patch(PATCH_TARGET, return_value=proc) as exec_mock:
|
|
result = await run_agent(
|
|
mock_config,
|
|
Path("/tmp/repo"),
|
|
"tok-123",
|
|
)
|
|
|
|
assert result == "Agent completed successfully."
|
|
# Verify env includes GITHUB_TOKEN and GH_TOKEN
|
|
call_kwargs = exec_mock.call_args.kwargs
|
|
assert call_kwargs["env"]["GITHUB_TOKEN"] == "tok-123"
|
|
assert call_kwargs["env"]["GH_TOKEN"] == "tok-123"
|
|
|
|
|
|
async def test_run_agent_passes_repo_dir_as_cwd(mock_config):
|
|
"""run_agent passes repo_dir through build_edit_cmd to cwd."""
|
|
proc = _mock_proc()
|
|
with patch(PATCH_TARGET, return_value=proc) as exec_mock:
|
|
await run_agent(mock_config, Path("/tmp/my-repo"), "tok")
|
|
|
|
call_kwargs = exec_mock.call_args.kwargs
|
|
assert call_kwargs["cwd"] == "/tmp/my-repo"
|
|
|
|
|
|
async def test_run_agent_failure_no_stderr_leak(mock_config):
|
|
"""RuntimeError must not contain stderr content (may have secrets)."""
|
|
proc = _mock_proc(
|
|
stdout=b"",
|
|
stderr=b"Error: token=ghp_secret123 expired",
|
|
rc=1,
|
|
)
|
|
with patch(PATCH_TARGET, return_value=proc):
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
await run_agent(mock_config, Path("/tmp/repo"), "tok")
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert "ghp_secret123" not in error_msg
|
|
assert "Agent" in error_msg
|
|
|
|
|
|
async def test_run_agent_timeout(mock_config):
|
|
"""run_agent raises TimeoutError when the process exceeds timeout."""
|
|
proc = AsyncMock()
|
|
proc.communicate = AsyncMock(side_effect=TimeoutError)
|
|
proc.kill = MagicMock()
|
|
|
|
with patch(PATCH_TARGET, return_value=proc):
|
|
with pytest.raises(TimeoutError, match="timed out"):
|
|
await run_agent(
|
|
mock_config,
|
|
Path("/tmp/repo"),
|
|
"tok",
|
|
timeout=1,
|
|
)
|
|
|
|
proc.kill.assert_called_once()
|
|
|
|
|
|
async def test_run_agent_uses_build_edit_cmd(mock_config):
|
|
"""run_agent calls build_edit_cmd (not build_cmd) for edit permissions."""
|
|
proc = _mock_proc()
|
|
with (
|
|
patch(PATCH_TARGET, return_value=proc),
|
|
patch("github_app.agents.get_backend") as get_backend_mock,
|
|
):
|
|
spec = MagicMock()
|
|
spec.build_edit_cmd.return_value = (["claude", "-p", "test"], "/tmp/repo")
|
|
get_backend_mock.return_value = spec
|
|
|
|
await run_agent(mock_config, Path("/tmp/repo"), "tok")
|
|
|
|
spec.build_edit_cmd.assert_called_once()
|
|
spec.build_cmd.assert_not_called()
|