# /// script # requires-python = ">=3.10" # dependencies = [] # /// """Course helper: check, push and test the n8n workflows that a CLI coding agent wrote. uv run tools/flow.py doctor # is everything ready? uv run tools/flow.py check my-work/assistant.json # find mistakes before n8n sees the file uv run tools/flow.py push agents/*.json my-work/assistant.json # import + publish (one n8n restart) uv run tools/flow.py off agents/meeting-agent.json # unpublish (simulate an agent that is down) uv run tools/flow.py call my-assistant requests/composite.json uv run tools/flow.py test cases/lab02.json -v [--repeat 3] # -v: show each request and response JSON Run it from the day folder (Last3Mon/68). n8n is the day 65 install (~/day65-n8n, Docker project day65-n8n), started from Last3Mon/65/n8n. Environment overrides: N8N_CONTAINER (default: the running n8n service of project day65-n8n, else day65-n8n-n8n-1) N8N_HOME (default ~/day65-n8n), N8N_URL (default http://localhost:5678) """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import os import re import shutil import subprocess import sys import time import urllib.error import urllib.request import uuid from pathlib import Path ROOT = Path(__file__).resolve().parent.parent CACHE = Path(__file__).resolve().parent / ".cache" def n8n_home() -> Path: """Where the official n8n install lives: N8N_HOME, else Last3Mon/65/n8n/.env, else ~/day65-n8n.""" if os.getenv("N8N_HOME"): return Path(os.environ["N8N_HOME"]).expanduser() env_file = ROOT.parent / "65" / "n8n" / ".env" if env_file.exists(): for line in env_file.read_text(encoding="utf-8").splitlines(): if line.startswith("N8N_HOME="): value = line.split("=", 1)[1].strip().strip("'\"") if value: return (env_file.parent / Path(value).expanduser()).resolve() return Path.home() / "day65-n8n" N8N_HOME = n8n_home() START_HINT = "cd ~/aio-01d4-stu/Last3Mon/65/n8n && docker compose start (n8n install: ~/day65-n8n)" def find_container() -> str: """The n8n container of the day 65 project, whichever folder it was started from.""" if os.getenv("N8N_CONTAINER"): return os.environ["N8N_CONTAINER"] if shutil.which("docker"): result = subprocess.run(["docker", "ps", "-a", "--filter", "label=com.docker.compose.project=day65-n8n", "--filter", "label=com.docker.compose.service=n8n", "--format", "{{.Names}}"], capture_output=True, text=True) names = result.stdout.split() if names: return names[0] return "day65-n8n-n8n-1" CONTAINER = find_container() N8N_URL = os.getenv("N8N_URL", "http://localhost:5678").rstrip("/") # Versions written in AGENTS.md. Other versions still work if n8n knows them (warning only). RECOMMENDED = { "n8n-nodes-base.webhook": 2, "n8n-nodes-base.respondToWebhook": 1.1, "n8n-nodes-base.if": 2.2, "n8n-nodes-base.switch": 3.2, "n8n-nodes-base.set": 3.4, "n8n-nodes-base.code": 2, "n8n-nodes-base.httpRequest": 4.2, "n8n-nodes-base.httpRequestTool": 4.2, "@n8n/n8n-nodes-langchain.agent": 3.1, "@n8n/n8n-nodes-langchain.lmChatOpenAi": 1.2, } AGENT_HOOKS = ("it-agent", "meeting-agent") # the agents in agents/ that the router calls COURSE_CREDENTIALS = {"openAiCourse0001": "OpenAI (.env)"} # OpenAI only from day 66 (no Ollama) CONNECTION_TYPES = {"main", "ai_tool", "ai_languageModel", "ai_memory", "ai_outputParser"} SECRET = re.compile(r"\bsk-[A-Za-z0-9_-]{16,}") OK, BAD, WARN = "✔", "✘", "!" # ---------------------------------------------------------------- helpers def die(message: str) -> None: print(f"{BAD} {message}", file=sys.stderr) sys.exit(1) def docker(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: if not shutil.which("docker"): die("docker command not found. Start Docker first.") result = subprocess.run(["docker", *args], input=stdin, capture_output=True, text=True) if check and result.returncode != 0: die(f"docker {' '.join(args[:2])} failed:\n{(result.stderr or result.stdout).strip()}") return result def n8n(*args: str, stdin: str | None = None, check: bool = True) -> subprocess.CompletedProcess: return docker("exec", "-i", CONTAINER, "n8n", *args, stdin=stdin, check=check) def container_running() -> bool: if not shutil.which("docker"): return False result = subprocess.run(["docker", "inspect", "-f", "{{.State.Running}}", CONTAINER], capture_output=True, text=True) return result.stdout.strip() == "true" def http(method: str, url: str, body: object | None = None, timeout: float = 300) -> tuple[int, object, float]: data = None if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8") request = urllib.request.Request(url, data=data, method=method, headers={"Content-Type": "application/json"}) start = time.monotonic() try: with urllib.request.urlopen(request, timeout=timeout) as response: status, raw = response.status, response.read() except urllib.error.HTTPError as error: status, raw = error.code, error.read() except (urllib.error.URLError, TimeoutError, ConnectionError) as error: return 0, {"error": str(getattr(error, "reason", error))}, time.monotonic() - start text = raw.decode("utf-8", errors="replace") try: payload: object = json.loads(text) if text.strip() else "" except json.JSONDecodeError: payload = text return status, payload, time.monotonic() - start def load_json(path: Path) -> object: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: die(f"file not found: {path}") except json.JSONDecodeError as error: die(f"{path} is not valid JSON (line {error.lineno}, column {error.colno}): {error.msg}") def resolve(path: str) -> Path: candidate = Path(path) if candidate.exists() or candidate.is_absolute(): return candidate return ROOT / path def node_versions() -> dict[str, list[float]]: """Node types and versions that this n8n really has (exported once, then cached).""" cache = CACHE / "node-versions.json" if cache.exists(): return json.loads(cache.read_text()) if not container_running(): return {} n8n("export:nodes", "--output=/tmp/course-nodes.json") raw = docker("exec", CONTAINER, "sh", "-c", "cat /tmp/course-nodes.json && rm -f /tmp/course-nodes.json").stdout versions: dict[str, set[float]] = {} for node in json.loads(raw): value = node.get("version") versions.setdefault(node["name"], set()).update(value if isinstance(value, list) else [value]) result = {name: sorted(values) for name, values in versions.items()} CACHE.mkdir(exist_ok=True) cache.write_text(json.dumps(result)) return result def walk_strings(value: object, where: str = ""): if isinstance(value, str): yield where, value elif isinstance(value, dict): for key, item in value.items(): yield from walk_strings(item, f"{where}.{key}" if where else key) elif isinstance(value, list): for index, item in enumerate(value): yield from walk_strings(item, f"{where}[{index}]") def webhooks(workflow: dict) -> list[tuple[str, str]]: found = [] for node in workflow.get("nodes", []): if node.get("type") == "n8n-nodes-base.webhook": parameters = node.get("parameters", {}) found.append((str(parameters.get("httpMethod", "GET")).upper(), str(parameters.get("path", "")).strip("/"))) return found # ---------------------------------------------------------------- check def check_workflow(workflow: object) -> tuple[list[str], list[str]]: errors: list[str] = [] warnings: list[str] = [] if not isinstance(workflow, dict): return ["the file must be one workflow object { name, nodes, connections }"], warnings if not str(workflow.get("name", "")).strip(): errors.append("missing workflow 'name'") nodes = workflow.get("nodes") connections = workflow.get("connections") if not isinstance(nodes, list) or not nodes: return errors + ["'nodes' must be a non-empty list"], warnings if not isinstance(connections, dict): errors.append("'connections' must be an object") connections = {} known = node_versions() names: dict[str, dict] = {} for index, node in enumerate(nodes): label = f"node #{index + 1}" if not isinstance(node, dict): errors.append(f"{label} is not an object") continue name = node.get("name") if not name: errors.append(f"{label} has no 'name'") continue label = f"node '{name}'" if name in names: errors.append(f"{label} appears twice (node names must be unique)") names[name] = node node_type, version = node.get("type"), node.get("typeVersion") if not node_type or version is None: errors.append(f"{label} needs 'type' and 'typeVersion'") elif known and node_type not in known: errors.append(f"{label}: type '{node_type}' does not exist in this n8n") elif known and version not in known[node_type]: errors.append(f"{label}: {node_type} has no typeVersion {version} (available: {known[node_type]})") elif node_type in RECOMMENDED and version != RECOMMENDED[node_type]: warnings.append(f"{label}: typeVersion {version} (AGENTS.md uses {RECOMMENDED[node_type]})") if not isinstance(node.get("parameters"), dict): errors.append(f"{label} needs 'parameters' (use {{}} when empty)") position = node.get("position") if not (isinstance(position, list) and len(position) == 2): errors.append(f"{label} needs 'position': [x, y]") for kind, credential in (node.get("credentials") or {}).items(): credential_id = (credential or {}).get("id") if credential_id not in COURSE_CREDENTIALS: errors.append(f"{label}: credential '{kind}' must use a course id {list(COURSE_CREDENTIALS)}, got {credential_id!r}") for where, text in walk_strings(node.get("parameters", {})): if SECRET.search(text): errors.append(f"{label}: parameters.{where} looks like an API key. Use a credential instead.") if "{{" in text and "}}" in text and not text.startswith("="): warnings.append(f"{label}: parameters.{where} has {{{{ }}}} but does not start with '=' (it stays plain text)") targets_of: dict[str, set[str]] = {} for source, outputs in connections.items(): if source not in names: errors.append(f"connections: source '{source}' is not a node name") continue if not isinstance(outputs, dict): errors.append(f"connections['{source}'] must be an object like {{\"main\": [[...]]}}") continue for kind, branches in outputs.items(): if kind not in CONNECTION_TYPES: errors.append(f"connections['{source}']: unknown connection type '{kind}'") for branch in branches or []: for link in branch or []: target = (link or {}).get("node") if target not in names: errors.append(f"connections['{source}'] points to '{target}', which is not a node name") elif link.get("type") != kind: errors.append(f"connections['{source}'] -> '{target}': link type must be '{kind}'") else: targets_of.setdefault(target, set()).add(kind) if kind != "main": targets_of.setdefault(f"{source}::from", set()).add(kind) has_respond = any(node.get("type") == "n8n-nodes-base.respondToWebhook" for node in names.values()) for name, node in names.items(): node_type = node.get("type", "") parameters = node.get("parameters", {}) if isinstance(node.get("parameters"), dict) else {} if node_type == "n8n-nodes-base.webhook": if not str(parameters.get("path", "")).strip("/"): errors.append(f"node '{name}': webhook needs a 'path'") if parameters.get("responseMode") == "responseNode" and not has_respond: errors.append(f"node '{name}': responseMode is 'responseNode' but there is no Respond to Webhook node") if node_type == "@n8n/n8n-nodes-langchain.agent" and "ai_languageModel" not in targets_of.get(name, set()): errors.append(f"node '{name}': AI Agent needs a chat model connected with 'ai_languageModel'") if node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool"): if "ai_tool" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a tool must be connected to an AI Agent with 'ai_tool'") if node_type.startswith("@n8n/n8n-nodes-langchain.lmChat"): if "ai_languageModel" not in targets_of.get(f"{name}::from", set()): errors.append(f"node '{name}': a chat model must be connected to an AI Agent with 'ai_languageModel'") if node_type != "n8n-nodes-base.webhook" and not node_type.startswith("@n8n/n8n-nodes-langchain.lm") \ and not (node_type.endswith("Tool") or node_type.startswith("@n8n/n8n-nodes-langchain.tool")) \ and "main" not in targets_of.get(name, set()) and "trigger" not in node_type.lower(): warnings.append(f"node '{name}': nothing is connected into it (it never runs)") if not webhooks(workflow): warnings.append("no Webhook node: 'push' works, but 'call' and 'test' need a webhook path") return errors, warnings def cmd_check(args: argparse.Namespace) -> int: path = resolve(args.file) errors, warnings = check_workflow(load_json(path)) for warning in warnings: print(f"{WARN} {warning}") for error in errors: print(f"{BAD} {error}") if errors: print(f"{BAD} {path.name}: {len(errors)} error(s). Fix them and run check again.") return 1 print(f"{OK} {path.name}: no errors" + (f" ({len(warnings)} warning(s))" if warnings else "")) return 0 # ---------------------------------------------------------------- push def stable_id(path: Path) -> str: return "w" + hashlib.sha1(path.stem.encode("utf-8")).hexdigest()[:15] def wait_until_ready(paths: list[tuple[str, str]], seconds: int = 90) -> bool: deadline = time.monotonic() + seconds while time.monotonic() < deadline: status, _, _ = http("GET", f"{N8N_URL}/healthz", timeout=3) if status == 200: break time.sleep(1) else: return False for method, path in paths: while time.monotonic() < deadline: # A webhook that is registered for POST answers a GET with a hint, not "is not registered". status, payload, _ = http("GET" if method != "GET" else "HEAD", f"{N8N_URL}/webhook/{path}", timeout=5) text = json.dumps(payload, ensure_ascii=False) if "not registered for" in text or status not in (0, 404, 500, 502, 503): break time.sleep(1) time.sleep(2) # webhooks of the other published workflows finish registering right after return time.monotonic() < deadline def exported_workflows() -> list[dict]: exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:workflow --all --output=/tmp/course-all.json >/dev/null 2>&1; " "cat /tmp/course-all.json 2>/dev/null; rm -f /tmp/course-all.json", check=False).stdout try: return json.loads(exported) if exported.strip() else [] except json.JSONDecodeError: return [] def prepare(path: Path) -> dict | None: workflow = load_json(path) errors, warnings = check_workflow(workflow) for warning in warnings: print(f"{WARN} {path.name}: {warning}") if errors: for error in errors: print(f"{BAD} {path.name}: {error}") return None workflow = dict(workflow) if not re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))): workflow["id"] = stable_id(path) workflow["active"] = False workflow.setdefault("settings", {}).setdefault("executionOrder", "v1") for node in workflow["nodes"]: node.setdefault("id", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/{node['name']}"))) if node["type"] == "n8n-nodes-base.webhook": node.setdefault("webhookId", str(uuid.uuid5(uuid.NAMESPACE_URL, f"{workflow['id']}/webhook/{node['name']}"))) return workflow def cmd_push(args: argparse.Namespace) -> int: paths = [resolve(file) for file in args.files] workflows = [prepare(path) for path in paths] if None in workflows: print(f"{BAD} not pushed. Fix the errors first (uv run tools/flow.py check ).") return 1 if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") mine_ids = {workflow["id"] for workflow in workflows} hooks = {hook for workflow in workflows for hook in webhooks(workflow)} # Another published workflow with the same webhook path would block this one. for other in exported_workflows(): if other.get("id") not in mine_ids and other.get("active") and hooks & set(webhooks(other)): n8n("unpublish:workflow", f"--id={other['id']}") print(f"{WARN} unpublished '{other.get('name')}' ({other['id']}): it used the same webhook path") for workflow in workflows: docker("exec", "-i", CONTAINER, "sh", "-c", "cat > /tmp/course-workflow.json && n8n import:workflow --input=/tmp/course-workflow.json; " "status=$?; rm -f /tmp/course-workflow.json; exit $status", stdin=json.dumps(workflow, ensure_ascii=False)) n8n("publish:workflow", f"--id={workflow['id']}") # The CLI changes the database only; one restart makes the running n8n load the new versions (about 5 s). docker("restart", CONTAINER) ready = wait_until_ready(sorted(hooks)) for workflow in workflows: print(f"{OK} pushed '{workflow['name']}' (id {workflow['id']})") for method, hook in sorted(webhooks(workflow)): print(f" {method} {N8N_URL}/webhook/{hook}") if not ready: print(f"{WARN} n8n is still starting. Wait a few seconds before 'call' or 'test'.") return 0 def cmd_off(args: argparse.Namespace) -> int: """Unpublish workflows (by file) to see how the others behave when one agent is down.""" if not container_running(): die(f"n8n container '{CONTAINER}' is not running. {START_HINT}") wanted = {stable_id(resolve(file)): file for file in args.files} for file in args.files: workflow = load_json(resolve(file)) if re.fullmatch(r"[A-Za-z0-9]{10,}", str(workflow.get("id", ""))): wanted[workflow["id"]] = file active = {item.get("id") for item in exported_workflows() if item.get("active")} for workflow_id, file in wanted.items(): if workflow_id in active: n8n("unpublish:workflow", f"--id={workflow_id}") print(f"{OK} unpublished {file} (id {workflow_id})") docker("restart", CONTAINER) wait_until_ready([]) print(f"{OK} n8n restarted. Turn it back on with: uv run tools/flow.py push {' '.join(args.files)}") return 0 # ---------------------------------------------------------------- call / test def request_body(spec: str | dict, base: Path = ROOT) -> object: if isinstance(spec, dict): return spec text = spec.strip() if text.startswith("{"): try: return json.loads(text) except json.JSONDecodeError as error: die(f"request is not valid JSON: {error.msg}") path = Path(text) if not path.exists(): path = base / text return load_json(path) def cmd_call(args: argparse.Namespace) -> int: body = request_body(args.request) if args.request else {} status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{args.path.strip('/')}", body) print(f"HTTP {status} · {seconds:.1f}s") print(json.dumps(payload, ensure_ascii=False, indent=2) if not isinstance(payload, str) else payload) return 0 if 200 <= status < 300 else 1 def field(payload: object, dotted: str) -> object: value = payload for part in dotted.split("."): if isinstance(value, dict): value = value.get(part) elif isinstance(value, list) and part.isdigit() and int(part) < len(value): value = value[int(part)] else: return None return value def as_text(value: object) -> str: return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) def judge(expect: dict, status: int, payload: object) -> list[str]: problems = [] if status != expect.get("status", 200): problems.append(f"status {status} (expected {expect.get('status', 200)})") for name, wanted in (expect.get("equals") or {}).items(): actual = field(payload, name) if actual != wanted: problems.append(f"{name} = {as_text(actual)[:60]} (expected {as_text(wanted)})") for name, words in (expect.get("contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") if not any(word in text for word in words): problems.append(f"{name} has none of {words}") for name, words in (expect.get("not_contains") or {}).items(): words = [words] if isinstance(words, str) else words text = as_text(field(payload, name) or "") leaked = [word for word in words if word in text] if leaked: problems.append(f"{name} must not have {leaked}") return problems def show_json(label: str, value: object) -> None: """Print one JSON value under a label, indented below the case line (test -v).""" text = json.dumps(value, ensure_ascii=False, indent=2) if not isinstance(value, str) else value print(f" {label}") for row in text.splitlines(): print(f" {row}") def cmd_test(args: argparse.Namespace) -> int: path = resolve(args.cases) suite = load_json(path) hook = (args.webhook or suite.get("webhook", "")).strip("/") cases = suite.get("cases", []) if not hook or not cases: die(f"{path.name} needs \"webhook\" and a non-empty \"cases\" list") print(f"POST {N8N_URL}/webhook/{hook} · {len(cases)} case(s) × {args.repeat}") results, passed = [], 0 for case in cases: body = request_body(case.get("body") or case.get("request", "{}")) case_hook = (case.get("webhook") or hook).strip("/") if not args.webhook else hook for run in range(1, args.repeat + 1): status, payload, seconds = http("POST", f"{N8N_URL}/webhook/{case_hook}", body) problems = judge(case.get("expect", {}), status, payload) passed += not problems mark = OK if not problems else BAD suffix = f" #{run}" if args.repeat > 1 else "" print(f"{mark} {case.get('name', '?')}{suffix} · {seconds:.1f}s" + ("" if not problems else " · " + "; ".join(problems))) if args.verbose: show_json(f"→ request POST /webhook/{case_hook}", body) show_json(f"← response HTTP {status}", payload) results.append({"case": case.get("name"), "webhook": case_hook, "run": run, "passed": not problems, "problems": problems, "seconds": round(seconds, 1), "status": status, "request": body, "response": payload}) total = len(results) print(f"Passed {passed}/{total}") runs = ROOT / "runs" runs.mkdir(exist_ok=True) stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S") report = runs / f"{path.stem}-{stamp}.json" report.write_text(json.dumps({"cases": str(path.name), "webhook": hook, "passed": passed, "total": total, "results": results}, ensure_ascii=False, indent=2), encoding="utf-8") print(f"report: {report.relative_to(ROOT)}") return 0 if passed == total else 1 # ---------------------------------------------------------------- doctor def cmd_doctor(args: argparse.Namespace) -> int: failures = 0 def line(good: bool, text: str, hint: str = "") -> None: nonlocal failures failures += not good print(f"{OK if good else BAD} {text}" + ("" if good or not hint else f"\n → {hint}")) installed = (N8N_HOME / "compose.yml").exists() print(f"{OK if installed else WARN} n8n install folder {N8N_HOME}" + ("" if installed else "\n → day 65 SETUP-N8N 5단계 (get.n8n.io 설치). 다른 곳에 설치했다면 Last3Mon/65/n8n/.env 에 N8N_HOME=<폴더>")) running = container_running() line(running, f"n8n container {CONTAINER}", START_HINT) status, _, _ = http("GET", f"{N8N_URL}/healthz", timeout=5) line(status == 200, f"n8n {N8N_URL}", "wait 10 seconds after starting n8n, then run doctor again") status, _, _ = http("GET", "http://localhost:8800/health", timeout=5) line(status == 200, "data API http://localhost:8800", "cd ~/aio-01d4-stu/Last3Mon/67/data-api && docker compose up -d --force-recreate (start n8n first)") if running: probe = docker("exec", CONTAINER, "node", "-e", "fetch('http://data-api:8800/health').then(r=>console.log(r.status)).catch(()=>console.log(0))", check=False).stdout.strip() line(probe == "200", "n8n → http://data-api:8800", "data API must join the day65-n8n_default network (compose.yml)") exported = docker("exec", CONTAINER, "sh", "-c", "n8n export:credentials --all --output=/tmp/course-cred.json >/dev/null 2>&1; " "cat /tmp/course-cred.json 2>/dev/null; rm -f /tmp/course-cred.json", check=False).stdout try: ids = {item.get("id") for item in json.loads(exported)} if exported.strip() else set() except json.JSONDecodeError: ids = set() for credential_id, name in COURSE_CREDENTIALS.items(): line(credential_id in ids, f"credential '{name}' ({credential_id})", "cd ~/aio-01d4-stu/Last3Mon/65/n8n && bash register-credentials.sh") line(bool(node_versions()), "node catalog (for check)") for hook in AGENT_HOOKS: _, payload, _ = http("GET", f"{N8N_URL}/webhook/{hook}", timeout=5) line("not registered for" in json.dumps(payload, ensure_ascii=False), f"agent webhook /{hook}", "uv run tools/flow.py push agents/it-agent.json agents/meeting-agent.json") agents = [name for name in ("codex", "claude", "agy") if shutil.which(name)] line(bool(agents), "CLI coding agent: " + (", ".join(agents) if agents else "none"), "install one of codex / claude / agy (see Lecture/68/00-DAY68-PREP.md)") print("ready" if not failures else f"{failures} item(s) to fix") return 0 if not failures else 1 def main() -> int: parser = argparse.ArgumentParser(description="Check, push and test course n8n workflows.") commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("doctor", help="check n8n, data API, credentials and CLI agents").set_defaults(run=cmd_doctor) item = commands.add_parser("check", help="find mistakes in a workflow file") item.add_argument("file") item.set_defaults(run=cmd_check) item = commands.add_parser("push", help="import and publish workflow files (one n8n restart)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_push) item = commands.add_parser("off", help="unpublish workflow files (an agent that is down)") item.add_argument("files", nargs="+") item.set_defaults(run=cmd_off) item = commands.add_parser("call", help="POST one request to a webhook path") item.add_argument("path", help="webhook path, e.g. my-assistant") item.add_argument("request", nargs="?", help="request file or inline JSON") item.set_defaults(run=cmd_call) item = commands.add_parser("test", help="run a test case file") item.add_argument("cases") item.add_argument("--repeat", type=int, default=1, help="run every case N times") item.add_argument("--webhook", help="override the webhook path in the case file") item.add_argument("-v", "--verbose", action="store_true", help="print the request and response JSON of every run") item.set_defaults(run=cmd_test) args = parser.parse_args() return args.run(args) if __name__ == "__main__": sys.exit(main())