smartbotic/scripts/verify-node.py

#!/usr/bin/env python3
"""Run one node verification case against the live SmartBotic services.

Usage: scripts/verify-node.py tests/nodes/<case>.json
Exit 0 if every assertion holds, 1 otherwise.
"""
import json
import sys
import time
import urllib.error
import urllib.request

BASE = "http://localhost:8090/api/v1"


def call(method, path, token=None, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(BASE + path, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    if token:
        req.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            raw = res.read()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        raise SystemExit(f"{method} {path} failed: {e.code} {e.read().decode()[:400]}")


def login():
    return call("POST", "/auth/login", body={"username": "admin", "password": "admin"})["accessToken"]


def subset_matches(expected, actual, path):
    """Every key in expected must be present and equal in actual. Returns list of failures."""
    fails = []
    if isinstance(expected, dict):
        if not isinstance(actual, dict):
            return [f"{path}: expected an object, got {type(actual).__name__}"]
        for key, want in expected.items():
            if key not in actual:
                fails.append(f"{path}.{key}: missing")
            else:
                fails += subset_matches(want, actual[key], f"{path}.{key}")
    elif isinstance(expected, list):
        if not isinstance(actual, list):
            return [f"{path}: expected a list, got {type(actual).__name__}"]
        if len(expected) != len(actual):
            fails.append(f"{path}: expected {len(expected)} items, got {len(actual)}")
        else:
            for i, want in enumerate(expected):
                fails += subset_matches(want, actual[i], f"{path}[{i}]")
    elif expected != actual:
        fails.append(f"{path}: expected {expected!r}, got {actual!r}")
    return fails


def main():
    if len(sys.argv) != 2:
        raise SystemExit("usage: verify-node.py <case.json>")
    case = json.load(open(sys.argv[1]))
    token = login()

    call("POST", "/nodes/migrate", token, {"nodesPath": "./nodes"})

    created = call("POST", "/workflows", token, {
        "name": case["name"],
        "nodes": case["nodes"],
        "connections": case["connections"],
        "settings": {},
    })
    workflow_id = created.get("id") or created.get("_id")
    if not workflow_id:
        raise SystemExit(f"no workflow id in create response: {json.dumps(created)[:300]}")

    try:
        started = call("POST", f"/workflows/{workflow_id}/execute", token, {})
        execution_id = started["executionId"]

        execution = None
        for _ in range(60):
            time.sleep(0.5)
            execution = call("GET", f"/executions/{execution_id}", token)
            if execution.get("status") in ("completed", "failed", "cancelled"):
                break
        else:
            raise SystemExit(f"execution {execution_id} did not finish in 30s")

        by_id = {n["nodeId"]: n for n in execution.get("nodeExecutions", [])}
        failures = []

        for node_id, want in case.get("expect", {}).items():
            got = by_id.get(node_id)
            if got is None:
                failures.append(f"{node_id}: did not run")
                continue
            if "status" in want and got.get("status") != want["status"]:
                failures.append(
                    f"{node_id}: status {got.get('status')!r}, expected {want['status']!r}"
                    + (f" (error: {got.get('error')})" if got.get("error") else "")
                )
            if "output" in want:
                failures += subset_matches(want["output"], got.get("output"), node_id)

        for node_id in case.get("expectMissing", []):
            got = by_id.get(node_id)
            if got is not None and got.get("status") == "completed":
                failures.append(f"{node_id}: ran, but should not have")

        print(f"case: {case['name']}  execution: {execution_id}  status: {execution.get('status')}")
        for node_id, node in sorted(by_id.items()):
            print(f"  {node_id:24} {node.get('status'):10} {json.dumps(node.get('output'))[:120]}")

        if failures:
            print("\nFAIL")
            for f in failures:
                print("  " + f)
            return 1
        print("\nPASS")
        return 0
    finally:
        call("DELETE", f"/workflows/{workflow_id}", token)


if __name__ == "__main__":
    sys.exit(main())