#!/usr/bin/env python3 """Install and optionally test a Revenanas Clay workflow bundle. The `.py.txt` filename is intentional: Webflow serves it as a safe download, and Python can execute the file without requiring readers to rename it. """ from __future__ import annotations import argparse import json import os import re import subprocess import tempfile from pathlib import Path def apply_settings(bundle: dict, settings: dict[str, str]) -> dict: """Fill {{set:name}} placeholders (workspace-specific ids) anywhere, keys included.""" text = json.dumps(bundle) for name, value in settings.items(): text = text.replace("{{set:%s}}" % name, value) missing = sorted(set(re.findall(r"\{\{set:([a-z0-9_]+)\}\}", text))) if missing: raise SystemExit("Missing --set values for: " + ", ".join(missing)) return json.loads(text) def replace_refs(value, ids): if isinstance(value, dict): return {key: replace_refs(item, ids) for key, item in value.items()} if isinstance(value, list): return [replace_refs(item, ids) for item in value] if isinstance(value, str) and value.startswith("{{") and value.endswith("}}"): key = value[2:-2] if key in ids: return ids[key] return value return value class Clay: def __init__(self, binary: str, config_home: str | None): self.binary = binary self.env = os.environ.copy() if config_home: self.env["XDG_CONFIG_HOME"] = config_home def run(self, *args: str) -> dict: completed = subprocess.run( [self.binary, *args], env=self.env, text=True, capture_output=True, check=False, ) stream = completed.stdout.strip() or completed.stderr.strip() try: payload = json.loads(stream) except json.JSONDecodeError as exc: raise RuntimeError(f"Clay returned non-JSON output for {' '.join(args)}: {stream}") from exc if completed.returncode != 0 or "error" in payload: raise RuntimeError(f"Clay command failed for {' '.join(args)}: {payload}") return payload def with_input(self, args: list[str], payload: dict, flag: str = "--input") -> dict: with tempfile.NamedTemporaryFile("w", suffix=".json") as handle: json.dump(payload, handle) handle.flush() return self.run(*args, flag, handle.name) def terminal_outputs(run: dict) -> dict: terminals = [node for node in run.get("nodes", []) if node.get("outputs", {}).get("isTerminal")] if len(terminals) != 1: raise AssertionError(f"Expected one terminal node, found {len(terminals)}") outputs = terminals[0]["outputs"] return outputs.get("structuredOutputs") or outputs def assert_subset(actual: dict, expected: dict, label: str) -> None: mismatches = { key: {"expected": value, "actual": actual.get(key)} for key, value in expected.items() if actual.get(key) != value } if mismatches: raise AssertionError(f"{label} failed: {mismatches}") def validate_bundle(bundle: dict) -> None: assert bundle.get("format") == "revenanas.clay-workflow/v1" assert bundle.get("name") assert bundle.get("trigger", {}).get("key") == "trigger" node_keys = [node.get("key") for node in bundle.get("nodes", [])] assert node_keys and len(node_keys) == len(set(node_keys)) assert all(node.get("spec") for node in bundle["nodes"]) def install(bundle: dict, clay: Clay, name: str, run_tests: bool, workflow_id: str | None = None) -> dict: ids = {} if workflow_id: created = clay.run("workflows", "get", workflow_id) graph = clay.run("workflows", "graph", "get", workflow_id, "--mode", "full") trigger_nodes = [node for node in graph.get("nodes", []) if node.get("nodeType") == "trigger"] executable_nodes = [node for node in graph.get("nodes", []) if node.get("nodeType") != "trigger"] if len(trigger_nodes) != 1: raise AssertionError("--workflow-id requires a draft with exactly one trigger") ids[bundle["trigger"]["key"]] = trigger_nodes[0]["id"] bundle_names = {node["spec"]["name"]: node["key"] for node in bundle["nodes"]} unexpected = [node["name"] for node in executable_nodes if node["name"] not in bundle_names] if unexpected: raise AssertionError(f"Draft contains nodes outside this bundle: {unexpected}") for node in executable_nodes: ids[bundle_names[node["name"]]] = node["id"] else: created = clay.run("workflows", "create", "--name", name) workflow_id = created["id"] trigger = clay.with_input( ["workflows", "triggers", "create", workflow_id], bundle["trigger"]["spec"], ) trigger_id = trigger["resourceId"] persisted_trigger = clay.run("workflows", "triggers", "get", trigger_id) ids[bundle["trigger"]["key"]] = persisted_trigger["workflowNodeId"] for node in bundle["nodes"]: if node["key"] in ids: spec = replace_refs(node["spec"], ids) clay.with_input( ["workflows", "nodes", "update", workflow_id, ids[node["key"]]], spec, ) continue spec = replace_refs(node["spec"], ids) created_node = clay.with_input(["workflows", "nodes", "create", workflow_id], spec) ids[node["key"]] = created_node["nodeId"] validation = clay.run("workflows", "graph", "validate", workflow_id) if not validation.get("valid"): raise AssertionError(f"Installed graph is invalid: {validation}") test_results = [] if run_tests: for test in bundle.get("tests", []): started = clay.with_input( ["workflows", "runs", "test", workflow_id], test["inputs"], flag="--inputs", ) run = clay.run( "workflows", "runs", "get", workflow_id, started["runId"], "--wait", "60", "--verbose", ) if run.get("status") != "completed": raise AssertionError(f"{test['name']} did not complete: {run.get('error')}") outputs = terminal_outputs(run) assert_subset(outputs, test["expected_terminal"], test["name"]) test_results.append( { "name": test["name"], "run_id": run["runId"], "passed": True, "data_credits_used": run.get("dataCreditsUsed", 0), "action_credits_used": run.get("actionCreditsUsed", 0), } ) return { "workflow_id": workflow_id, "workflow_url": created["url"], "published": False, "graph_valid": True, "node_ids": ids, "tests": test_results, } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("bundle", type=Path) parser.add_argument("--clay-bin", default=os.environ.get("CLAY_CLI", "clay")) parser.add_argument("--config-home") parser.add_argument("--name") parser.add_argument("--workflow-id", help="Resume a partial draft whose existing node names match this bundle") parser.add_argument("--run-tests", action="store_true") parser.add_argument("--check", action="store_true") parser.add_argument("--set", action="append", default=[], metavar="NAME=VALUE", help="Fill a {{set:NAME}} placeholder with a workspace-specific id") args = parser.parse_args() bundle = json.loads(args.bundle.read_text()) validate_bundle(bundle) if args.check: print(json.dumps({"valid": True, "nodes": len(bundle["nodes"]), "tests": len(bundle.get("tests", []))})) return bundle = apply_settings(bundle, dict(item.split("=", 1) for item in args.set)) clay = Clay(args.clay_bin, args.config_home) result = install(bundle, clay, args.name or bundle["name"], args.run_tests, args.workflow_id) print(json.dumps(result, indent=2)) if __name__ == "__main__": main()