574 lines
23 KiB
Python
574 lines
23 KiB
Python
"""
|
||
Tests for CodingAgent (backend/agent/coding_agent.py)
|
||
|
||
Structure:
|
||
TestHelpers – truncate_result, trim_messages, _strip_code_fences
|
||
TestDispatcher – dispatch_tool routing
|
||
TestTools – tool functions (read_file, write_file, …) using tmp workspace
|
||
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
|
||
"""
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
||
from backend.agent.coding_agent import (
|
||
MAX_HISTORY_CHARS,
|
||
MAX_RESULT_LENGTH,
|
||
CodingAgent,
|
||
_strip_code_fences,
|
||
dispatch_tool,
|
||
done,
|
||
grep_search,
|
||
list_files,
|
||
read_file,
|
||
run_python,
|
||
truncate_result,
|
||
trim_messages,
|
||
validate_python,
|
||
write_file,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 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)
|
||
# Must be significantly shorter than the original
|
||
# (slightly above MAX_HISTORY_CHARS is acceptable due to the injected reminder message)
|
||
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"
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestDispatcher
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestDispatcher:
|
||
|
||
def test_unknown_tool_returns_error(self):
|
||
result = dispatch_tool("nonexistent_tool", {})
|
||
assert "ERROR" in result
|
||
assert "nonexistent_tool" in result
|
||
|
||
def test_done_tool_dispatched(self):
|
||
result = dispatch_tool("done", {"summary": "finished"})
|
||
assert "finished" in result
|
||
|
||
def test_wrong_arguments_returns_error(self):
|
||
result = dispatch_tool("read_file", {"wrong_param": "x"})
|
||
assert "ERROR" in result
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestTools (patched WORKSPACE → tmp_path)
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
class TestWriteFile:
|
||
|
||
def test_write_creates_file(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = write_file("hello.py", "print('hi')")
|
||
assert result.startswith("OK:")
|
||
assert (tmp_path / "hello.py").read_text() == "print('hi')"
|
||
|
||
def test_write_outside_workspace_blocked(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = write_file("../evil.py", "bad")
|
||
assert "ERROR" in result
|
||
|
||
def test_write_unsupported_extension_blocked(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = write_file("script.sh", "echo hi")
|
||
assert "ERROR" in result
|
||
|
||
|
||
class TestReadFile:
|
||
|
||
def test_read_existing_file(self, tmp_path):
|
||
(tmp_path / "data.txt").write_text("hello world")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = read_file("data.txt")
|
||
assert result == "hello world"
|
||
|
||
def test_read_nonexistent_file(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = read_file("ghost.py")
|
||
assert "ERROR" in result
|
||
|
||
def test_read_outside_workspace_blocked(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = read_file("../secret.py")
|
||
assert "ERROR" in result
|
||
|
||
def test_read_unsupported_extension(self, tmp_path):
|
||
(tmp_path / "data.csv").write_text("a,b")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = read_file("data.csv")
|
||
assert "ERROR" in result
|
||
|
||
|
||
class TestListFiles:
|
||
|
||
def test_empty_workspace(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = list_files()
|
||
assert "No files" in result
|
||
|
||
def test_lists_existing_files(self, tmp_path):
|
||
(tmp_path / "a.py").touch()
|
||
(tmp_path / "b.txt").touch()
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = list_files()
|
||
assert "a.py" in result
|
||
assert "b.txt" in result
|
||
|
||
def test_glob_filter(self, tmp_path):
|
||
(tmp_path / "a.py").touch()
|
||
(tmp_path / "b.txt").touch()
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = list_files("*.py")
|
||
assert "a.py" in result
|
||
assert "b.txt" not in result
|
||
|
||
|
||
class TestGrepSearch:
|
||
|
||
def test_finds_pattern(self, tmp_path):
|
||
(tmp_path / "code.py").write_text("def hello():\n pass\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = grep_search("def hello")
|
||
assert "code.py" in result
|
||
assert "def hello" in result
|
||
|
||
def test_no_match_returns_message(self, tmp_path):
|
||
(tmp_path / "code.py").write_text("x = 1\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = grep_search("nonexistent_pattern")
|
||
assert "No matches" in result
|
||
|
||
def test_returns_line_number(self, tmp_path):
|
||
(tmp_path / "code.py").write_text("x = 1\ndef foo():\n pass\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = grep_search("def foo")
|
||
assert ":2:" in result
|
||
|
||
|
||
class TestValidatePython:
|
||
|
||
def test_valid_syntax(self, tmp_path):
|
||
(tmp_path / "good.py").write_text("def f(x):\n return x * 2\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = validate_python("good.py")
|
||
assert result == "OK: syntax is valid."
|
||
|
||
def test_invalid_syntax(self, tmp_path):
|
||
(tmp_path / "bad.py").write_text("def f(x)\n return x\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = validate_python("bad.py")
|
||
assert "SYNTAX ERROR" in result
|
||
|
||
def test_file_not_found(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = validate_python("ghost.py")
|
||
assert "ERROR" in result
|
||
|
||
|
||
class TestRunPython:
|
||
|
||
def test_successful_execution(self, tmp_path):
|
||
(tmp_path / "hello.py").write_text("print('hello world')\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = run_python("hello.py")
|
||
assert "hello world" in result
|
||
assert "Exit code: 0" in result
|
||
|
||
def test_runtime_error_captured(self, tmp_path):
|
||
(tmp_path / "bad.py").write_text("raise ValueError('oops')\n")
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = run_python("bad.py")
|
||
assert "ValueError" in result
|
||
assert "Exit code: 1" in result
|
||
|
||
def test_file_not_found(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
result = run_python("ghost.py")
|
||
assert "ERROR" in result
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# 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)
|
||
|
||
def test_returns_dict_with_required_keys(self, agent):
|
||
self._mock_api(agent)
|
||
action = agent.propose_next_action()
|
||
assert "thought" in action
|
||
assert "tool" in action
|
||
assert "arguments" in action
|
||
|
||
def test_increments_iteration(self, agent):
|
||
self._mock_api(agent)
|
||
agent.propose_next_action()
|
||
assert agent.iteration == 1
|
||
|
||
def test_stores_pending_action(self, agent):
|
||
self._mock_api(agent)
|
||
agent.propose_next_action()
|
||
assert agent.pending_action is not None
|
||
|
||
def test_returns_correct_tool(self, agent):
|
||
self._mock_api(agent, tool="list_files")
|
||
action = agent.propose_next_action()
|
||
assert action["tool"] == "list_files"
|
||
|
||
def test_handles_json_parse_error_gracefully(self, agent):
|
||
agent._call_api = MagicMock(return_value="this is not json {{")
|
||
action = agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
def test_handles_api_exception_gracefully(self, agent):
|
||
agent._call_api = MagicMock(side_effect=Exception("connection refused"))
|
||
action = agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
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 = agent.propose_next_action()
|
||
assert action["tool"] == "list_files"
|
||
|
||
def test_already_done_returns_done_action(self, agent):
|
||
agent.is_done = True
|
||
action = agent.propose_next_action()
|
||
assert action["tool"] == "done"
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# TestApprove (mocked API + real tool execution via tmp_path)
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
|
||
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},
|
||
}
|
||
|
||
def test_approve_without_pending_raises(self, agent):
|
||
with pytest.raises(Exception):
|
||
agent.approve()
|
||
|
||
def test_approve_done_sets_is_done(self, agent):
|
||
self._set_pending(agent, "done", summary="all done")
|
||
result = agent.approve()
|
||
assert result["is_done"] is True
|
||
assert agent.is_done is True
|
||
|
||
def test_approve_done_returns_summary(self, agent):
|
||
self._set_pending(agent, "done", summary="finished successfully")
|
||
result = agent.approve()
|
||
assert "finished successfully" in result["result"]
|
||
|
||
def test_approve_clears_pending_action(self, agent):
|
||
self._set_pending(agent, "done", summary="x")
|
||
agent.approve()
|
||
assert agent.pending_action is None
|
||
|
||
def test_approve_appends_assistant_message(self, agent):
|
||
self._set_pending(agent, "done", summary="x")
|
||
before = len(agent.messages)
|
||
agent.approve()
|
||
assert len(agent.messages) > before
|
||
|
||
def test_approve_tool_result_appended_to_messages(self, agent, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
self._set_pending(agent, "list_files")
|
||
agent.approve()
|
||
tool_results = [m for m in agent.messages if "tool_result" in m["content"]]
|
||
assert len(tool_results) == 1
|
||
|
||
def test_approve_error_result_adds_replan_tag(self, agent, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
self._set_pending(agent, "read_file", path="nonexistent.py")
|
||
agent.approve()
|
||
last_msg = agent.messages[-1]["content"]
|
||
assert "replan" in last_msg
|
||
|
||
def test_approve_returns_tool_name_in_result(self, agent, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
self._set_pending(agent, "list_files")
|
||
result = 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") # should not raise
|
||
|
||
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("*")) # no files created
|
||
|
||
|
||
# ═════════════════════════════════════════════════════════════════════════════
|
||
# 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 # safety limit for the test loop
|
||
|
||
def _run_until_done(self, agent) -> list:
|
||
"""Drive the agent loop until done or MAX_STEPS reached."""
|
||
steps = []
|
||
for _ in range(self.MAX_STEPS):
|
||
action = agent.propose_next_action()
|
||
result = agent.approve()
|
||
steps.append(result)
|
||
if result["is_done"]:
|
||
break
|
||
return steps
|
||
|
||
def test_agent_completes_hello_world_task(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
agent = CodingAgent()
|
||
try:
|
||
agent.start_task(
|
||
"Write a Python file called hello.py that prints 'Hello World'. "
|
||
"Validate it and run it."
|
||
)
|
||
steps = self._run_until_done(agent)
|
||
except Exception as e:
|
||
pytest.skip(f"API not reachable: {e}")
|
||
|
||
assert agent.is_done, "Agent did not reach done state"
|
||
tools_used = [s["tool"] for s in steps]
|
||
assert "write_file" in tools_used
|
||
assert "done" in tools_used
|
||
|
||
def test_agent_creates_file_on_disk(self, tmp_path):
|
||
with patch("backend.agent.coding_agent.WORKSPACE", tmp_path):
|
||
agent = CodingAgent()
|
||
try:
|
||
agent.start_task("Write a file called output.txt containing the text 'test passed'.")
|
||
self._run_until_done(agent)
|
||
except Exception as e:
|
||
pytest.skip(f"API not reachable: {e}")
|
||
|
||
py_files = list(tmp_path.glob("*.txt")) + list(tmp_path.glob("*.py"))
|
||
assert len(py_files) > 0, "Agent did not create any file"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v", "-s"])
|