qe-n8n-workflow-testing-fundamentals

n8n 工作流测试基础

<default_to_action> 测试 n8n 工作流时：

1. 执行前验证工作流结构
2. 使用真实的测试数据进行测试
3. 验证节点之间的数据流
4. 检查错误处理路径
5. 衡量执行性能

n8n 快速测试清单：

• 所有节点均已正确连接（无孤立节点）
• 触发器节点配置正确
• 节点之间的数据映射有效
• 已定义错误工作流
• 正确引用凭据

关键成功因素：

• 分别测试每条执行路径
• 验证每个节点的数据转换
• 检查重试和错误处理行为
• 验证与外部服务的集成 </default_to_action>

快速参考卡

适用场景

• 测试新的 n8n 工作流
• 验证工作流变更
• 调试失败的执行
• 优化性能
• 部署前验证

n8n 工作流组件

组件	用途	测试重点
触发器	启动工作流	可靠激活、负载处理
操作节点	处理数据	配置、数据映射
逻辑节点	控制流程	条件路由、分支
集成节点	外部 API	身份验证、速率限制、错误
错误工作流	处理故障	恢复、通知

工作流执行状态

状态	含义	测试操作
running	正在执行	监控进度
success	已成功完成	验证输出
failed	执行失败	分析错误
waiting	等待触发器	测试触发机制

---

工作流结构验证

// Validate workflow structure before execution
async function validateWorkflowStructure(workflowId: string) {
  const workflow = await getWorkflow(workflowId);

  // Check for trigger node
  const triggerNode = workflow.nodes.find(n =>
    n.type.includes('trigger') || n.type.includes('webhook')
  );
  if (!triggerNode) {
    throw new Error('Workflow must have a trigger node');
  }

  // Check for orphan nodes (no connections)
  const connectedNodes = new Set();
  for (const [source, targets] of Object.entries(workflow.connections)) {
    connectedNodes.add(source);
    for (const outputs of Object.values(targets)) {
      for (const connections of outputs) {
        for (const conn of connections) {
          connectedNodes.add(conn.node);
        }
      }
    }
  }

  const orphans = workflow.nodes.filter(n => !connectedNodes.has(n.name));
  if (orphans.length > 0) {
    console.warn('Orphan nodes detected:', orphans.map(n => n.name));
  }

  // Validate credentials
  for (const node of workflow.nodes) {
    if (node.credentials) {
      for (const [type, ref] of Object.entries(node.credentials)) {
        if (!await credentialExists(ref.id)) {
          throw new Error(`Missing credential: ${type} for node ${node.name}`);
        }
      }
    }
  }

  return { valid: true, orphans, triggerNode };
}

---

执行测试

// Test workflow execution with various inputs
async function testWorkflowExecution(workflowId: string, testCases: TestCase[]) {
  const results: TestResult[] = [];

  for (const testCase of testCases) {
    const startTime = Date.now();

    // Execute workflow
    const execution = await executeWorkflow(workflowId, testCase.input);

    // Wait for completion
    const result = await waitForCompletion(execution.id, testCase.timeout || 30000);

    // Validate output
    const outputValid = validateOutput(result.data, testCase.expected);

    results.push({
      testCase: testCase.name,
      success: result.status === 'success' && outputValid,
      duration: Date.now() - startTime,
      actualOutput: result.data,
      expectedOutput: testCase.expected
    });
  }

  return results;
}

// Example test cases
const testCases = [
  {
    name: 'Valid customer data',
    input: { name: 'John Doe', email: '[email protected]' },
    expected: { processed: true, customerId: /^cust_/ },
    timeout: 10000
  },
  {
    name: 'Missing email',
    input: { name: 'Jane Doe' },
    expected: { error: 'Email required' },
    timeout: 5000
  },
  {
    name: 'Invalid email format',
    input: { name: 'Bob', email: 'not-an-email' },
    expected: { error: 'Invalid email' },
    timeout: 5000
  }
];

---

数据流验证

// Trace data through workflow nodes
async function validateDataFlow(executionId: string) {
  const execution = await getExecution(executionId);
  const nodeResults = execution.data.resultData.runData;

  const dataFlow: DataFlowStep[] = [];

  for (const [nodeName, runs] of Object.entries(nodeResults)) {
    for (const run of runs) {
      dataFlow.push({
        node: nodeName,
        input: run.data?.main?.[0]?.[0]?.json || {},
        output: run.data?.main?.[0]?.[0]?.json || {},
        executionTime: run.executionTime,
        status: run.executionStatus
      });
    }
  }

  // Validate data transformations
  for (let i = 1; i < dataFlow.length; i++) {
    const prev = dataFlow[i - 1];
    const curr = dataFlow[i];

    // Check if expected data passed through
    validateDataMapping(prev.output, curr.input);
  }

  return dataFlow;
}

// Validate data mapping between nodes
function validateDataMapping(sourceOutput: any, targetInput: any) {
  // Check all required fields are present
  const missingFields: string[] = [];

  for (const [key, value] of Object.entries(targetInput)) {
    if (value === undefined && sourceOutput[key] === undefined) {
      missingFields.push(key);
    }
  }

  if (missingFields.length > 0) {
    console.warn('Missing fields in data mapping:', missingFields);
  }

  return missingFields.length === 0;
}

