Setting the file. One moment.
Run Scenarios · N8n:testmax N8n AI · n8n-io/n8n · Skills Docs
Back to the top of the page
Python scripts/ run_scenarios.py
Python · 172 lines · 7 KB
"""
15
16 import json
17 import shutil
18 import subprocess
19 import sys
20 import urllib.request
21 from pathlib import Path
22
23 from n8n_api import N8nClient, load_token
24
25 HERE = Path( __file__ ).resolve().parent
26 TIMEOUT_SECONDS = 90
27 ITEMS_PER_NODE = 3
28
29
30 class Mcp :
31 def __init__ (self, base_url, token):
32 self .url, self .token = base_url + '/mcp-server/http' , token
33
34 def call (self, tool, arguments):
35 body = json.dumps({ 'jsonrpc' : '2.0' , 'id' : 1 , 'method' : 'tools/call' ,
36 'params' : { 'name' : tool, 'arguments' : arguments}}).encode()
37 req = urllib.request.Request( self .url, data = body, method = 'POST' , headers = {
38 'content-type' : 'application/json' ,
39 'accept' : 'application/json, text/event-stream' ,
40 'authorization' : f 'Bearer { self .token } ' ,
41 })
42 with urllib.request.urlopen(req, timeout = TIMEOUT_SECONDS + 60 ) as res:
43 raw = res.read().decode()
44 lines = [line[ 5 :].strip() for line in raw.splitlines() if line.startswith( 'data:' )]
45 reply = json.loads(lines[ - 1 ] if lines else raw)
46 if 'error' in reply:
47 raise RuntimeError ( f ' { tool } : { reply[ "error" ].get( "message" ) } ' )
48 result = reply[ 'result' ]
49 return result.get( 'structuredContent' ) or json.loads(result[ 'content' ][ 0 ][ 'text' ])
50
51
52 def matches (node_type, pattern):
53 short = node_type.rsplit( '.' , 1 )[ - 1 ]
54 return node_type == pattern or short.lower() == pattern.lower()
55
56
57 def read_run (run_data, types):
58 """Per node: did it run, its error, and its first output items (all outputs)."""
59 nodes = {}
60 for name, runs in (run_data or {}).items():
61 run = runs[ - 1 ] if runs else {}
62 items = []
63 for output in ((run.get( 'data' ) or {}).get( 'main' ) or []):
64 items.extend(item.get( 'json' ) for item in (output or [])[: ITEMS_PER_NODE ])
65 error = run.get( 'error' )
66 nodes[name] = {
67 'type' : types.get(name),
68 'status' : run.get( 'executionStatus' ),
69 'error' : (error.get( 'message' ) or error.get( 'description' )) if isinstance (error, dict ) else error,
70 'items' : items[: ITEMS_PER_NODE * 2 ],
71 }
72 return nodes
73
74
75 def expectations (expect, nodes):
76 """Check reach and skip: node types that must run, and node types that must not run."""
77 results = []
78 ran = [n for n in nodes.values() if n[ 'status' ] == 'success' and not n[ 'error' ]]
79 for pattern in expect.get( 'reach' ) or []:
80 results.append({ 'kind' : 'reach' , 'type' : pattern, 'ok' : any (matches(n[ 'type' ] or '' , pattern) for n in ran)})
81 for pattern in expect.get( 'skip' ) or []:
82 results.append({ 'kind' : 'skip' , 'type' : pattern,
83 'ok' : not any (matches(n[ 'type' ] or '' , pattern) for n in nodes.values())})
84 return results
85
86
87 def set_mcp_access (client, workflow_id, on):
88 """Set MCP access for the workflow. Returns True when the script changed the setting."""
89 wf = client.workflow(workflow_id)
90 settings = wf.get( 'settings' ) or {}
91 if bool (settings.get( 'availableInMCP' )) == on:
92 return False
93 client.request( 'PATCH' , f '/rest/workflows/ { workflow_id } ' ,
94 { 'versionId' : wf[ 'versionId' ], 'settings' : { ** settings, 'availableInMCP' : on}})
95 return True
96
97
98 def run_builder (sub, scenarios_file, client, mcp, node):
99 meta = json.loads((sub / 'run.json' ).read_text())
100 workflow_id = meta.get( 'workflowId' )
101 wf_file = sub / 'workflow.json'
102 out = { 'workflowId' : workflow_id, 'scenarios' : []}
103 if not workflow_id or not wf_file.exists():
104 out[ 'skipped' ] = 'No workflow.'
105 return out
106 workflow = json.loads(wf_file.read_text())
107 types = {n[ 'name' ]: n[ 'type' ] for n in workflow.get( 'nodes' , [])}
108 cmd = [node, str ( HERE / 'sdk.mjs' ), 'pin-plan' , str (wf_file)] + ([ str (scenarios_file)] if scenarios_file else [])
109 plan = json.loads(subprocess.run(cmd, cwd = HERE , capture_output = True , text = True , check = True ).stdout)
110 out.update( pinned = plan[ 'pinned' ], executed = plan[ 'executed' ])
111 try :
112 out[ 'mcpAccessTurnedOn' ] = set_mcp_access(client, workflow_id, True )
113 except Exception as e:
114 out[ 'skipped' ] = f 'Could not turn on MCP access: { e } '
115 return out
116 try :
117 run_plan(plan, workflow_id, types, mcp, sub, out)
118 finally :
119 if out[ 'mcpAccessTurnedOn' ]:
120 try :
121 set_mcp_access(client, workflow_id, False )
122 except Exception as e:
123 out[ 'mcpAccessRestoreError' ] = str (e)
124 print ( f '[scenarios] { sub.name } : could not turn MCP access off again: { e } ' , flush = True )
125 return out
126
127
128 def run_plan (plan, workflow_id, types, mcp, sub, out):
129 for scenario in plan[ 'scenarios' ]:
130 entry = { 'name' : scenario[ 'name' ], 'expect' : scenario[ 'expect' ], 'provenance' : scenario[ 'provenance' ]}
131 args = { 'workflowId' : workflow_id, 'pinData' : scenario[ 'pinData' ], 'timeout' : TIMEOUT_SECONDS }
132 if scenario.get( 'triggerNode' ):
133 args[ 'triggerNodeName' ] = entry[ 'triggerNode' ] = scenario[ 'triggerNode' ]
134 try :
135 result = mcp.call( 'test_workflow' , args)
136 entry.update( status = result.get( 'status' ), error = result.get( 'error' ), executionId = result.get( 'executionId' ))
137 if result.get( 'executionId' ):
138 execution = mcp.call( 'get_workflow_execution' , { 'workflowId' : workflow_id,
139 'executionId' : result[ 'executionId' ], 'includeData' : True })
140 run_data = ((execution.get( 'data' ) or {}).get( 'resultData' ) or {}).get( 'runData' )
141 entry[ 'nodes' ] = read_run(run_data, types)
142 entry[ 'checks' ] = expectations(scenario[ 'expect' ], entry[ 'nodes' ])
143 except Exception as e:
144 entry.update( status = 'error' , error = f ' { type (e). __name__ } : { e } ' )
145 print ( f '[scenarios] { sub.name } : { scenario[ "name" ] } -> { entry.get( "status" ) } '
146 + ( f ' ( { entry[ "error" ] } )' if entry.get( 'error' ) else '' ), flush = True )
147 out[ 'scenarios' ].append(entry)
148
149
150 def main ():
151 run_dir = Path(sys.argv[ 1 ] if len (sys.argv) > 1 else '.' ).resolve()
152 node = shutil.which( 'node' )
153 token = load_token()
154 if not node or not token:
155 sys.exit( 'Scenarios need node on PATH and a working MCP token.' )
156 client = N8nClient().login()
157 mcp = Mcp(client.base_url, token)
158 scenarios_file = run_dir / 'scenarios.json'
159 scenarios_file = scenarios_file if scenarios_file.exists() else None
160 for meta_file in sorted (run_dir.glob( '*/run.json' )):
161 sub = meta_file.parent
162 try :
163 out = run_builder(sub, scenarios_file, client, mcp, node)
164 except Exception as e:
165 out = { 'skipped' : f ' { type (e). __name__ } : { e } ' , 'scenarios' : []}
166 (sub / 'exec.json' ).write_text(json.dumps(out, indent = 1 ))
167 if out.get( 'skipped' ):
168 print ( f '[scenarios] { sub.name } : skipped, { out[ "skipped" ] } ' , flush = True )
169
170
171 if __name__ == '__main__' :
172 main()