435 lines
18 KiB
Python
435 lines
18 KiB
Python
"""
|
||
Tests for CodingAgent (backend/agent/coding_agent.py)
|
||
|
||
Structure:
|
||
TestHelpers – truncate_result, trim_messages, _strip_code_fences
|
||
TestCodingAgentInit – __init__ and start_task
|
||
TestProposeNextAction – propose_next_action with mocked API
|
||
TestApprove – approve with mocked API + real tool execution
|
||
TestReject – reject injects feedback correctly
|
||
TestFullLoop – integration: real API, skipped if unreachable
|
||
|
||
Note: TestTools (write_file, read_file, etc.) and TestDispatcher were removed
|
||
because those tool functions are now MCP server tools, not standalone functions
|
||
in coding_agent.py. They will be tested via test_mcp_server_*.py once the MCP
|
||
servers are finalised.
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
||
from backend.agent.coding_agent import (
|
||
MAX_HISTORY_CHARS,
|
||
MAX_ITERATIONS,
|
||
MAX_RESULT_LENGTH,
|
||
CodingAgent,
|
||
_strip_code_fences,
|
||
truncate_result,
|
||
trim_messages,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _make_api_response(content: str, status_code: int = 200):
|
||
"""Build a mock requests.Response that returns *content* as the AI message."""
|
||
mock = MagicMock()
|
||
mock.status_code = status_code
|
||
mock.json.return_value = {
|
||
"choices": [{"message": {"role": "assistant", "content": content}}]
|
||
}
|
||
mock.text = "error body"
|
||
return mock
|
||
|
||
|
||
def _agent_action_json(tool: str, thought: str = "thinking...", **arguments) -> str:
|
||
return json.dumps({"thought": thought, "tool": tool, "arguments": arguments})
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestHelpers
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestTruncateResult:
|
||
|
||
def test_short_result_unchanged(self):
|
||
assert truncate_result("hello") == "hello"
|
||
|
||
def test_long_result_is_truncated(self):
|
||
long = "x" * (MAX_RESULT_LENGTH + 100)
|
||
result = truncate_result(long)
|
||
assert len(result) < len(long)
|
||
assert "TRUNCATED" in result
|
||
|
||
def test_exact_limit_not_truncated(self):
|
||
text = "a" * MAX_RESULT_LENGTH
|
||
assert truncate_result(text) == text
|
||
|
||
def test_truncated_keeps_start_and_end(self):
|
||
text = "START" + "x" * MAX_RESULT_LENGTH + "END"
|
||
result = truncate_result(text)
|
||
assert "START" in result
|
||
assert "END" in result
|
||
|
||
|
||
class TestTrimMessages:
|
||
|
||
def _make_messages(self, n_extra: int, chars_each: int = 100) -> list:
|
||
msgs = [
|
||
{"role": "system", "content": "sys"},
|
||
{"role": "user", "content": "original task"},
|
||
]
|
||
for i in range(n_extra):
|
||
msgs.append({"role": "assistant", "content": "x" * chars_each})
|
||
msgs.append({"role": "user", "content": "y" * chars_each})
|
||
return msgs
|
||
|
||
def test_short_history_unchanged(self):
|
||
msgs = self._make_messages(2)
|
||
assert trim_messages(msgs) == msgs
|
||
|
||
def test_long_history_is_trimmed(self):
|
||
msgs = self._make_messages(n_extra=500, chars_each=200)
|
||
original_total = sum(len(m["content"]) for m in msgs)
|
||
trimmed = trim_messages(msgs)
|
||
trimmed_total = sum(len(m["content"]) for m in trimmed)
|
||
assert trimmed_total < original_total
|
||
assert len(trimmed) < len(msgs)
|
||
|
||
def test_system_message_always_kept(self):
|
||
msgs = self._make_messages(n_extra=500, chars_each=200)
|
||
trimmed = trim_messages(msgs)
|
||
assert trimmed[0]["role"] == "system"
|
||
|
||
def test_original_task_always_kept(self):
|
||
msgs = self._make_messages(n_extra=500, chars_each=200)
|
||
trimmed = trim_messages(msgs)
|
||
assert trimmed[1]["content"] == "original task"
|
||
|
||
def test_reminder_injected_when_trimmed(self):
|
||
msgs = self._make_messages(n_extra=500, chars_each=200)
|
||
trimmed = trim_messages(msgs)
|
||
contents = [m["content"] for m in trimmed]
|
||
assert any("system_note" in c for c in contents)
|
||
|
||
|
||
class TestStripCodeFences:
|
||
|
||
def test_plain_text_unchanged(self):
|
||
assert _strip_code_fences("hello") == "hello"
|
||
|
||
def test_removes_json_fence(self):
|
||
text = "```json\n{\"key\": 1}\n```"
|
||
assert _strip_code_fences(text) == '{"key": 1}'
|
||
|
||
def test_removes_plain_fence(self):
|
||
text = "```\nhello\n```"
|
||
assert _strip_code_fences(text) == "hello"
|
||
|
||
def test_strips_whitespace(self):
|
||
assert _strip_code_fences(" hello ") == "hello"
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestCodingAgentInit
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestCodingAgentInit:
|
||
|
||
def test_initial_state_is_clean(self):
|
||
agent = CodingAgent()
|
||
assert agent.messages == []
|
||
assert agent.pending_action is None
|
||
assert agent.is_done is False
|
||
assert agent.iteration == 0
|
||
|
||
def test_api_url_is_set(self):
|
||
agent = CodingAgent()
|
||
assert agent.api_url.startswith("http://")
|
||
assert "/v1/chat/completions" in agent.api_url
|
||
|
||
def test_start_task_sets_messages(self):
|
||
agent = CodingAgent()
|
||
agent.start_task("Write fibonacci.py")
|
||
assert len(agent.messages) == 2
|
||
assert agent.messages[0]["role"] == "system"
|
||
assert agent.messages[1]["role"] == "user"
|
||
assert "fibonacci" in agent.messages[1]["content"]
|
||
|
||
def test_start_task_resets_state(self):
|
||
agent = CodingAgent()
|
||
agent.is_done = True
|
||
agent.iteration = 5
|
||
agent.start_task("New task")
|
||
assert agent.is_done is False
|
||
assert agent.iteration == 0
|
||
assert agent.pending_action is None
|
||
|
||
def test_start_task_twice_resets_messages(self):
|
||
agent = CodingAgent()
|
||
agent.start_task("First task")
|
||
agent.start_task("Second task")
|
||
assert "Second task" in agent.messages[1]["content"]
|
||
assert len(agent.messages) == 2
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestProposeNextAction (mocked API)
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestProposeNextAction:
|
||
|
||
@pytest.fixture
|
||
def agent(self):
|
||
a = CodingAgent()
|
||
a.start_task("Write hello.py")
|
||
return a
|
||
|
||
def _mock_api(self, agent, tool="list_files", thought="planning", **args):
|
||
payload = _agent_action_json(tool, thought, **args)
|
||
agent._call_api = MagicMock(return_value=payload)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_dict_with_required_keys(self, agent):
|
||
self._mock_api(agent)
|
||
action = await agent.propose_next_action()
|
||
assert "thought" in action
|
||
assert "tool" in action
|
||
assert "arguments" in action
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_increments_iteration(self, agent):
|
||
self._mock_api(agent)
|
||
await agent.propose_next_action()
|
||
assert agent.iteration == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_stores_pending_action(self, agent):
|
||
self._mock_api(agent)
|
||
await agent.propose_next_action()
|
||
assert agent.pending_action is not None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_correct_tool(self, agent):
|
||
self._mock_api(agent, tool="list_files")
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "list_files"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handles_json_parse_error_gracefully(self, agent):
|
||
agent._call_api = MagicMock(return_value="this is not json {{")
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_handles_api_exception_gracefully(self, agent):
|
||
agent._call_api = MagicMock(side_effect=Exception("connection refused"))
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_strips_code_fences_from_response(self, agent):
|
||
payload = "```json\n" + _agent_action_json("list_files", "thinking") + "\n```"
|
||
agent._call_api = MagicMock(return_value=payload)
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "list_files"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_already_done_returns_done_action(self, agent):
|
||
agent.is_done = True
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_max_iterations_returns_done_without_api_call(self, agent):
|
||
agent.iteration = MAX_ITERATIONS
|
||
agent._call_api = MagicMock(side_effect=AssertionError("API must not be called"))
|
||
action = await agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
agent._call_api.assert_not_called()
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestApprove (mocked API + mocked dispatch_tool)
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestApprove:
|
||
|
||
@pytest.fixture
|
||
def agent(self):
|
||
a = CodingAgent()
|
||
a.start_task("Do something")
|
||
return a
|
||
|
||
def _set_pending(self, agent, tool: str, **arguments):
|
||
raw = _agent_action_json(tool, "thought", **arguments)
|
||
agent.pending_action = {
|
||
"raw": raw,
|
||
"action": {"thought": "thought", "tool": tool, "arguments": arguments},
|
||
}
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_without_pending_raises(self, agent):
|
||
with pytest.raises(Exception):
|
||
await agent.approve()
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_done_sets_is_done(self, agent):
|
||
self._set_pending(agent, "done", summary="all done")
|
||
result = await agent.approve()
|
||
assert result["is_done"] is True
|
||
assert agent.is_done is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_done_returns_summary(self, agent):
|
||
self._set_pending(agent, "done", summary="finished successfully")
|
||
result = await agent.approve()
|
||
assert "finished successfully" in result["result"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_clears_pending_action(self, agent):
|
||
self._set_pending(agent, "done", summary="x")
|
||
await agent.approve()
|
||
assert agent.pending_action is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_appends_assistant_message(self, agent):
|
||
self._set_pending(agent, "done", summary="x")
|
||
before = len(agent.messages)
|
||
await agent.approve()
|
||
assert len(agent.messages) > before
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_tool_result_appended_to_messages(self, agent):
|
||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="file list"):
|
||
self._set_pending(agent, "list_files")
|
||
await agent.approve()
|
||
tool_results = [m for m in agent.messages if "tool_result" in m["content"]]
|
||
assert len(tool_results) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_error_result_adds_replan_tag(self, agent):
|
||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="ERROR: file not found"):
|
||
self._set_pending(agent, "read_file", path="nonexistent.py")
|
||
await agent.approve()
|
||
last_msg = agent.messages[-1]["content"]
|
||
assert "replan" in last_msg
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_approve_returns_tool_name_in_result(self, agent):
|
||
with patch("backend.agent.coding_agent.dispatch_tool", return_value="(empty)"):
|
||
self._set_pending(agent, "list_files")
|
||
result = await agent.approve()
|
||
assert result["tool"] == "list_files"
|
||
assert result["is_done"] is False
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestReject
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestReject:
|
||
|
||
@pytest.fixture
|
||
def agent(self):
|
||
a = CodingAgent()
|
||
a.start_task("Do something")
|
||
return a
|
||
|
||
def _set_pending(self, agent, tool="write_file"):
|
||
raw = _agent_action_json(tool, "thought")
|
||
agent.pending_action = {
|
||
"raw": raw,
|
||
"action": {"thought": "thought", "tool": tool, "arguments": {}},
|
||
}
|
||
|
||
def test_reject_clears_pending_action(self, agent):
|
||
self._set_pending(agent)
|
||
agent.reject("Use a different approach")
|
||
assert agent.pending_action is None
|
||
|
||
def test_reject_appends_human_message(self, agent):
|
||
self._set_pending(agent)
|
||
agent.reject("Do it differently")
|
||
user_msgs = [m for m in agent.messages if m["role"] == "user"]
|
||
assert any("Do it differently" in m["content"] for m in user_msgs)
|
||
|
||
def test_reject_adds_replan_tag(self, agent):
|
||
self._set_pending(agent)
|
||
agent.reject("feedback")
|
||
last = agent.messages[-1]["content"]
|
||
assert "replan" in last.lower()
|
||
|
||
def test_reject_without_pending_does_not_crash(self, agent):
|
||
agent.pending_action = None
|
||
agent.reject("no pending action")
|
||
|
||
def test_reject_does_not_execute_tool(self, agent, tmp_path):
|
||
self._set_pending(agent, "write_file")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
agent.reject("Do not write anything")
|
||
assert not list(tmp_path.glob("*"))
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestFullLoop (integration – real API, skipped if unreachable)
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestFullLoop:
|
||
"""End-to-end test: agent runs a real task against the live API.
|
||
Skipped automatically if the API is not reachable.
|
||
"""
|
||
|
||
MAX_STEPS = 15
|
||
|
||
async def _run_until_done(self, agent) -> list:
|
||
steps = []
|
||
for _ in range(self.MAX_STEPS):
|
||
action = await agent.propose_next_action()
|
||
result = await agent.approve()
|
||
steps.append(result)
|
||
if result["is_done"]:
|
||
break
|
||
return steps
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_completes_hello_world_task(self, tmp_path):
|
||
try:
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
agent = CodingAgent()
|
||
agent.start_task(
|
||
"Write a Python file called hello.py that prints 'Hello World'. "
|
||
"Validate it and run it."
|
||
)
|
||
steps = await self._run_until_done(agent)
|
||
assert agent.is_done, "Agent did not reach done state"
|
||
tools_used = [s["tool"] for s in steps]
|
||
assert "done" in tools_used
|
||
except Exception as e:
|
||
pytest.skip(f"API not reachable or environment incomplete: {e}")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_agent_creates_file_on_disk(self, tmp_path):
|
||
try:
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
agent = CodingAgent()
|
||
agent.start_task("Write a file called output.txt containing the text 'test passed'.")
|
||
await self._run_until_done(agent)
|
||
py_files = list(tmp_path.glob("*.txt")) + list(tmp_path.glob("*.py"))
|
||
assert len(py_files) > 0, "Agent did not create any file"
|
||
except Exception as e:
|
||
pytest.skip(f"API not reachable or environment incomplete: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v", "-s"])
|