132 lines
5.2 KiB
Python
Executable File
132 lines
5.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Validate an n8n workflow JSON file for structural correctness.
|
|
|
|
Usage: validate-n8n-workflow.py path/to/workflow.json
|
|
Exit codes: 0 = valid, 1 = invalid, 2 = usage error.
|
|
"""
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
KNOWN_MIN_TYPE_VERSIONS = {
|
|
"n8n-nodes-base.code": 2,
|
|
"n8n-nodes-base.httpRequest": 4.2,
|
|
"n8n-nodes-base.webhook": 2,
|
|
"n8n-nodes-base.set": 3.4,
|
|
"n8n-nodes-base.scheduleTrigger": 1.2,
|
|
}
|
|
|
|
REQUIRED_TOP = ["name", "nodes", "connections"]
|
|
REQUIRED_NODE = ["id", "name", "type", "typeVersion", "position", "parameters"]
|
|
|
|
|
|
def validate(workflow):
|
|
errors, warnings = [], []
|
|
if not isinstance(workflow, dict):
|
|
return ["Root must be a JSON object."], []
|
|
for key in REQUIRED_TOP:
|
|
if key not in workflow:
|
|
errors.append(f"Top-level: missing required field '{key}'.")
|
|
if errors:
|
|
return errors, warnings
|
|
if not isinstance(workflow["nodes"], list):
|
|
errors.append("Top-level 'nodes' must be an array.")
|
|
if not isinstance(workflow["connections"], dict):
|
|
errors.append("Top-level 'connections' must be an object.")
|
|
if errors:
|
|
return errors, warnings
|
|
|
|
seen_names, seen_ids = set(), set()
|
|
for i, node in enumerate(workflow["nodes"]):
|
|
loc = f"nodes[{i}]"
|
|
if not isinstance(node, dict):
|
|
errors.append(f"{loc}: must be an object.")
|
|
continue
|
|
for key in REQUIRED_NODE:
|
|
if key not in node:
|
|
errors.append(f"{loc}: missing required field '{key}'.")
|
|
name = node.get("name")
|
|
if name:
|
|
loc = f"nodes[{i}] '{name}'"
|
|
if name in seen_names:
|
|
errors.append(f"{loc}: duplicate node name.")
|
|
seen_names.add(name)
|
|
nid = node.get("id")
|
|
if nid:
|
|
if nid in seen_ids:
|
|
errors.append(f"{loc}: duplicate node id '{nid}'.")
|
|
seen_ids.add(nid)
|
|
pos = node.get("position")
|
|
if pos is not None and not (
|
|
isinstance(pos, list) and len(pos) == 2
|
|
and all(isinstance(v, (int, float)) for v in pos)
|
|
):
|
|
errors.append(f"{loc}: 'position' must be [number, number].")
|
|
if node.get("parameters") is not None and not isinstance(node["parameters"], dict):
|
|
errors.append(f"{loc}: 'parameters' must be an object.")
|
|
creds = node.get("credentials")
|
|
if isinstance(creds, dict):
|
|
for ctype, cred in creds.items():
|
|
if not isinstance(cred, dict) or "name" not in cred:
|
|
errors.append(f"{loc}: credentials['{ctype}'] must have a 'name' field.")
|
|
ntype, tv = node.get("type"), node.get("typeVersion")
|
|
if isinstance(tv, (int, float)) and ntype in KNOWN_MIN_TYPE_VERSIONS:
|
|
min_tv = KNOWN_MIN_TYPE_VERSIONS[ntype]
|
|
if tv < min_tv:
|
|
warnings.append(f"{loc}: typeVersion {tv} for '{ntype}' is below recommended {min_tv}.")
|
|
|
|
for src, outputs in workflow["connections"].items():
|
|
if src not in seen_names:
|
|
errors.append(f"connections: source '{src}' does not match any node name.")
|
|
if not isinstance(outputs, dict):
|
|
errors.append(f"connections['{src}']: must be an object.")
|
|
continue
|
|
for out_type, out_list in outputs.items():
|
|
if not isinstance(out_list, list):
|
|
continue
|
|
for oi, branch in enumerate(out_list):
|
|
if branch is None:
|
|
continue
|
|
if not isinstance(branch, list):
|
|
errors.append(f"connections['{src}']['{out_type}'][{oi}]: must be an array.")
|
|
continue
|
|
for ti, target in enumerate(branch):
|
|
if not isinstance(target, dict):
|
|
errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: must be an object.")
|
|
continue
|
|
for k in ("node", "type", "index"):
|
|
if k not in target:
|
|
errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: missing '{k}'.")
|
|
tnode = target.get("node")
|
|
if tnode and tnode not in seen_names:
|
|
errors.append(f"connections['{src}']['{out_type}'][{oi}][{ti}]: target '{tnode}' not found.")
|
|
return errors, warnings
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) != 2:
|
|
print("Usage: validate-n8n-workflow.py <workflow.json>", file=sys.stderr)
|
|
sys.exit(2)
|
|
path = Path(sys.argv[1])
|
|
try:
|
|
workflow = json.loads(path.read_text())
|
|
except FileNotFoundError:
|
|
print(f"File not found: {path}", file=sys.stderr)
|
|
sys.exit(2)
|
|
except json.JSONDecodeError as e:
|
|
print(f"Invalid JSON: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
errors, warnings = validate(workflow)
|
|
for w in warnings:
|
|
print(f"WARN: {w}")
|
|
for e in errors:
|
|
print(f"ERROR: {e}", file=sys.stderr)
|
|
if errors:
|
|
print(f"\n{len(errors)} error(s), {len(warnings)} warning(s).", file=sys.stderr)
|
|
sys.exit(1)
|
|
print(f"OK — {len(workflow['nodes'])} nodes, {len(workflow['connections'])} connections, {len(warnings)} warning(s).")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|