From 71ffb7e1a2436d7a45818dfb6a52abb5479eabda Mon Sep 17 00:00:00 2001 From: Nikhil-Doye Date: Mon, 27 Oct 2025 20:08:04 -0400 Subject: [PATCH] Add unit tests for ExecutionEngine node type validation and error handling - Introduced a new test suite for the ExecutionEngine to validate node types before workflow execution, ensuring only supported types are processed. - Implemented tests for various scenarios, including invalid node types, missing type specifications, and detailed error messages for unsupported types. - Enhanced error handling in the ExecutionEngine to provide user-friendly messages for missing or corrupted processors, improving overall robustness. - Ensured comprehensive coverage of error message quality, providing actionable feedback for users during workflow execution. --- .../executionEngine.processor.test.ts | 206 ++++++++++++++++ src/services/executionEngine.ts | 227 +++++++++++++++--- 2 files changed, 402 insertions(+), 31 deletions(-) create mode 100644 src/__tests__/services/executionEngine.processor.test.ts diff --git a/src/__tests__/services/executionEngine.processor.test.ts b/src/__tests__/services/executionEngine.processor.test.ts new file mode 100644 index 0000000..15455f3 --- /dev/null +++ b/src/__tests__/services/executionEngine.processor.test.ts @@ -0,0 +1,206 @@ +import { ExecutionEngine } from "../../services/executionEngine"; + +describe("ExecutionEngine - Processor Resolution", () => { + let engine: ExecutionEngine; + + beforeEach(() => { + engine = new ExecutionEngine(); + }); + + describe("Node Type Validation", () => { + it("should reject workflow with invalid node types", async () => { + const nodes = [ + { + id: "node-1", + type: "invalidType", + position: { x: 0, y: 0 }, + data: { + id: "node-1", + type: "invalidType", + label: "Invalid Node", + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + }, + ]; + + await expect( + engine.executeWorkflow("test-workflow", nodes, []) + ).rejects.toThrow(/Invalid node types detected/); + }); + + it("should accept workflow with all valid node types", async () => { + const validTypes = [ + "dataInput", + "dataOutput", + "webScraping", + "llmTask", + "database", + "slack", + "discord", + "gmail", + "structuredOutput", + "embeddingGenerator", + "similaritySearch", + ]; + + for (const type of validTypes) { + const nodes = [ + { + id: "node-1", + type: type, + position: { x: 0, y: 0 }, + data: { + id: "node-1", + type: type, + label: `${type} Node`, + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + }, + ]; + + // Should not throw validation error for valid types + // Note: May fail on execution for missing config, but not on validation + try { + await engine.executeWorkflow("test-workflow", nodes, []); + } catch (error) { + // Should not be a validation error about node types + expect((error as Error).message).not.toMatch( + /Invalid node types detected/ + ); + } + } + }); + + it("should provide detailed error for nodes without type", async () => { + const nodes = [ + { + id: "node-1", + position: { x: 0, y: 0 }, + data: { + id: "node-1", + label: "Typeless Node", + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + } as any, // Cast to any to allow missing type + ]; + + await expect( + engine.executeWorkflow("test-workflow", nodes, []) + ).rejects.toThrow(/has no type specified/); + }); + + it("should list supported types in error message", async () => { + const nodes = [ + { + id: "node-1", + type: "unknownType", + position: { x: 0, y: 0 }, + data: { + id: "node-1", + type: "unknownType", + label: "Unknown Node", + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + }, + ]; + + try { + await engine.executeWorkflow("test-workflow", nodes, []); + fail("Should have thrown error"); + } catch (error) { + expect((error as Error).message).toMatch(/Supported types:/); + expect((error as Error).message).toMatch(/dataInput/); + expect((error as Error).message).toMatch(/llmTask/); + } + }); + }); + + describe("Processor Import Error Handling", () => { + it("should handle missing processor gracefully", async () => { + // This test verifies the error handling for corrupted/missing processors + // In practice, this shouldn't happen as we validate node types first + + const nodes = [ + { + id: "node-1", + type: "dataInput", + position: { x: 0, y: 0 }, + data: { + id: "node-1", + type: "dataInput", + label: "Data Input", + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + }, + ]; + + // The execution should either succeed or fail with a clear error + // Not with a cryptic module resolution error + try { + await engine.executeWorkflow("test-workflow", nodes, []); + // If it succeeds, that's fine (processor exists) + } catch (error) { + // If it fails, error should be informative + const errorMessage = (error as Error).message; + + // Should not be a cryptic webpack/module error + expect(errorMessage).not.toMatch(/Cannot find module/); + expect(errorMessage).not.toMatch(/Module parse failed/); + + // Should be a user-friendly error + if (errorMessage.includes("processor")) { + expect(errorMessage).toMatch(/node|type|processor/i); + } + } + }); + }); + + describe("Error Message Quality", () => { + it("should provide actionable error messages", async () => { + const nodes = [ + { + id: "node-1", + type: "legacyNodeType", + position: { x: 0, y: 0 }, + data: { + id: "node-1", + type: "legacyNodeType", + label: "Legacy Node", + status: "idle", + config: {}, + inputs: [], + outputs: [], + }, + }, + ]; + + try { + await engine.executeWorkflow("test-workflow", nodes, []); + fail("Should have thrown error"); + } catch (error) { + const errorMessage = (error as Error).message; + + // Error should mention the problematic node + expect(errorMessage).toMatch(/legacyNodeType/); + + // Error should provide context + expect(errorMessage.length).toBeGreaterThan(50); + } + }); + }); +}); diff --git a/src/services/executionEngine.ts b/src/services/executionEngine.ts index 660a9cf..a61e167 100644 --- a/src/services/executionEngine.ts +++ b/src/services/executionEngine.ts @@ -68,6 +68,48 @@ export class ExecutionEngine { private maxConcurrentExecutions = 10; private defaultTimeout = 300000; // 5 minutes + // Validate node types before execution + private validateNodeTypes(nodes: any[]): { + isValid: boolean; + errors: string[]; + } { + const errors: string[] = []; + const supportedTypes = new Set([ + "dataInput", + "dataOutput", + "webScraping", + "llmTask", + "database", + "slack", + "discord", + "gmail", + "structuredOutput", + "embeddingGenerator", + "similaritySearch", + ]); + + nodes.forEach((node, index) => { + if (!node.type) { + errors.push( + `Node at index ${index} (id: ${ + node.id || "unknown" + }) has no type specified` + ); + } else if (!supportedTypes.has(node.type)) { + errors.push( + `Node "${node.data?.label || node.id}" has unsupported type: "${ + node.type + }". ` + `Supported types: ${Array.from(supportedTypes).join(", ")}` + ); + } + }); + + return { + isValid: errors.length === 0, + errors, + }; + } + // Main execution methods async executeWorkflow( workflowId: string, @@ -116,6 +158,30 @@ export class ExecutionEngine { }; executionEventBus.emit(executionStartEvent); + // Validate node types before execution + const nodeTypeValidation = this.validateNodeTypes(nodes); + if (!nodeTypeValidation.isValid) { + const errorMessage = `Invalid node types detected:\n${nodeTypeValidation.errors.join( + "\n" + )}`; + console.error( + "[ExecutionEngine] Node type validation failed:", + nodeTypeValidation.errors + ); + + // Emit validation error event + const validationErrorEvent: ExecutionErrorEvent = { + type: "execution:error", + executionId, + stage: "validation", + error: errorMessage, + timestamp: new Date(), + }; + executionEventBus.emit(validationErrorEvent); + + throw new Error(errorMessage); + } + // Validate workflow before execution try { // Use comprehensive validation engine @@ -900,8 +966,37 @@ export class ExecutionEngine { substitutedConfig: substitutedConfig, }); - // Import the appropriate node processor - const processor = await this.getNodeProcessor(context.nodeType); + // Import the appropriate node processor with error handling + let processor; + try { + processor = await this.getNodeProcessor(context.nodeType); + } catch (processorError) { + // Processor resolution failed - provide detailed error + const errorMessage = + processorError instanceof Error + ? processorError.message + : "Unknown processor resolution error"; + + throw new Error( + `Cannot execute node "${context.nodeId}" (type: ${context.nodeType}): ${errorMessage}\n\n` + + `Possible causes:\n` + + `- Node type "${context.nodeType}" is invalid or unsupported\n` + + `- Processor file is missing or corrupted\n` + + `- Workflow was created with an older version\n\n` + + `To fix this issue:\n` + + `1. Check if the node type is valid\n` + + `2. Remove the node and re-add it from the node library\n` + + `3. Contact support if the issue persists` + ); + } + + // Validate processor is a function + if (typeof processor !== "function") { + throw new Error( + `Invalid processor for node "${context.nodeId}" (type: ${context.nodeType}): ` + + `Processor is not a function. This indicates a corrupted processor module.` + ); + } // Execute the node const result = await processor(context, plan); @@ -978,39 +1073,109 @@ export class ExecutionEngine { } } - // Get node processor based on type + // Get node processor based on type with robust error handling private async getNodeProcessor(nodeType: string): Promise { - console.log(`Loading processor for node type: ${nodeType}`); + console.log( + `[ExecutionEngine] Loading processor for node type: ${nodeType}` + ); - // Import the appropriate processor - switch (nodeType) { - case "dataInput": - return (await import("./processors/dataInputProcessor")).default; - case "dataOutput": - return (await import("./processors/dataOutputProcessor")).default; - case "webScraping": - return (await import("./processors/webScrapingProcessor")).default; - case "llmTask": - return (await import("./processors/llmProcessor")).default; - case "database": - return (await import("./processors/unifiedDatabaseProcessor")).default; - case "slack": - return (await import("./processors/slackProcessor")).default; - case "discord": - return (await import("./processors/discordProcessor")).default; - case "gmail": - return (await import("./processors/gmailProcessor")).default; - case "structuredOutput": - return (await import("./processors/structuredOutputProcessor")).default; - case "embeddingGenerator": - return (await import("./processors/embeddingProcessor")).default; - case "similaritySearch": - return (await import("./processors/similarityProcessor")).default; - default: + // Registry of supported node types and their processor paths + const processorRegistry: Record = { + dataInput: "./processors/dataInputProcessor", + dataOutput: "./processors/dataOutputProcessor", + webScraping: "./processors/webScrapingProcessor", + llmTask: "./processors/llmProcessor", + database: "./processors/unifiedDatabaseProcessor", + slack: "./processors/slackProcessor", + discord: "./processors/discordProcessor", + gmail: "./processors/gmailProcessor", + structuredOutput: "./processors/structuredOutputProcessor", + embeddingGenerator: "./processors/embeddingProcessor", + similaritySearch: "./processors/similarityProcessor", + }; + + try { + // Check if node type is supported + const processorPath = processorRegistry[nodeType]; + + if (processorPath) { + // Attempt to import the specific processor + try { + const processor = await import(processorPath); + + if (!processor || !processor.default) { + throw new Error( + `Processor module for "${nodeType}" exists but does not export a default function` + ); + } + + console.log( + `[ExecutionEngine] ✓ Successfully loaded processor for: ${nodeType}` + ); + return processor.default; + } catch (importError) { + console.error( + `[ExecutionEngine] Failed to import processor for "${nodeType}":`, + importError + ); + + throw new Error( + `Failed to load processor for node type "${nodeType}". ` + + `This may be due to a missing or corrupted processor file. ` + + `Error: ${ + importError instanceof Error + ? importError.message + : String(importError) + }` + ); + } + } else { + // Unsupported/unknown node type - use default processor console.warn( - `Using default processor for unknown/unsupported type: ${nodeType}` + `[ExecutionEngine] ⚠ Unknown node type: "${nodeType}". ` + + `Supported types: ${Object.keys(processorRegistry).join(", ")}` ); - return (await import("./processors/defaultProcessor")).default; + + try { + const defaultProcessor = await import( + "./processors/defaultProcessor" + ); + + if (!defaultProcessor || !defaultProcessor.default) { + throw new Error("Default processor is not available"); + } + + console.log( + `[ExecutionEngine] Using default processor for: ${nodeType}` + ); + return defaultProcessor.default; + } catch (defaultImportError) { + console.error( + `[ExecutionEngine] Critical: Failed to load default processor:`, + defaultImportError + ); + + throw new Error( + `Critical error: Cannot load default processor for unknown node type "${nodeType}". ` + + `This indicates a system configuration issue. ` + + `Error: ${ + defaultImportError instanceof Error + ? defaultImportError.message + : String(defaultImportError) + }` + ); + } + } + } catch (error) { + // Re-throw with enhanced context if not already a processor error + if (error instanceof Error && error.message.includes("processor")) { + throw error; + } + + throw new Error( + `Unexpected error while resolving processor for node type "${nodeType}": ` + + `${error instanceof Error ? error.message : String(error)}` + ); } }