test_work_graph_execute.py
3.97 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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""U-006 tests: c_work_graph_execute's dynamic-workflow wrapper, mocked at the
subprocess boundary (CLAUDE.md Testing constraint). Real end-to-end coverage
is in src/test_work_graph_execute_e2e.py (RUN_REAL_AGENT_E2E=1), reusing the
exact spike script the A-1 assumption was validated against.
Run: python3 -m unittest src.test_work_graph_execute -v
"""
import json
import os
import subprocess
import tempfile
import unittest
from unittest.mock import patch
from src import work_graph_execute
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 WriteWorkflowScriptTest(unittest.TestCase):
def test_writes_into_target_claude_workflows_not_graphlemon_repo(self):
with tempfile.TemporaryDirectory() as tmp:
path = work_graph_execute.write_workflow_script(tmp, "run-123", "export const meta = {}")
self.assertEqual(path, os.path.join(tmp, ".claude", "workflows", "run-123.js"))
with open(path, encoding="utf-8") as f:
self.assertEqual(f.read(), "export const meta = {}")
class ExecuteWorkGraphTest(unittest.TestCase):
SCHEMA = {"type": "object", "required": ["sum"], "properties": {"sum": {"type": "number"}}}
def test_happy_path_returns_structured_output(self):
with patch("src.work_graph_execute.subprocess.run") as run:
run.return_value = _completed_process(
stdout_obj={"is_error": False, "structured_output": {"sum": 29}}
)
output = work_graph_execute.execute_work_graph(
"/tmp/some-project", "run-123", {"numbers": [2, 3, 4]}, self.SCHEMA
)
self.assertEqual(output, {"sum": 29})
def test_invokes_claude_with_expected_args_and_cwd(self):
with patch("src.work_graph_execute.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": False, "structured_output": {"sum": 1}})
work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {"n": 1}, self.SCHEMA)
args, kwargs = run.call_args
argv = args[0]
self.assertEqual(argv[0], "claude")
self.assertIn("-p", argv)
self.assertIn("Run /run-123 with args {\"n\": 1}", argv)
self.assertIn("--json-schema", argv)
self.assertIn(json.dumps(self.SCHEMA), 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.work_graph_execute.subprocess.run") as run:
run.return_value = _completed_process(returncode=1, stdout_obj=None, stderr="boom")
with self.assertRaises(work_graph_execute.WorkGraphExecuteError):
work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA)
def test_is_error_true_raises(self):
with patch("src.work_graph_execute.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": True})
with self.assertRaises(work_graph_execute.WorkGraphExecuteError):
work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA)
def test_missing_structured_output_raises(self):
with patch("src.work_graph_execute.subprocess.run") as run:
run.return_value = _completed_process(stdout_obj={"is_error": False})
with self.assertRaises(work_graph_execute.WorkGraphExecuteError):
work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA)
if __name__ == "__main__":
unittest.main()