test_work_graph_execute_e2e.py
1.82 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
"""U-006's real-call coverage: reuses the exact script the A-1 spike validated
(spikes/a1-dynamic-workflows/workflow-under-test.js) through the new
write_workflow_script/execute_work_graph wrapper, not a re-typed copy --
"直接复用...已验证的代码路径" per development-plan.md U-006. Gated behind
RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing constraint).
Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_work_graph_execute_e2e -v
"""
import os
import subprocess
import tempfile
import unittest
from src import work_graph_execute
from src.test_support import skip_unless_env
_SPIKE_SCRIPT = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"spikes", "a1-dynamic-workflows", "workflow-under-test.js",
)
_SCHEMA = {
"type": "object",
"required": ["input_numbers", "squares", "sum"],
"properties": {
"input_numbers": {"type": "array"},
"squares": {"type": "array"},
"sum": {"type": "number"},
},
}
@skip_unless_env("RUN_REAL_AGENT_E2E")
class RealWorkGraphExecuteTest(unittest.TestCase):
def test_reused_a1_spike_script_runs_through_the_new_wrapper(self):
with tempfile.TemporaryDirectory() as tmp:
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
with open(_SPIKE_SCRIPT, encoding="utf-8") as f:
script_content = f.read()
work_graph_execute.write_workflow_script(tmp, "graphlemon-a1-spike", script_content)
output = work_graph_execute.execute_work_graph(
tmp, "graphlemon-a1-spike", {"numbers": [2, 3, 4]}, _SCHEMA, timeout=120
)
self.assertEqual(output["input_numbers"], [2, 3, 4])
self.assertEqual(output["squares"], [4, 9, 16])
self.assertEqual(output["sum"], 29)
if __name__ == "__main__":
unittest.main()