test_org_node_trigger.py
2.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
"""U-005 tests: c_org_node_trigger's subprocess wrapper, mocked at the
subprocess boundary (CLAUDE.md Testing constraint). Real end-to-end coverage
is in src/test_agent_e2e.py (RUN_REAL_AGENT_E2E=1).
Run: python3 -m unittest src.test_org_node_trigger -v
"""
import json
import subprocess
import unittest
from unittest.mock import patch
from src import org_node_trigger
def _completed_process(returncode=0, stdout_obj=None, stderr=""):
stdout = json.dumps(stdout_obj) if stdout_obj is not None else ""
return subprocess.CompletedProcess(args=["claude"], returncode=returncode, stdout=stdout, stderr=stderr)
class TriggerOrgNodeTest(unittest.TestCase):
def test_rejects_unknown_org_node_without_shelling_out(self):
with patch("src.org_node_trigger.subprocess.run") as run:
with self.assertRaises(ValueError):
org_node_trigger.trigger_org_node("not-a-real-node", "hi", "/tmp/some-project")
run.assert_not_called()
def test_happy_path_returns_result_text(self):
with patch("src.org_node_trigger.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": False, "result": "hello from planner"})
answer = org_node_trigger.trigger_org_node("planner", "say hi", "/tmp/some-project")
self.assertEqual(answer, "hello from planner")
def test_invokes_claude_with_expected_args_and_cwd(self):
with patch("src.org_node_trigger.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": False, "result": "ok"})
org_node_trigger.trigger_org_node("coder", "do the thing", "/tmp/some-project")
args, kwargs = run.call_args
argv = args[0]
self.assertEqual(argv[:3], ["claude", "--agent", "coder"])
self.assertIn("-p", argv)
self.assertIn("do the thing", argv)
self.assertIn("--permission-mode", argv)
self.assertIn("auto", argv)
self.assertNotIn("--bare", argv)
self.assertNotIn("--dangerously-skip-permissions", argv)
self.assertNotIn("bypassPermissions", argv)
self.assertEqual(kwargs["cwd"], "/tmp/some-project")
self.assertEqual(kwargs["stdin"], subprocess.DEVNULL)
def test_nonzero_exit_raises(self):
with patch("src.org_node_trigger.subprocess.run") as run:
run.return_value = _completed_process(returncode=1, stdout_obj=None, stderr="boom")
with self.assertRaises(org_node_trigger.OrgNodeTriggerError):
org_node_trigger.trigger_org_node("planner", "hi", "/tmp/some-project")
def test_is_error_true_raises(self):
with patch("src.org_node_trigger.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": True, "result": ""})
with self.assertRaises(org_node_trigger.OrgNodeTriggerError):
org_node_trigger.trigger_org_node("validator", "hi", "/tmp/some-project")
if __name__ == "__main__":
unittest.main()