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 if one does not, 2 if the case declares a
precondition this machine does not meet - reported as a skip, never as a pass.
"""
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 run_http_case(case, token, workflow_id):
    """A webhook case asserts on the HTTP response, not on node outputs."""
    spec = case["http"]
    url = BASE.replace("/api/v1", "") + spec["path"].replace("{workflowId}", workflow_id)
    data = json.dumps(spec.get("body", {})).encode()
    req = urllib.request.Request(url, data=data, method=spec.get("method", "POST"))
    req.add_header("Content-Type", "application/json")

    failures = []
    try:
        with urllib.request.urlopen(req, timeout=45) as res:
            status, headers, raw = res.status, dict(res.headers), res.read().decode()
    except urllib.error.HTTPError as e:
        status, headers, raw = e.code, dict(e.headers), e.read().decode()

    print("http: %s %s -> %s" % (spec.get("method", "POST"), spec["path"], status))
    print("  body: %s" % raw[:200])

    if "expectStatus" in spec and status != spec["expectStatus"]:
        failures.append("status: expected %s, got %s" % (spec["expectStatus"], status))

    for name, want in spec.get("expectHeaders", {}).items():
        got = headers.get(name)
        if got is None:
            failures.append("header %s: missing" % name)
        elif want not in got:
            failures.append("header %s: expected to contain %r, got %r" % (name, want, got))

    for fragment in spec.get("expectBodyContains", []):
        if fragment not in raw:
            failures.append("body: expected to contain %r" % fragment)

    return failures


def run_resume_case(case, token, execution_id):
    """Wait for the execution to pause, answer it, then let it finish."""
    spec = case["resume"]
    want_status = spec.get("waitForStatus", "waiting")

    execution = {}
    for _ in range(60):
        time.sleep(0.5)
        execution = call("GET", "/executions/%s" % execution_id, token)
        if execution.get("status") == want_status:
            break
        if execution.get("status") in ("completed", "failed", "cancelled"):
            raise SystemExit(
                "execution %s reached %s without pausing" % (execution_id, execution.get("status")))
    else:
        raise SystemExit("execution %s never reached %s" % (execution_id, want_status))

    by_id = {n["nodeId"]: n for n in execution.get("nodeExecutions", [])}
    source = by_id.get(spec["tokenFrom"])
    if not source:
        raise SystemExit("resume: node %s did not run" % spec["tokenFrom"])
    resume_token = (source.get("output") or {}).get("token")
    if not resume_token:
        raise SystemExit("resume: node %s produced no token" % spec["tokenFrom"])

    body = dict(spec.get("payload", {}))
    body["token"] = resume_token
    call("POST", "/executions/%s/resume" % execution_id, token, body)
    print("resumed %s with token %s" % (execution_id, resume_token))


def precondition_unmet(case):
    """Why this case cannot run here, or None if it can.

    A couple of cases need a model loaded on the SD.cpp server, which unloads
    itself when idle. Without this they fail for a reason that has nothing to do
    with the code, and a failure everyone learns to ignore is worse than no
    test. Skipping is only honest because it is reported as a skip - never as a
    pass.
    """
    needs = case.get("requires")
    if not needs:
        return None

    probe = needs.get("http")
    if probe:
        try:
            with urllib.request.urlopen(probe, timeout=10) as response:
                body = json.loads(response.read().decode())
        except Exception as exc:
            return f"{probe} could not be read ({exc})"

        for path, expected in (needs.get("expect") or {}).items():
            actual = body
            for part in path.split("."):
                actual = (actual or {}).get(part) if isinstance(actual, dict) else None
            if actual != expected:
                return f"{probe} reports {path}={actual!r}, this case needs {expected!r}"

    return None


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

    unmet = precondition_unmet(case)
    if unmet:
        print(f"SKIP {case['name']}: {unmet}")
        return 2

    token = login()

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

    # Workflows the case needs to exist before it runs - a sub-workflow it
    # calls. Created here rather than named by id, so a case is not silently
    # tied to a row someone made by hand once and can delete.
    helper_ids = []
    for helper in case.get("helpers", []):
        made = call("POST", "/workflows", token, {
            "name": helper["name"],
            "nodes": helper["nodes"],
            "connections": helper.get("connections", []),
        })
        helper_id = made.get("id") or made.get("_id")
        if not helper_id:
            raise SystemExit(f"no id for helper {helper['key']}")
        helper_ids.append(helper_id)
        if helper.get("active", True):
            call("POST", f"/workflows/{helper_id}/activate", token, {})
        # Substituted everywhere, so the case refers to it by name.
        case = json.loads(json.dumps(case).replace("{{helper:" + helper["key"] + "}}", helper_id))

    created = call("POST", "/workflows", token, {
        "name": case["name"],
        "nodes": case["nodes"],
        "connections": case["connections"],
        "settings": case.get("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:
        if "http" in case:
            call("POST", f"/workflows/{workflow_id}/activate", token, {})
            failures = run_http_case(case, token, workflow_id)
            print("\nFAIL" if failures else "\nPASS")
            for f in failures:
                print("  " + f)
            return 1 if failures else 0

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

        if "resume" in case:
            run_resume_case(case, token, execution_id)

        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 "errorContains" in want:
                actual_error = got.get("error") or ""
                if want["errorContains"] not in actual_error:
                    failures.append(
                        f"{node_id}: error does not contain {want['errorContains']!r}, "
                        f"got {actual_error!r}"
                    )
            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") != "skipped":
                failures.append(
                    f"{node_id}: ran, but should not have (status: {got.get('status')!r})"
                )

        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)
        for helper_id in helper_ids:
            call("DELETE", f"/workflows/{helper_id}", token)


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