""" 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"])