4370be33b4
The previous commit (f98046c) added shutil.which() to resolve executables,
which broke two tests that only mocked subprocess execution. This commit
adds shutil.which() mocking to both test files to restore test compatibility.
Co-authored-by: Claude <noreply@anthropic.com>
76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
import asyncio
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from clink.agents.base import CLIAgentError
|
|
from clink.agents.codex import CodexAgent
|
|
from clink.models import ResolvedCLIClient, ResolvedCLIRole
|
|
|
|
|
|
class DummyProcess:
|
|
def __init__(self, *, stdout: bytes = b"", stderr: bytes = b"", returncode: int = 0):
|
|
self._stdout = stdout
|
|
self._stderr = stderr
|
|
self.returncode = returncode
|
|
|
|
async def communicate(self, _input):
|
|
return self._stdout, self._stderr
|
|
|
|
|
|
@pytest.fixture()
|
|
def codex_agent():
|
|
prompt_path = Path("systemprompts/clink/codex_default.txt").resolve()
|
|
role = ResolvedCLIRole(name="default", prompt_path=prompt_path, role_args=[])
|
|
client = ResolvedCLIClient(
|
|
name="codex",
|
|
executable=["codex"],
|
|
internal_args=["exec"],
|
|
config_args=["--json", "--dangerously-bypass-approvals-and-sandbox"],
|
|
env={},
|
|
timeout_seconds=30,
|
|
parser="codex_jsonl",
|
|
roles={"default": role},
|
|
output_to_file=None,
|
|
working_dir=None,
|
|
)
|
|
return CodexAgent(client), role
|
|
|
|
|
|
async def _run_agent_with_process(monkeypatch, agent, role, process):
|
|
async def fake_create_subprocess_exec(*_args, **_kwargs):
|
|
return process
|
|
|
|
def fake_which(executable_name):
|
|
return f"/usr/bin/{executable_name}"
|
|
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_create_subprocess_exec)
|
|
monkeypatch.setattr(shutil, "which", fake_which)
|
|
return await agent.run(role=role, prompt="do something", files=[], images=[])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_codex_agent_recovers_jsonl(monkeypatch, codex_agent):
|
|
agent, role = codex_agent
|
|
stdout = b"""
|
|
{"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"Hello from Codex"}}
|
|
{"type":"turn.completed","usage":{"input_tokens":10,"output_tokens":5}}
|
|
"""
|
|
process = DummyProcess(stdout=stdout, returncode=124)
|
|
result = await _run_agent_with_process(monkeypatch, agent, role, process)
|
|
|
|
assert result.returncode == 124
|
|
assert "Hello from Codex" in result.parsed.content
|
|
assert result.parsed.metadata["usage"]["output_tokens"] == 5
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_codex_agent_propagates_invalid_json(monkeypatch, codex_agent):
|
|
agent, role = codex_agent
|
|
stdout = b"not json"
|
|
process = DummyProcess(stdout=stdout, returncode=1)
|
|
|
|
with pytest.raises(CLIAgentError):
|
|
await _run_agent_with_process(monkeypatch, agent, role, process)
|