112 lines
3.8 KiB
Python
Executable File
112 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REQUIRED_NODE_NAMES = {
|
|
'Webhook',
|
|
'route-action',
|
|
'Respond to Webhook',
|
|
}
|
|
|
|
EXPECTED_DIRECT_TYPES = {
|
|
'Webhook': 'n8n-nodes-base.webhook',
|
|
'route-action': 'n8n-nodes-base.code',
|
|
'Respond to Webhook': 'n8n-nodes-base.respondToWebhook',
|
|
}
|
|
|
|
|
|
def fail(msg: str):
|
|
print(f'ERROR: {msg}', file=sys.stderr)
|
|
raise SystemExit(1)
|
|
|
|
|
|
def load_json(path: Path):
|
|
try:
|
|
return json.loads(path.read_text())
|
|
except Exception as e:
|
|
fail(f'failed to parse {path}: {e}')
|
|
|
|
|
|
def main():
|
|
path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('assets/openclaw-action.workflow.json')
|
|
data = load_json(path)
|
|
|
|
if not isinstance(data, dict):
|
|
fail('workflow file must decode to a JSON object')
|
|
|
|
nodes = data.get('nodes')
|
|
connections = data.get('connections')
|
|
if not isinstance(nodes, list):
|
|
fail('workflow is missing a nodes array')
|
|
if not isinstance(connections, dict):
|
|
fail('workflow is missing a connections object')
|
|
|
|
by_name = {}
|
|
for node in nodes:
|
|
if not isinstance(node, dict):
|
|
fail('every node entry must be an object')
|
|
name = node.get('name')
|
|
if not name:
|
|
fail('every node must have a name')
|
|
by_name[name] = node
|
|
|
|
missing = sorted(REQUIRED_NODE_NAMES - set(by_name))
|
|
if missing:
|
|
fail(f'missing required nodes: {", ".join(missing)}')
|
|
|
|
for name, node_type in EXPECTED_DIRECT_TYPES.items():
|
|
actual = by_name[name].get('type')
|
|
if actual != node_type:
|
|
fail(f'node {name!r} should have type {node_type!r}, got {actual!r}')
|
|
|
|
webhook = by_name['Webhook']
|
|
webhook_params = webhook.get('parameters', {})
|
|
if webhook_params.get('path') != 'openclaw-action':
|
|
fail('Webhook.path must be openclaw-action')
|
|
if webhook_params.get('httpMethod') != 'POST':
|
|
fail('Webhook.httpMethod must be POST')
|
|
if webhook_params.get('responseMode') != 'responseNode':
|
|
fail('Webhook.responseMode must be responseNode')
|
|
|
|
router = by_name['route-action'].get('parameters', {})
|
|
if router.get('mode') != 'runOnceForEachItem':
|
|
fail('route-action code node must use runOnceForEachItem mode')
|
|
if router.get('language') != 'javaScript':
|
|
fail('route-action code node must use javaScript language')
|
|
js_code = router.get('jsCode', '')
|
|
for snippet in ("append_log", "notify", "unknown_action", "invalid_request", "status_code", "response_body"):
|
|
if snippet not in js_code:
|
|
fail(f'route-action jsCode missing expected snippet: {snippet!r}')
|
|
|
|
route_outputs = connections.get('route-action', {}).get('main', [])
|
|
if len(route_outputs) < 1:
|
|
fail('route-action must connect to Respond to Webhook')
|
|
|
|
responder = by_name['Respond to Webhook'].get('parameters', {})
|
|
if responder.get('respondWith') != 'json':
|
|
fail('Respond to Webhook must respondWith json')
|
|
if responder.get('responseBody') != '={{$json.response_body}}':
|
|
fail('Respond to Webhook must use $json.response_body as responseBody')
|
|
|
|
sample_paths = [
|
|
path.parent / 'test-append-log.json',
|
|
path.parent / 'test-notify.json',
|
|
]
|
|
for sample in sample_paths:
|
|
sample_data = load_json(sample)
|
|
if not isinstance(sample_data, dict):
|
|
fail(f'sample payload must be an object: {sample}')
|
|
if 'action' not in sample_data or 'args' not in sample_data:
|
|
fail(f'sample payload missing action/args: {sample}')
|
|
|
|
print('OK: workflow asset structure looks consistent')
|
|
print(f'- workflow: {path}')
|
|
print(f'- nodes: {len(nodes)}')
|
|
print('- router: code node with append_log + notify + fallback')
|
|
print('- samples: test-append-log.json, test-notify.json')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|