Thomas SharedInbox
a0e8b4359a
Add TestMain class covering the main() flow: asserts that _set_labels is called with State/InProgress (and State/Ready removed) strictly before _start_agent, and that no labels or agents are touched when there are no ready issues. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
208 lines
8.1 KiB
Python
208 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for agent_loop.py."""
|
|
import io
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
|
|
import agent_loop
|
|
|
|
|
|
class TestStateFile(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".json")
|
|
self._tmp.close()
|
|
self._orig = agent_loop.STATE_FILE
|
|
agent_loop.STATE_FILE = Path(self._tmp.name)
|
|
Path(self._tmp.name).unlink() # Start with no state file.
|
|
|
|
def tearDown(self):
|
|
agent_loop.STATE_FILE = self._orig
|
|
Path(self._tmp.name).unlink(missing_ok=True)
|
|
|
|
def test_write_state_stores_pid(self):
|
|
agent_loop._write_state(12345, 91, "issue")
|
|
data = json.loads(Path(self._tmp.name).read_text())
|
|
self.assertEqual(data["pid"], 12345)
|
|
self.assertNotIn("tmux_session", data)
|
|
|
|
def test_write_state_stores_issue_and_kind(self):
|
|
agent_loop._write_state(99, 7, "ci-fix")
|
|
data = json.loads(Path(self._tmp.name).read_text())
|
|
self.assertEqual(data["issue"], 7)
|
|
self.assertEqual(data["type"], "ci-fix")
|
|
self.assertIn("started_at", data)
|
|
|
|
def test_read_state_returns_none_when_missing(self):
|
|
self.assertIsNone(agent_loop._read_state())
|
|
|
|
def test_read_and_write_roundtrip(self):
|
|
agent_loop._write_state(42, 10, "issue")
|
|
state = agent_loop._read_state()
|
|
self.assertIsNotNone(state)
|
|
self.assertEqual(state["pid"], 42)
|
|
self.assertEqual(state["issue"], 10)
|
|
|
|
def test_clear_state_removes_file(self):
|
|
agent_loop._write_state(1, None, "ci-fix")
|
|
agent_loop._clear_state()
|
|
self.assertIsNone(agent_loop._read_state())
|
|
|
|
|
|
class TestAgentAlive(unittest.TestCase):
|
|
def test_own_pid_is_alive(self):
|
|
self.assertTrue(agent_loop._agent_alive({"pid": os.getpid()}))
|
|
|
|
def test_nonexistent_pid_is_dead(self):
|
|
self.assertFalse(agent_loop._agent_alive({"pid": 999999999}))
|
|
|
|
def test_missing_pid_returns_false(self):
|
|
self.assertFalse(agent_loop._agent_alive({}))
|
|
self.assertFalse(agent_loop._agent_alive({"pid": None}))
|
|
|
|
|
|
class TestKillAgent(unittest.TestCase):
|
|
def test_kill_sends_sigkill(self):
|
|
with patch("agent_loop.os.kill") as mock_kill:
|
|
agent_loop._kill_agent({"pid": 1234})
|
|
mock_kill.assert_called_once_with(1234, 9)
|
|
|
|
def test_kill_ignores_missing_process(self):
|
|
with patch("agent_loop.os.kill", side_effect=ProcessLookupError):
|
|
agent_loop._kill_agent({"pid": 1234}) # Should not raise.
|
|
|
|
def test_kill_noop_when_no_pid(self):
|
|
with patch("agent_loop.os.kill") as mock_kill:
|
|
agent_loop._kill_agent({})
|
|
mock_kill.assert_not_called()
|
|
|
|
|
|
class TestStartAgent(unittest.TestCase):
|
|
def _make_mock_proc(self, pid=42):
|
|
proc = MagicMock()
|
|
proc.pid = pid
|
|
proc.stdin = io.BytesIO()
|
|
return proc
|
|
|
|
def test_start_agent_returns_pid(self):
|
|
mock_proc = self._make_mock_proc(pid=42)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("agent_loop.subprocess.Popen", return_value=mock_proc):
|
|
with patch.object(agent_loop.Path, "home", return_value=Path(tmpdir)):
|
|
result = agent_loop._start_agent("do something", "issue-99")
|
|
self.assertEqual(result, 42)
|
|
|
|
def test_start_agent_uses_popen_not_tmux(self):
|
|
mock_proc = self._make_mock_proc(pid=7)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("agent_loop.subprocess.Popen", return_value=mock_proc) as mock_popen:
|
|
with patch("agent_loop.subprocess.run") as mock_run:
|
|
with patch.object(agent_loop.Path, "home", return_value=Path(tmpdir)):
|
|
agent_loop._start_agent("prompt", "ci-fix")
|
|
mock_popen.assert_called_once()
|
|
mock_run.assert_not_called()
|
|
|
|
def test_start_agent_passes_session_name_to_claude(self):
|
|
mock_proc = self._make_mock_proc(pid=7)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("agent_loop.subprocess.Popen", return_value=mock_proc) as mock_popen:
|
|
with patch.object(agent_loop.Path, "home", return_value=Path(tmpdir)):
|
|
agent_loop._start_agent("prompt", "issue-55")
|
|
cmd = mock_popen.call_args[0][0]
|
|
self.assertIn("issue-55", cmd)
|
|
self.assertIn("claude", cmd[0])
|
|
|
|
def test_start_agent_uses_start_new_session(self):
|
|
mock_proc = self._make_mock_proc(pid=7)
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
with patch("agent_loop.subprocess.Popen", return_value=mock_proc) as mock_popen:
|
|
with patch.object(agent_loop.Path, "home", return_value=Path(tmpdir)):
|
|
agent_loop._start_agent("prompt", "issue-55")
|
|
kwargs = mock_popen.call_args[1]
|
|
self.assertTrue(kwargs.get("start_new_session"))
|
|
|
|
|
|
class TestMain(unittest.TestCase):
|
|
"""Tests for the main() flow."""
|
|
|
|
def _make_mock_proc(self, pid=42):
|
|
proc = MagicMock()
|
|
proc.pid = pid
|
|
proc.stdin = io.BytesIO()
|
|
return proc
|
|
|
|
def _make_issue(self, number=10, title="Do something"):
|
|
return {"number": number, "title": title, "body": "", "labels": []}
|
|
|
|
def test_sets_in_progress_before_starting_agent(self):
|
|
"""_set_labels(InProgress) must be called before _start_agent."""
|
|
call_order = []
|
|
mock_proc = self._make_mock_proc(pid=55)
|
|
|
|
def fake_set_labels(issue, add, remove):
|
|
call_order.append(("set_labels", add, remove))
|
|
|
|
def fake_start_agent(prompt, session_name):
|
|
call_order.append(("start_agent", session_name))
|
|
return 55
|
|
|
|
with patch("agent_loop._read_state", return_value=None), \
|
|
patch("agent_loop._latest_ci_run", return_value=None), \
|
|
patch("agent_loop._ready_issues", return_value=[self._make_issue(10)]), \
|
|
patch("agent_loop._set_labels", side_effect=fake_set_labels), \
|
|
patch("agent_loop._start_agent", side_effect=fake_start_agent), \
|
|
patch("agent_loop._write_state"):
|
|
result = agent_loop.main()
|
|
|
|
self.assertEqual(result, 0)
|
|
labels_idx = next(
|
|
i for i, c in enumerate(call_order) if c[0] == "set_labels"
|
|
)
|
|
agent_idx = next(
|
|
i for i, c in enumerate(call_order) if c[0] == "start_agent"
|
|
)
|
|
self.assertLess(labels_idx, agent_idx,
|
|
"_set_labels must be called before _start_agent")
|
|
|
|
def test_sets_in_progress_label_and_removes_ready(self):
|
|
"""The InProgress label is added and the Ready label is removed."""
|
|
captured = {}
|
|
|
|
def fake_set_labels(issue, add, remove):
|
|
captured["add"] = add
|
|
captured["remove"] = remove
|
|
|
|
with patch("agent_loop._read_state", return_value=None), \
|
|
patch("agent_loop._latest_ci_run", return_value=None), \
|
|
patch("agent_loop._ready_issues", return_value=[self._make_issue(7)]), \
|
|
patch("agent_loop._set_labels", side_effect=fake_set_labels), \
|
|
patch("agent_loop._start_agent", return_value=99), \
|
|
patch("agent_loop._write_state"):
|
|
agent_loop.main()
|
|
|
|
self.assertIn(agent_loop.LABEL_IN_PROGRESS, captured.get("add", []))
|
|
self.assertIn(agent_loop.LABEL_READY, captured.get("remove", []))
|
|
|
|
def test_no_ready_issues_does_nothing(self):
|
|
"""main() exits cleanly with 0 when there are no ready issues."""
|
|
with patch("agent_loop._read_state", return_value=None), \
|
|
patch("agent_loop._latest_ci_run", return_value=None), \
|
|
patch("agent_loop._ready_issues", return_value=[]), \
|
|
patch("agent_loop._set_labels") as mock_labels, \
|
|
patch("agent_loop._start_agent") as mock_start:
|
|
result = agent_loop.main()
|
|
|
|
self.assertEqual(result, 0)
|
|
mock_labels.assert_not_called()
|
|
mock_start.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|