---

错误处理测试

// Test error handling paths
async function testErrorHandling(workflowId: string) {
  const errorScenarios = [
    {
      name: 'API timeout',
      inject: { delay: 35000 }, // Trigger timeout
      expectedError: 'timeout'
    },
    {
      name: 'Invalid data',
      inject: { invalidField: true },
      expectedError: 'validation'
    },
    {
      name: 'Missing credentials',
      inject: { removeCredentials: true },
      expectedError: 'authentication'
    }
  ];

  const results: ErrorTestResult[] = [];

  for (const scenario of errorScenarios) {
    // Execute with error injection
    const execution = await executeWithErrorInjection(workflowId, scenario.inject);

    // Check error was caught
    const result = await waitForCompletion(execution.id);

    // Validate error handling
    results.push({
      scenario: scenario.name,
      errorCaught: result.status === 'failed',
      errorType: result.data?.resultData?.error?.type,
      expectedError: scenario.expectedError,
      errorWorkflowTriggered: await checkErrorWorkflowTriggered(execution.id),
      alertSent: await checkAlertSent(execution.id)
    });
  }

  return results;
}

// Verify error workflow was triggered
async function checkErrorWorkflowTriggered(executionId: string): Promise<boolean> {
  const errorExecutions = await getExecutions({
    filter: {
      metadata: { errorTriggeredBy: executionId }
    }
  });

  return errorExecutions.length > 0;
}

---

节点连接模式

线性流程

Trigger → Process → Transform → Output

测试： 执行一次，验证每个节点的输出

分支流程

Trigger → IF → [Branch A] → Merge → Output
              → [Branch B] →

测试： 分别测试两个分支，验证合并行为

并行流程

Trigger → Split → [Process A] → Merge → Output
                → [Process B] →

测试： 验证并行执行，检查合并时机

循环流程

Trigger → SplitInBatches → Process → [Loop back until done] → Output

测试： 使用不同的批次大小进行测试，验证所有条目均已处理

---

常见测试模式

测试数据生成

// Generate test data for common n8n patterns
const testDataGenerators = {
  webhook: () => ({
    body: { event: 'test', timestamp: new Date().toISOString() },
    headers: { 'Content-Type': 'application/json' },
    query: { source: 'test' }
  }),

  slack: () => ({
    type: 'message',
    channel: 'C123456',
    user: 'U789012',
    text: 'Test message'
  }),

  github: () => ({
    action: 'opened',
    issue: {
      number: 1,
      title: 'Test Issue',
      body: 'Test body'
    },
    repository: {
      full_name: 'test/repo'
    }
  }),

  stripe: () => ({
    type: 'payment_intent.succeeded',
    data: {
      object: {
        id: 'pi_test123',
        amount: 1000,
        currency: 'usd'
      }
    }
  })
};

执行断言

// Common assertions for workflow execution
const workflowAssertions = {
  // Assert workflow completed
  assertCompleted: (execution) => {
    expect(execution.finished).toBe(true);
    expect(execution.status).toBe('success');
  },

  // Assert specific node executed
  assertNodeExecuted: (execution, nodeName) => {
    const nodeData = execution.data.resultData.runData[nodeName];
    expect(nodeData).toBeDefined();
    expect(nodeData[0].executionStatus).toBe('success');
  },

  // Assert data transformation
  assertDataTransformed: (execution, nodeName, expectedData) => {
    const nodeOutput = execution.data.resultData.runData[nodeName][0].data.main[0][0].json;
    expect(nodeOutput).toMatchObject(expectedData);
  },

  // Assert execution time
  assertExecutionTime: (execution, maxMs) => {
    const duration = new Date(execution.stoppedAt) - new Date(execution.startedAt);
    expect(duration).toBeLessThan(maxMs);
  }
};

---

Agent 协作提示

内存命名空间

aqe/n8n/
├── workflows/*          - Cached workflow definitions
├── test-results/*       - Test execution results
├── validations/*        - Validation reports
├── patterns/*           - Discovered testing patterns
└── executions/*         - Execution tracking

Agent 集群协作

// Comprehensive n8n testing with fleet
const n8nFleet = await FleetManager.coordinate({
  strategy: 'n8n-testing',
  agents: [
    'n8n-workflow-executor',  // Execute and validate
    'n8n-node-validator',     // Validate configurations
    'n8n-trigger-test',       // Test triggers
    'n8n-expression-validator', // Validate expressions
    'n8n-integration-test'    // Test integrations
  ],
  topology: 'parallel'
});

---

相关技能

• n8n-expression-testing - 表达式验证
• n8n-trigger-testing-strategies - 触发器测试
• n8n-integration-testing-patterns - 集成测试
• n8n-security-testing - 安全验证

---

请记住

n8n 工作流是基于 JSON 的执行流，用于连接 400 多种服务。测试需要验证：

• 工作流结构（节点、连接）
• 触发器可靠性（Webhook、计划任务）
• 数据流（节点之间的转换）
• 错误处理（重试、回退、通知）
• 性能（执行时间、资源使用情况）

使用 Agent 时： 使用 n8n-workflow-executor 进行执行测试，使用 n8n-node-validator 验证配置，并协调多个 Agent 全面测试工作流。