129 lines
4.5 KiB
Python
Executable File
129 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REQUIRED_NODE_NAMES = {
|
|
'Webhook',
|
|
'normalize-request',
|
|
'route-action',
|
|
'append-log-response',
|
|
'respond-append-log',
|
|
'notify-response',
|
|
'respond-notify',
|
|
'unknown-action-response',
|
|
'respond-unknown-action',
|
|
}
|
|
|
|
EXPECTED_DIRECT_TYPES = {
|
|
'Webhook': 'n8n-nodes-base.webhook',
|
|
'normalize-request': 'n8n-nodes-base.set',
|
|
'route-action': 'n8n-nodes-base.switch',
|
|
'append-log-response': 'n8n-nodes-base.set',
|
|
'respond-append-log': 'n8n-nodes-base.respondToWebhook',
|
|
'notify-response': 'n8n-nodes-base.set',
|
|
'respond-notify': 'n8n-nodes-base.respondToWebhook',
|
|
'unknown-action-response': 'n8n-nodes-base.set',
|
|
'respond-unknown-action': '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')
|
|
|
|
normalize = by_name['normalize-request'].get('parameters', {})
|
|
normalize_assignments = normalize.get('assignments', {}).get('assignments', [])
|
|
normalize_fields = {a.get('name') for a in normalize_assignments if isinstance(a, dict)}
|
|
for field in ('action', 'args', 'request_id'):
|
|
if field not in normalize_fields:
|
|
fail(f'normalize-request must assign {field!r}')
|
|
|
|
route = by_name['route-action'].get('parameters', {})
|
|
rule_values = route.get('rules', {}).get('values', [])
|
|
if len(rule_values) < 2:
|
|
fail('route-action must define at least two routing rules')
|
|
rule_names = {rule.get('outputKey') for rule in rule_values if isinstance(rule, dict)}
|
|
for action in ('append_log', 'notify'):
|
|
if action not in rule_names:
|
|
fail(f'route-action must have a routing rule for {action!r}')
|
|
|
|
route_outputs = connections.get('route-action', {}).get('main', [])
|
|
if len(route_outputs) < 3:
|
|
fail('route-action must expose append_log, notify, and fallback outputs')
|
|
|
|
for responder_name in ('respond-append-log', 'respond-notify', 'respond-unknown-action'):
|
|
params = by_name[responder_name].get('parameters', {})
|
|
if params.get('respondWith') != 'json':
|
|
fail(f'{responder_name} must respondWith json')
|
|
|
|
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(f'- rules: {len(rule_values)} + fallback')
|
|
print('- samples: test-append-log.json, test-notify.json')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|