Yicong-Huang commented on code in PR #5274: URL: https://github.com/apache/texera/pull/5274#discussion_r3741174850
########## agent-service/src/agent/util/workflow-utils.test.ts: ########## @@ -0,0 +1,95 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { describe, expect, test } from "bun:test"; +import { extractOperatorInputPortSchemaMap, WorkflowUtilService } from "./workflow-utils"; +import { makeMetadataFixture } from "./metadata-fixture"; +import { WorkflowState } from "../workflow-state"; +import type { OperatorPredicate, OperatorLink, OperatorPortSchemaMap } from "../../types/workflow"; + +function targetOperator(): OperatorPredicate { + return { + operatorID: "tgt", + operatorType: "Filter", + operatorVersion: "1.0", + operatorProperties: {}, + inputPorts: [{ portID: "input-0", displayName: "Input 0" }], + outputPorts: [{ portID: "output-0", displayName: "Output 0" }], + showAdvanced: false, + }; +} + +function link(): OperatorLink { + return { + linkID: "l1", + source: { operatorID: "src", portID: "output-0" }, + target: { operatorID: "tgt", portID: "input-0" }, + }; +} + +describe("extractOperatorInputPortSchemaMap", () => { + test("resolves the upstream output schema onto the matching input port", () => { + const outputSchemas: Record<string, OperatorPortSchemaMap> = { + src: { "0_false": [{ attributeName: "a", attributeType: "string" }] }, + }; + + const result = extractOperatorInputPortSchemaMap("tgt", targetOperator(), outputSchemas, [link()]); + + expect(result).toBeDefined(); + expect(result!["0_false"]).toEqual([{ attributeName: "a", attributeType: "string" }]); + }); + + test("returns undefined when the operator has no inbound links", () => { + expect(extractOperatorInputPortSchemaMap("tgt", targetOperator(), {}, [])).toBeUndefined(); + }); + + test("returns undefined when the upstream operator has no known schema", () => { + // Link exists but there is no entry for "src" in outputSchemas. Review Comment: All three cases use a single inbound link, so the function's one deliberate divergence from the frontend — "we just pick the first defined one" when several links feed a port (`workflow-utils.ts:88-92`) — is never exercised. A documented deviation is exactly what wants pinning: two links into port 0 where the first source has no known schema, asserting the second's schema wins. ########## agent-service/src/agent/tools/workflow-crud-tools.test.ts: ########## @@ -0,0 +1,241 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { describe, expect, test } from "bun:test"; +import { + createAddOperatorTool, + createModifyOperatorTool, + createDeleteOperatorTool, + type ToolContext, +} from "./workflow-crud-tools"; +import { WorkflowState } from "../workflow-state"; +import { makeMetadataFixture, FIXTURE_METADATA } from "../util/metadata-fixture"; +import type { WorkflowSystemMetadata } from "../util/workflow-system-metadata"; + +// The AI SDK tool().execute takes (args, options); the CRUD tools ignore the +// options argument, so a minimal stub is sufficient. +const EXEC_OPTS = { toolCallId: "test-call", messages: [] } as any; + +function buildOperatorSchemas(store: WorkflowSystemMetadata): Map<string, any> { + const map = new Map<string, any>(); + for (const op of FIXTURE_METADATA.operators) { + map.set(op.operatorType, { jsonSchema: store.getSchema(op.operatorType) }); + } + return map; +} + +function setup() { + const state = new WorkflowState(); + const store = makeMetadataFixture(); + const context: ToolContext = { metadataStore: store }; + const addTool = createAddOperatorTool(state, buildOperatorSchemas(store), context); + const modifyTool = createModifyOperatorTool(state, context); + const deleteTool = createDeleteOperatorTool(state, context); + return { state, addTool, modifyTool, deleteTool }; +} + +async function run(tool: any, args: unknown): Promise<string> { + return (await tool.execute(args, EXEC_OPTS)) as string; +} + +describe("addOperator tool", () => { + test("adds a valid source operator", async () => { + const { state, addTool } = setup(); + const out = await run(addTool, { + operatorId: "op1", + operatorType: "CSVFileScan", + properties: { fileName: "data.csv" }, + summary: "Load CSV", + }); + expect(out).toContain("Added operator op1"); + expect(state.getOperator("op1")).toBeDefined(); + expect(state.getOperator("op1")?.operatorProperties.fileName).toBe("data.csv"); + }); + + test("connects input links via inputOperatorIds", async () => { + const { state, addTool } = setup(); + await run(addTool, { + operatorId: "op1", + operatorType: "CSVFileScan", + properties: { fileName: "d.csv" }, + summary: "src", + }); + const out = await run(addTool, { + operatorId: "op2", + operatorType: "Filter", + properties: {}, + inputOperatorIds: { "0": ["op1"] }, + summary: "filter", + }); + expect(out).toContain("created links"); + expect(state.getAllLinks()).toHaveLength(1); + expect(state.getAllLinks()[0].source.operatorID).toBe("op1"); + expect(state.getAllLinks()[0].target.operatorID).toBe("op2"); + }); + + test("rejects an unknown operator type", async () => { + const { addTool } = setup(); + const out = await run(addTool, { operatorId: "op1", operatorType: "Nope", properties: {}, summary: "s" }); + expect(out).toContain("[ERROR]"); + expect(out).toContain("Unknown operator type"); + }); + + test("rejects properties that violate the operator schema", async () => { + const { addTool } = setup(); + // CSVFileScan requires fileName. + const out = await run(addTool, { operatorId: "op1", operatorType: "CSVFileScan", properties: {}, summary: "s" }); + expect(out).toContain("[ERROR]"); + expect(out).toContain("Invalid properties"); + }); + + test("rejects an operatorId that is not in op<N> form", async () => { + const { addTool } = setup(); + const out = await run(addTool, { + operatorId: "weird-id", + operatorType: "CSVFileScan", + properties: { fileName: "d.csv" }, + summary: "s", + }); + expect(out).toContain("[ERROR]"); + expect(out).toContain("Invalid operatorId"); + }); + + test("rejects a duplicate operatorId", async () => { + const { addTool } = setup(); + const args = { operatorId: "op1", operatorType: "CSVFileScan", properties: { fileName: "d.csv" }, summary: "s" }; + await run(addTool, args); + const out = await run(addTool, args); + expect(out).toContain("already exists"); + }); + + test("rejects a link to a non-existent source operator", async () => { + const { addTool } = setup(); + const out = await run(addTool, { + operatorId: "op1", + operatorType: "Filter", + properties: {}, + inputOperatorIds: { "0": ["ghost"] }, + summary: "s", + }); + expect(out).toContain("[ERROR]"); + expect(out).toContain('Source operator "ghost" not found'); + }); + + test("rejects an out-of-range input port index", async () => { + const { state, addTool } = setup(); + await run(addTool, { + operatorId: "op1", + operatorType: "CSVFileScan", + properties: { fileName: "d.csv" }, + summary: "s", + }); + const out = await run(addTool, { + operatorId: "op2", + operatorType: "Filter", + properties: {}, + inputOperatorIds: { "5": ["op1"] }, + summary: "s", + }); + expect(out).toContain("[ERROR]"); + expect(out).toContain("out of range"); + expect(state.getOperator("op2")).toBeDefined(); // operator was added before link validation failed Review Comment: This assertion pins a partial write. `addOperator` adds the operator at `workflow-crud-tools.ts:146` before it validates link wiring at `:153-171`, and the error return does not roll back — so a call that reports `[ERROR]` still leaves `op2` behind, and the model's natural retry then hits "already exists". The comment reads as endorsement rather than as a known defect. The modify path has the same shape and is worse: `:272-275` deletes every inbound link before its own validation at `:279-297` can fail, so a bad `inputOperatorIds` destroys the operator's existing wiring and still reports an error. Nothing covers that path. Production is out of scope for a test-only PR, so I would mark this assertion as characterization with an issue reference (AGENTS.md sanctions characterization tests) and add the modify-path case — that is where the data loss lives. ########## agent-service/src/agent/util/context-utils.test.ts: ########## @@ -0,0 +1,170 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { describe, expect, test } from "bun:test"; +import { assembleContext } from "./context-utils"; +import { WorkflowState } from "../workflow-state"; +import type { ReActStep } from "../../types/agent"; +import type { OperatorPredicate, OperatorLink } from "../../types/workflow"; + +function userStep(messageId: string, content: string): ReActStep { + return { + id: `${messageId}-u`, + messageId, + stepId: 0, + timestamp: 0, + role: "user", + content, + isBegin: true, + isEnd: true, + }; +} + +function agentStep(messageId: string, opts: Partial<ReActStep> & { stepId: number }): ReActStep { + return { + id: `${messageId}-a${opts.stepId}`, + messageId, + stepId: opts.stepId, + timestamp: 0, + role: "agent", + content: opts.content ?? "", + isBegin: false, + isEnd: opts.isEnd ?? false, + toolCalls: opts.toolCalls, + toolResults: opts.toolResults, + }; +} + +function makeOperator(id: string, overrides: Partial<OperatorPredicate> = {}): OperatorPredicate { Review Comment: These two helpers are byte-identical to `workflow-state.test.ts:24-43` (`diff` between them is empty), and `workflow-utils.test.ts:26-44` carries a third near-copy. The next required field added to `OperatorPredicate` then has three places to land, and one of them will be missed. This PR already ships the fix: `metadata-fixture.ts` exists so suites stop hand-rolling fixtures. I would move `makeOperator`/`makeLink` next to `makeMetadataFixture` and import them in all three suites. ########## agent-service/src/agent/tools/workflow-execution-tools.test.ts: ########## @@ -0,0 +1,179 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { afterEach, beforeAll, describe, expect, test } from "bun:test"; +import { executeOperatorAndFormat, type ExecutionConfig } from "./workflow-execution-tools"; +import { WorkflowState } from "../workflow-state"; +import { WorkflowUtilService } from "../util/workflow-utils"; +import { WorkflowSystemMetadata } from "../util/workflow-system-metadata"; +import { FIXTURE_METADATA } from "../util/metadata-fixture"; +import type { OperatorInfo, SyncExecutionResult } from "../../types/execution"; + +const realFetch = globalThis.fetch; +let lastUrl = ""; +let lastInit: RequestInit | undefined; + +function stubFetch(result: SyncExecutionResult, status = 200): void { + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + lastUrl = String(url); + lastInit = init; + return new Response(JSON.stringify(result), { status }); + }) as unknown as typeof fetch; +} + +// validateWorkflow / schema validation read from the process-wide singleton. +beforeAll(() => { + WorkflowSystemMetadata.getInstance().loadFromMetadata(FIXTURE_METADATA); +}); + +afterEach(() => { + globalThis.fetch = realFetch; + lastUrl = ""; + lastInit = undefined; +}); + +function csvScanState(): { state: WorkflowState; operatorId: string } { + const state = new WorkflowState(); + const util = new WorkflowUtilService(WorkflowSystemMetadata.getInstance(), state); + let op = util.getNewOperatorPredicate("CSVFileScan"); + op = { ...op, operatorProperties: { ...op.operatorProperties, fileName: "data.csv" } }; + state.addOperator(op); + return { state, operatorId: op.operatorID }; +} + +const baseConfig: ExecutionConfig = { userToken: "tok-abc", workflowId: 7, maxOperatorResultCharLimit: 2000 }; Review Comment: A 2000-char limit against 2-row fixtures means the truncation branch at `workflow-execution-tools.ts:540-569` never runs — roughly 30 lines of front/back row budgeting, the most intricate logic in the file and the easiest place for an off-by-one to hide. One case with a small limit and a dozen rows exercises all of it. ########## agent-service/src/api/workflow-api.test.ts: ########## @@ -0,0 +1,117 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { persistWorkflow, retrieveWorkflow } from "./workflow-api"; +import type { WorkflowContent } from "../types/workflow"; + +interface RecordedCall { + url: string; + init?: RequestInit; +} + +const realFetch = globalThis.fetch; +let calls: RecordedCall[] = []; + +function stubFetch(handler: (url: string, init?: RequestInit) => Response): void { + calls = []; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + calls.push({ url: String(url), init }); + return handler(String(url), init); + }) as unknown as typeof fetch; +} + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } }); +} + +const emptyContent: WorkflowContent = { + operators: [], + operatorPositions: {}, + links: [], + commentBoxes: [], + settings: { dataTransferBatchSize: 400 }, +}; + +beforeEach(() => { + calls = []; +}); + +afterEach(() => { + globalThis.fetch = realFetch; +}); + +describe("retrieveWorkflow", () => { + test("GETs the workflow by id with a Bearer token and parses string content", async () => { + stubFetch(() => jsonResponse({ wid: 5, name: "wf", content: JSON.stringify({ operators: [], links: [] }) })); + + const wf = await retrieveWorkflow("tok-123", 5); + + expect(wf.wid).toBe(5); + // content arrives as a JSON string and must be parsed into an object + expect(wf.content).toEqual({ operators: [], links: [] } as unknown as WorkflowContent); + expect(calls).toHaveLength(1); + expect(calls[0].url).toMatch(/\/api\/workflow\/5$/); + expect(calls[0].init?.method).toBe("GET"); + expect((calls[0].init?.headers as Record<string, string>).Authorization).toBe("Bearer tok-123"); + }); + + test("leaves already-parsed object content untouched", async () => { + stubFetch(() => jsonResponse({ wid: 1, name: "wf", content: { operators: [], links: [] } })); + const wf = await retrieveWorkflow("t", 1); + expect(wf.content).toEqual({ operators: [], links: [] } as unknown as WorkflowContent); + }); + + test("throws with status text on a non-ok response", async () => { + stubFetch(() => new Response("nope", { status: 404, statusText: "Not Found" })); + await expect(retrieveWorkflow("t", 99)).rejects.toThrow(/Failed to retrieve workflow: 404/); Review Comment: No suite here exercises a malformed response body, though the PR description ("positive + non-ok + malformed") and issue #5266 ("including bad/malformed responses") both claim it — the only `malformed` in the tree is auth-api's malformed-*token* test. It matters beyond bookkeeping: `data.content = JSON.parse(...)` at `workflow-api.ts:73` and `:94` throws a bare `SyntaxError` that escapes unwrapped, unlike every HTTP failure on these paths, and `texera-agent.ts:427-429` then swallows it into a warn — so the workflow silently stays stale. One test per client closes the gap and makes the description true. ########## agent-service/src/agent/util/context-utils.test.ts: ########## @@ -0,0 +1,170 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { describe, expect, test } from "bun:test"; +import { assembleContext } from "./context-utils"; +import { WorkflowState } from "../workflow-state"; +import type { ReActStep } from "../../types/agent"; +import type { OperatorPredicate, OperatorLink } from "../../types/workflow"; + +function userStep(messageId: string, content: string): ReActStep { + return { + id: `${messageId}-u`, + messageId, + stepId: 0, + timestamp: 0, + role: "user", + content, + isBegin: true, + isEnd: true, + }; +} + +function agentStep(messageId: string, opts: Partial<ReActStep> & { stepId: number }): ReActStep { + return { + id: `${messageId}-a${opts.stepId}`, + messageId, + stepId: opts.stepId, + timestamp: 0, + role: "agent", + content: opts.content ?? "", + isBegin: false, + isEnd: opts.isEnd ?? false, + toolCalls: opts.toolCalls, + toolResults: opts.toolResults, + }; +} + +function makeOperator(id: string, overrides: Partial<OperatorPredicate> = {}): OperatorPredicate { + return { + operatorID: id, + operatorType: "TestOp", + operatorVersion: "1.0", + operatorProperties: {}, + inputPorts: [{ portID: "input-0", displayName: "Input 0" }], + outputPorts: [{ portID: "output-0", displayName: "Output 0" }], + showAdvanced: false, + ...overrides, + }; +} + +function makeLink(linkId: string, sourceId: string, targetId: string): OperatorLink { + return { + linkID: linkId, + source: { operatorID: sourceId, portID: "output-0" }, + target: { operatorID: targetId, portID: "input-0" }, + }; +} + +function contentOf( + steps: ReActStep[], + state: WorkflowState, + results = new Map<string, string>(), + redact = false Review Comment: No caller passes `redact` as `true`, so `showProperties` at `context-utils.ts:230` only ever takes one branch and this parameter is effectively dead. `assembleContext`'s `compilationResult` argument is likewise never supplied, leaving the input/output-schema and compilation-error lines (`context-utils.ts:236-266`) unrendered in every test. Two cases — one redacted, one carrying a compilation result — cover both. ########## agent-service/src/agent/texera-agent.test.ts: ########## @@ -0,0 +1,149 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { describe, expect, test } from "bun:test"; +import { TexeraAgent } from "./texera-agent"; +import type { LanguageModel } from "ai"; +import { AgentState, INITIAL_STEP_ID } from "../types/agent"; + +// These tests exercise the agent's in-memory tree/settings/tool surface, which +// do not invoke the LLM, so a stub model is sufficient. The ReAct generation +// loop (sendMessage) talks to a real provider and is covered separately. Review Comment: Two fixes here: the relative clause attaches to the singular "surface", and nothing actually covers `sendMessage` — grep over every `*.test.ts` returns only this comment, so "covered separately" is not accurate. ```suggestion // These tests exercise the agent's in-memory tree/settings/tool surface, which // does not invoke the LLM, so a stub model is sufficient. The ReAct generation // loop (sendMessage) has no coverage yet. ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
