111 lines
4.0 KiB
Python
Executable File
111 lines
4.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REQUIRED_NODE_NAMES = {
|
|
'Webhook',
|
|
'route-action',
|
|
'route-dispatch',
|
|
'Send Telegram Notification',
|
|
'Send Discord Notification',
|
|
'Respond to Webhook',
|
|
}
|
|
|
|
EXPECTED_TYPES = {
|
|
'Webhook': 'n8n-nodes-base.webhook',
|
|
'route-action': 'n8n-nodes-base.code',
|
|
'route-dispatch': 'n8n-nodes-base.switch',
|
|
'Send Telegram Notification': 'n8n-nodes-base.telegram',
|
|
'Send Discord Notification': 'n8n-nodes-base.httpRequest',
|
|
'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)
|
|
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_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'].get('parameters', {})
|
|
if webhook.get('path') != 'openclaw-action':
|
|
fail('Webhook.path must be openclaw-action')
|
|
if webhook.get('httpMethod') != 'POST':
|
|
fail('Webhook.httpMethod must be POST')
|
|
if webhook.get('responseMode') != 'responseNode':
|
|
fail('Webhook.responseMode must be responseNode')
|
|
|
|
router = by_name['route-action'].get('parameters', {})
|
|
js_code = router.get('jsCode', '')
|
|
for snippet in ('append_log', 'notify', 'unknown_action', 'invalid_request', '$getWorkflowStaticData', 'actionLog', 'retained_entries', 'notify_text'):
|
|
if snippet not in js_code:
|
|
fail(f'route-action jsCode missing expected snippet: {snippet!r}')
|
|
|
|
switch = by_name['route-dispatch'].get('parameters', {})
|
|
values = switch.get('rules', {}).get('values', [])
|
|
names = {v.get('outputKey') for v in values if isinstance(v, dict)}
|
|
if 'notify' not in names:
|
|
fail('route-dispatch must route notify')
|
|
|
|
telegram = by_name['Send Telegram Notification']
|
|
if telegram.get('credentials', {}).get('telegramApi', {}).get('name') != 'Telegram Bot (OpenClaw)':
|
|
fail('Send Telegram Notification must use Telegram Bot (OpenClaw) credential')
|
|
|
|
discord = by_name['Send Discord Notification']
|
|
if discord.get('credentials', {}).get('httpHeaderAuth', {}).get('name') != 'Discord Bot Auth':
|
|
fail('Send Discord Notification must use Discord Bot Auth credential')
|
|
|
|
responder = by_name['Respond to Webhook'].get('parameters', {})
|
|
if responder.get('respondWith') != 'json':
|
|
fail('Respond to Webhook must respondWith json')
|
|
|
|
for sample in (path.parent / 'test-append-log.json', path.parent / 'test-notify.json'):
|
|
sample_data = load_json(sample)
|
|
if not isinstance(sample_data, dict) or '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('- routes: append_log -> workflow static data, notify -> Telegram + Discord, fallback -> JSON error')
|
|
print('- samples: test-append-log.json, test-notify.json')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|