This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7159-a5cdabbf4eacb4599e116a9806ff70e8a604a0c8 in repository https://gitbox.apache.org/repos/asf/texera.git
commit a03782846338f0883d0c680328b35d41d9a2f7de Author: Meng Wang <[email protected]> AuthorDate: Sat Aug 1 01:24:31 2026 -0700 test(agent-service): cover WorkflowState logical-plan, traversal, validation and content methods (#7159) ### What changes were proposed in this PR? Extends the existing `WorkflowState` spec to cover the remaining pure, in-memory methods (`agent-service/src/agent/workflow-state.ts`). All are deterministic; no external services are mocked. No production code was changed. 15 tests, grouped by area: - **`toLogicalPlan`** — the produced operators, port-indexed links (indices resolved from each operator's port list), and the empty reuse list. - **Traversal** — `getFrontierOperators(depth)` returning the leaf set and expanding one hop upstream per depth (topologically ordered), and `getSubDAG` collecting a target's upstream operators and links. - **Validation state** — `setValidationError` / `clearValidationError` / `setAllValidationErrors` / `getValidationOutput`, the `workflowEmpty` flag (no operators, or all disabled), and `getValidationChangedStream` emitting on each mutation. - **`updateOperatorInputPorts`** — rebuilding the input-port list to a requested count (extras flagged dynamic), reducing the count, and the missing-operator guard. - **Content round-trip** — `setWorkflowContent` replacing the state and `getWorkflowContent` reflecting it, plus the default-settings fallback. ### Any related issues, documentation, discussions? Closes #7158 ### How was this PR tested? Extended unit tests, run locally in `agent-service/` (all green; the failure path was verified by breaking an assertion to confirm the suite goes red): ``` bun test src/agent/workflow-state.spec.ts # 28 pass, 0 fail bun run typecheck # tsc --noEmit, clean prettier --check src/agent/workflow-state.spec.ts # clean ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 4.8 [1M context]) --- agent-service/src/agent/workflow-state.spec.ts | 203 ++++++++++++++++++++++++- 1 file changed, 202 insertions(+), 1 deletion(-) diff --git a/agent-service/src/agent/workflow-state.spec.ts b/agent-service/src/agent/workflow-state.spec.ts index bffc3769eb..2fd27af652 100644 --- a/agent-service/src/agent/workflow-state.spec.ts +++ b/agent-service/src/agent/workflow-state.spec.ts @@ -19,7 +19,7 @@ import { describe, expect, test } from "bun:test"; import { WorkflowState } from "./workflow-state"; -import type { OperatorPredicate, OperatorLink } from "../types/workflow"; +import type { OperatorPredicate, OperatorLink, ValidationError, WorkflowContent } from "../types/workflow"; function makeOperator(id: string, overrides: Partial<OperatorPredicate> = {}): OperatorPredicate { return { @@ -174,3 +174,204 @@ describe("WorkflowState - getSubDAG", () => { expect(subDag.operators.map(o => o.operatorID)).toEqual(["op2"]); }); }); + +describe("WorkflowState - toLogicalPlan", () => { + test("produces operators, port-indexed links, and an empty reuse list", () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op1")); + state.addOperator(makeOperator("op2")); + state.addLink(makeLink("l1", "op1", "op2")); + + const plan = state.toLogicalPlan(); + + expect(plan.operators.map(o => o.operatorID)).toEqual(["op1", "op2"]); + // operatorProperties are spread onto each logical operator; type + ports carry through + expect(plan.operators[0].operatorType).toBe("TestOp"); + expect(plan.links).toEqual([ + { fromOpId: "op1", fromPortId: { id: 0, internal: false }, toOpId: "op2", toPortId: { id: 0, internal: false } }, + ]); + expect(plan.opsToReuseResult).toEqual([]); + }); + + test("resolves the link port indices from the operators' port lists", () => { + const state = new WorkflowState(); + state.addOperator( + makeOperator("src", { + outputPorts: [ + { portID: "output-0", displayName: "Output 0" }, + { portID: "output-1", displayName: "Output 1" }, + ], + }) + ); + state.addOperator( + makeOperator("dst", { + inputPorts: [ + { portID: "input-0", displayName: "Input 0" }, + { portID: "input-1", displayName: "Input 1" }, + ], + }) + ); + state.addLink({ + linkID: "l1", + source: { operatorID: "src", portID: "output-1" }, + target: { operatorID: "dst", portID: "input-1" }, + }); + + expect(state.toLogicalPlan().links[0]).toEqual({ + fromOpId: "src", + fromPortId: { id: 1, internal: false }, + toOpId: "dst", + toPortId: { id: 1, internal: false }, + }); + }); +}); + +describe("WorkflowState - traversal", () => { + // chain: op1 -> op2 -> op3 + function seedChain(): WorkflowState { + const state = new WorkflowState(); + state.addOperator(makeOperator("op1")); + state.addOperator(makeOperator("op2")); + state.addOperator(makeOperator("op3")); + state.addLink(makeLink("l1", "op1", "op2")); + state.addLink(makeLink("l2", "op2", "op3")); + return state; + } + + test("getFrontierOperators returns the leaf operators at depth 1", () => { + // op3 is the only operator that is not the source of a link + expect(seedChain().getFrontierOperators(1)).toEqual(["op3"]); + }); + + test("getFrontierOperators expands one hop upstream per depth, topologically ordered", () => { + const state = seedChain(); + expect(state.getFrontierOperators(2)).toEqual(["op2", "op3"]); + expect(state.getFrontierOperators(3)).toEqual(["op1", "op2", "op3"]); + }); + + test("getFrontierOperators returns an empty list for an empty workflow", () => { + expect(new WorkflowState().getFrontierOperators(3)).toEqual([]); + }); + + test("getSubDAG collects the target and every upstream operator and link", () => { + const sub = seedChain().getSubDAG("op3"); + expect(sub.operators.map(o => o.operatorID).sort()).toEqual(["op1", "op2", "op3"]); + expect(sub.links.map(l => l.linkID).sort()).toEqual(["l1", "l2"]); + }); +}); + +describe("WorkflowState - validation state", () => { + const err = (message: string): ValidationError => ({ isValid: false, messages: { general: message } }); + + test("setValidationError stores and getValidationOutput aggregates; clearValidationError removes", () => { + const state = new WorkflowState(); + + state.setValidationError("op1", err("bad")); + expect(state.getValidationOutput().errors).toEqual({ op1: err("bad") }); + + state.clearValidationError("op1"); + expect(state.getValidationOutput().errors).toEqual({}); + }); + + test("setAllValidationErrors replaces the map and recomputes the empty-state flag", () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op1")); + + state.setAllValidationErrors({ op1: err("x") }); + + const out = state.getValidationOutput(); + expect(out.errors).toEqual({ op1: err("x") }); + expect(out.workflowEmpty).toBe(false); // one enabled operator present + }); + + test("workflowEmpty is true with no operators and when every operator is disabled", () => { + const state = new WorkflowState(); + + state.setAllValidationErrors({}); + expect(state.getValidationOutput().workflowEmpty).toBe(true); + + state.addOperator(makeOperator("op1", { isDisabled: true })); + state.setAllValidationErrors({}); + expect(state.getValidationOutput().workflowEmpty).toBe(true); + }); + + test("getValidationChangedStream emits the aggregated output on every mutation", () => { + const state = new WorkflowState(); + const emitted: Array<ReturnType<WorkflowState["getValidationOutput"]>> = []; + state.getValidationChangedStream().subscribe(v => emitted.push(v)); + + state.setValidationError("op1", err("x")); + state.clearValidationError("op1"); + + expect(emitted).toHaveLength(2); + expect(emitted[0].errors).toEqual({ op1: err("x") }); + expect(emitted[1].errors).toEqual({}); + }); +}); + +describe("WorkflowState - updateOperatorInputPorts", () => { + test("rebuilds the input ports to the requested count, flagging the extras dynamic", () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("op1")); + + expect(state.updateOperatorInputPorts("op1", 3)).toBe(true); + + const ports = state.getOperator("op1")!.inputPorts; + expect(ports.map(p => p.portID)).toEqual(["input-0", "input-1", "input-2"]); + expect(ports[0].isDynamicPort).toBe(false); // port 0 is static + expect(ports[1].isDynamicPort).toBe(true); // extra ports are dynamic + }); + + test("reducing the count drops the extra ports", () => { + const state = new WorkflowState(); + state.addOperator( + makeOperator("op1", { + inputPorts: [ + { portID: "input-0", displayName: "Input 0" }, + { portID: "input-1", displayName: "Input 1" }, + ], + }) + ); + + state.updateOperatorInputPorts("op1", 1); + + expect(state.getOperator("op1")!.inputPorts.map(p => p.portID)).toEqual(["input-0"]); + }); + + test("returns false for a missing operator", () => { + expect(new WorkflowState().updateOperatorInputPorts("missing", 2)).toBe(false); + }); +}); + +describe("WorkflowState - workflow content round-trip", () => { + test("setWorkflowContent replaces the state and getWorkflowContent reflects it", () => { + const state = new WorkflowState(); + state.addOperator(makeOperator("stale")); // must be cleared by setWorkflowContent + + const content: WorkflowContent = { + operators: [makeOperator("op1"), makeOperator("op2")], + operatorPositions: { op1: { x: 1, y: 2 } }, + links: [makeLink("l1", "op1", "op2")], + commentBoxes: [], + settings: { dataTransferBatchSize: 123 }, + }; + state.setWorkflowContent(content); + + const out = state.getWorkflowContent(); + expect(out.operators.map(o => o.operatorID)).toEqual(["op1", "op2"]); + expect(out.links.map(l => l.linkID)).toEqual(["l1"]); + expect(out.operatorPositions).toEqual({ op1: { x: 1, y: 2 } }); + expect(out.settings).toEqual({ dataTransferBatchSize: 123 }); + expect(state.getOperator("stale")).toBeUndefined(); + }); + + test("setWorkflowContent falls back to defaults when settings/commentBoxes are absent", () => { + const state = new WorkflowState(); + // The runtime guards against missing optional fields even though the type requires them. + state.setWorkflowContent({ operators: [], operatorPositions: {}, links: [] } as unknown as WorkflowContent); + + const out = state.getWorkflowContent(); + expect(out.settings).toEqual({ dataTransferBatchSize: 400 }); // DEFAULT_WORKFLOW_SETTINGS + expect(out.commentBoxes).toEqual([]); + }); +});
