314 lines
11 KiB
TypeScript
314 lines
11 KiB
TypeScript
import { readFileSync } from "node:fs";
|
|
import { mkdtemp, readFile, rm } from "node:fs/promises";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import { afterEach, describe, expect, it } from "vitest";
|
|
import type {
|
|
Catalog,
|
|
CatalogComponent,
|
|
FlowDefinition,
|
|
FlowEdge,
|
|
FlowNode,
|
|
RunEvent
|
|
} from "../../src/types";
|
|
import "../../src/capabilities";
|
|
import { register } from "../../src/engine/registry";
|
|
import { runFlow } from "../../src/engine/runtime";
|
|
|
|
const catalog = JSON.parse(
|
|
readFileSync(path.resolve(process.cwd(), "resources/catalog.json"), "utf8")
|
|
) as Catalog;
|
|
const components = new Map(catalog.components.map((component) => [component.id, component]));
|
|
const tempDirs: string[] = [];
|
|
|
|
const baseOptions = {
|
|
timeoutMs: 5_000,
|
|
retryCount: 0,
|
|
retryIntervalMs: 1_000,
|
|
skipOnError: false
|
|
};
|
|
|
|
function node(
|
|
id: string,
|
|
componentId: string,
|
|
values: Record<string, unknown> = {},
|
|
options: FlowNode["options"] = {},
|
|
extra: Partial<FlowNode> = {}
|
|
): FlowNode {
|
|
return {
|
|
id,
|
|
componentId,
|
|
position: { x: 0, y: 0 },
|
|
values,
|
|
options: { ...baseOptions, ...options },
|
|
...extra
|
|
};
|
|
}
|
|
|
|
function edge(source: string, target: string): FlowEdge {
|
|
return { id: `${source}->${target}`, source, target };
|
|
}
|
|
|
|
function flow(id: string, nodes: FlowNode[], edges: FlowEdge[] = []): FlowDefinition {
|
|
const now = new Date().toISOString();
|
|
return {
|
|
schemaVersion: 1,
|
|
id,
|
|
name: id,
|
|
variables: [],
|
|
nodes,
|
|
edges,
|
|
meta: { createdAt: now, updatedAt: now, appVersion: "test" }
|
|
};
|
|
}
|
|
|
|
function testComponent(id: string): CatalogComponent {
|
|
return {
|
|
id,
|
|
name: id,
|
|
kind: "native",
|
|
group: "test",
|
|
categories: ["test"],
|
|
categorySignature: "test",
|
|
description: "runtime integration test component",
|
|
messageTemplate: "",
|
|
baseType: id,
|
|
functionName: id,
|
|
functionCalls: [],
|
|
version: "1",
|
|
tags: [],
|
|
platformTag: null,
|
|
icon: "",
|
|
risk: "low",
|
|
fields: [],
|
|
output: { label: "", key: "", suggestedName: "", valueTypes: [], fields: [] },
|
|
slots: [],
|
|
};
|
|
}
|
|
|
|
async function tempDir(): Promise<string> {
|
|
const directory = await mkdtemp(path.join(os.tmpdir(), "rpa-runtime-integration-"));
|
|
tempDirs.push(directory);
|
|
return directory;
|
|
}
|
|
|
|
async function execute(
|
|
definition: FlowDefinition,
|
|
variables: Record<string, unknown> = {},
|
|
extraComponents: CatalogComponent[] = []
|
|
) {
|
|
const directory = await tempDir();
|
|
const events: Array<Omit<RunEvent, "runId">> = [];
|
|
const logs: Array<{ message: string; level?: "info" | "warn" | "error"; nodeId?: string }> = [];
|
|
const componentMap = new Map(components);
|
|
for (const component of extraComponents) componentMap.set(component.id, component);
|
|
|
|
const result = await runFlow(definition, {
|
|
runId: `run-${definition.id}`,
|
|
evidenceDir: directory,
|
|
signal: new AbortController().signal,
|
|
variables,
|
|
emit: (event) => events.push(event),
|
|
log: (message, level, nodeId) => logs.push({ message, level, nodeId }),
|
|
componentOf: (current) => componentMap.get(current.componentId)
|
|
});
|
|
|
|
return { result, events, logs, directory };
|
|
}
|
|
|
|
afterEach(async () => {
|
|
await Promise.all(tempDirs.splice(0).map((directory) => rm(directory, { recursive: true, force: true })));
|
|
});
|
|
|
|
describe("runFlow integration", () => {
|
|
it("follows edges in a straight line and passes variables between real capabilities", async () => {
|
|
const definition = flow(
|
|
"straight-variable-flow",
|
|
[
|
|
node("debug", "Basic_DebugOutput", { msg: "${upper}" }, {}, { position: { x: 0, y: 10 } }),
|
|
node("upper", "StrUtil_UpperCase", { string: "${source}" }, { outputVar: "upper" }),
|
|
node("trigger", "Trigger.Manual", {}, {}, { position: { x: 0, y: 300 } }),
|
|
node("set", "Basic_SetVariable", { value: "hello rpa" }, { outputVar: "source" })
|
|
],
|
|
[edge("trigger", "set"), edge("set", "upper"), edge("upper", "debug")]
|
|
);
|
|
|
|
const { result, events, logs } = await execute(definition);
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.variables).toMatchObject({ source: "hello rpa", upper: "HELLO RPA" });
|
|
expect(result.nodes.map((state) => state.nodeId)).toEqual(["trigger", "set", "upper", "debug"]);
|
|
expect(events.filter((event) => event.type === "node.started").map((event) => event.nodeId)).toEqual([
|
|
"trigger",
|
|
"set",
|
|
"upper",
|
|
"debug"
|
|
]);
|
|
expect(logs.some((entry) => entry.nodeId === "debug" && entry.message === "HELLO RPA")).toBe(true);
|
|
});
|
|
|
|
it.each([
|
|
{ approved: true, expected: "true", executed: "true-node", absent: "false-node" },
|
|
{ approved: false, expected: "false", executed: "false-node", absent: "true-node" }
|
|
])("executes only the IF $expected branch", async ({ approved, expected, executed, absent }) => {
|
|
const branch = node("branch", "logic_control_if_else", { condition: "${approved}" });
|
|
const definition = flow("if-branch-flow", [
|
|
branch,
|
|
node("true-node", "Basic_SetVariable", { value: "true" }, { outputVar: "branchResult" }, {
|
|
parentId: branch.id,
|
|
slot: "statement_if"
|
|
}),
|
|
node("false-node", "Basic_SetVariable", { value: "false" }, { outputVar: "branchResult" }, {
|
|
parentId: branch.id,
|
|
slot: "statement_else"
|
|
})
|
|
]);
|
|
|
|
const { result } = await execute(definition, { approved });
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.variables.branchResult).toBe(expected);
|
|
expect(result.nodes.find((state) => state.nodeId === executed)?.status).toBe("succeeded");
|
|
expect(result.nodes.some((state) => state.nodeId === absent)).toBe(false);
|
|
expect(result.nodes.find((state) => state.nodeId === "branch")?.output).toBe(approved);
|
|
});
|
|
|
|
it("runs a count loop inclusively and exposes the loop index to child nodes", async () => {
|
|
const loop = node("loop", "logic_control_count_for", {
|
|
count_index: "iteration",
|
|
count_start_value: "1",
|
|
count_end_value: "3",
|
|
count_step: "1"
|
|
});
|
|
const definition = flow("count-loop-flow", [
|
|
loop,
|
|
node("accumulate", "Basic_SetVariable", { value: "${total + iteration}" }, { outputVar: "total" }, {
|
|
parentId: loop.id,
|
|
slot: "count_for_fn"
|
|
})
|
|
]);
|
|
|
|
const { result, events } = await execute(definition, { total: 0 });
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.variables).toMatchObject({ iteration: 3, total: 6 });
|
|
expect(result.nodes.find((state) => state.nodeId === "loop")?.output).toBe(3);
|
|
expect(events.filter((event) => event.type === "node.started" && event.nodeId === "accumulate")).toHaveLength(3);
|
|
});
|
|
|
|
it("runs an array loop through the last item when end_index is -1", async () => {
|
|
const loop = node("loop", "logic_control_array_for_v3", {
|
|
array_name: "items",
|
|
reverse: false,
|
|
start_index: 0,
|
|
end_index: -1
|
|
});
|
|
const definition = flow("array-loop-flow", [
|
|
loop,
|
|
node("collect", "Basic_DebugOutput", { msg: "${item}" }, {}, {
|
|
parentId: loop.id,
|
|
slot: "array_for_fn"
|
|
})
|
|
]);
|
|
|
|
const { result, events } = await execute(definition, { items: ["glm", "mini", "other", "kimi"] });
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.nodes.find((state) => state.nodeId === "loop")?.output).toBe(4);
|
|
expect(events.filter((event) => event.type === "node.started" && event.nodeId === "collect").map((event) => event.nodeId)).toHaveLength(4);
|
|
});
|
|
|
|
it("retries a transient failure and records the successful attempt", async () => {
|
|
const component = testComponent("Test.TransientFailure");
|
|
let calls = 0;
|
|
register(component.id, () => {
|
|
calls += 1;
|
|
if (calls < 3) throw new Error(`temporary failure ${calls}`);
|
|
return "recovered";
|
|
});
|
|
const definition = flow("retry-flow", [
|
|
node("retry", component.id, {}, { outputVar: "retryResult", retryCount: 2, retryIntervalMs: 1_000 })
|
|
]);
|
|
|
|
const { result, logs } = await execute(definition, {}, [component]);
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(calls).toBe(3);
|
|
expect(result.variables.retryResult).toBe("recovered");
|
|
expect(result.nodes[0]).toMatchObject({ status: "succeeded", attempt: 3, output: "recovered" });
|
|
expect(logs.filter((entry) => entry.level === "warn" && entry.nodeId === "retry")).toHaveLength(2);
|
|
}, 5_000);
|
|
|
|
it("continues after a skipped failure and executes the following node", async () => {
|
|
const component = testComponent("Test.AlwaysFails");
|
|
register(component.id, () => {
|
|
throw new Error("expected failure");
|
|
});
|
|
const definition = flow(
|
|
"skip-on-error-flow",
|
|
[
|
|
node("failure", component.id, {}, { skipOnError: true }),
|
|
node("after", "Basic_SetVariable", { value: "continued" }, { outputVar: "afterFailure" })
|
|
],
|
|
[edge("failure", "after")]
|
|
);
|
|
|
|
const { result } = await execute(definition, {}, [component]);
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.nodes.find((state) => state.nodeId === "failure")).toMatchObject({
|
|
status: "skipped",
|
|
error: { code: "unknown", reason: "expected failure" }
|
|
});
|
|
expect(result.nodes.find((state) => state.nodeId === "after")?.status).toBe("succeeded");
|
|
expect(result.variables.afterFailure).toBe("continued");
|
|
});
|
|
|
|
it("runs an offline file pipeline with interpolation and persisted output", async () => {
|
|
const directory = await tempDir();
|
|
const sourcePath = path.join(directory, "source.txt");
|
|
const targetPath = path.join(directory, "result.txt");
|
|
const definition = flow(
|
|
"file-pipeline-flow",
|
|
[
|
|
node("write-source", "File_Write", {
|
|
path: "${sourcePath}",
|
|
content: "${inputText}",
|
|
encode: "utf-8",
|
|
append: "False"
|
|
}),
|
|
node("read-source", "File_Read", { path: "${sourcePath}", encode: "utf-8" }, { outputVar: "fileText" }),
|
|
node("transform", "StrUtil_UpperCase", { string: "${fileText}" }, { outputVar: "transformed" }),
|
|
node("write-target", "File_Write", {
|
|
path: "${targetPath}",
|
|
content: "${transformed}",
|
|
encode: "utf-8",
|
|
append: "False"
|
|
}),
|
|
node("read-target", "File_Read", { path: "${targetPath}", encode: "utf-8" }, { outputVar: "finalText" })
|
|
],
|
|
[
|
|
edge("write-source", "read-source"),
|
|
edge("read-source", "transform"),
|
|
edge("transform", "write-target"),
|
|
edge("write-target", "read-target")
|
|
]
|
|
);
|
|
|
|
const { result } = await execute(definition, {
|
|
sourcePath,
|
|
targetPath,
|
|
inputText: "offline rpa pipeline"
|
|
});
|
|
|
|
expect(result.status).toBe("succeeded");
|
|
expect(result.variables).toMatchObject({
|
|
fileText: "offline rpa pipeline",
|
|
transformed: "OFFLINE RPA PIPELINE",
|
|
finalText: "OFFLINE RPA PIPELINE"
|
|
});
|
|
await expect(readFile(targetPath, "utf8")).resolves.toBe("OFFLINE RPA PIPELINE");
|
|
expect(result.nodes.every((state) => state.status === "succeeded")).toBe(true);
|
|
});
|
|
});
|