From 42490473189028e428261f25597d7a5e758564d5 Mon Sep 17 00:00:00 2001 From: mytharcher Date: Wed, 26 Jan 2022 17:46:22 +0800 Subject: [PATCH] fix(plugin-workflow): test for error job --- .../src/__tests__/execution.test.ts | 185 ++++++++++-------- .../plugin-workflow/src/__tests__/index.ts | 13 +- .../__tests__/instructions/condition.test.ts | 27 +-- .../plugin-workflow/src/models/Execution.ts | 10 +- 4 files changed, 141 insertions(+), 94 deletions(-) diff --git a/packages/plugin-workflow/src/__tests__/execution.test.ts b/packages/plugin-workflow/src/__tests__/execution.test.ts index 1d9e57ff5..ae26107cb 100644 --- a/packages/plugin-workflow/src/__tests__/execution.test.ts +++ b/packages/plugin-workflow/src/__tests__/execution.test.ts @@ -1,7 +1,6 @@ import { Application } from '@nocobase/server'; -import Database, { Model, ModelCtor } from '@nocobase/database'; +import Database from '@nocobase/database'; import { getApp } from '.'; -import { WorkflowModel } from '../models/Workflow'; import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants'; jest.setTimeout(300000); @@ -9,32 +8,32 @@ jest.setTimeout(300000); describe('execution', () => { let app: Application; let db: Database; - let PostModel: ModelCtor; - // let Target: ModelCtor; - let WorkflowModel: ModelCtor; + let PostModel; + let WorkflowModel; + let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows') as any; + WorkflowModel = db.getModel('workflows'); PostModel = db.getModel('posts'); // Target = db.getModel('targets'); + + workflow = await WorkflowModel.create({ + title: 'test workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); }); afterEach(() => db.close()); describe('base', () => { it('empty workflow without any nodes', async () => { - const workflow = await WorkflowModel.create({ - title: 'empty workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const post = await PostModel.create({ title: 't1' }); const [execution] = await workflow.getExecutions(); @@ -42,16 +41,24 @@ describe('execution', () => { expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); }); - it('workflow with single simple node', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } + it('execute resolved workflow', async () => { + await workflow.createNode({ + title: 'echo', + type: 'echo' }); + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + + expect(execution.start()).rejects.toThrow(); + expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(1); + }); + + it('workflow with single simple node', async () => { await workflow.createNode({ title: 'echo', type: 'echo' @@ -71,15 +78,6 @@ describe('execution', () => { }); it('workflow with multiple simple nodes', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'echo 1', type: 'echo' @@ -106,44 +104,27 @@ describe('execution', () => { expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); }); - it('execute resolved workflow', async () => { - const workflow = await WorkflowModel.create({ - title: 'simple workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - + it('workflow with error node', async () => { await workflow.createNode({ - title: 'echo', - type: 'echo' + title: 'error', + type: 'error' }); const post = await PostModel.create({ title: 't1' }); - - const [execution] = await workflow.getExecutions(); - expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); - expect(execution.start()).rejects.toThrow(); - expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + const jobs = await execution.getJobs(); expect(jobs.length).toEqual(1); + const { status, result } = jobs[0].get(); + expect(status).toEqual(JOB_STATUS.REJECTED); + expect(result).toBe('Error: definite error'); }); }); describe('manual nodes', () => { - it('manual node should pause execution, and could be manually resume', async () => { - const workflow = await WorkflowModel.create({ - title: 'manual workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - + it('manual node should suspend execution, and could be manually resume', async () => { const n1 = await workflow.createNode({ title: 'prompt', type: 'prompt', @@ -176,19 +157,40 @@ describe('execution', () => { expect(jobs[1].status).toEqual(JOB_STATUS.RESOLVED); expect(jobs[1].result).toEqual(123); }); + + it('manual node should suspend execution, resuming with error should end execution', async () => { + const n1 = await workflow.createNode({ + title: 'prompt error', + type: 'prompt->error', + }); + const n2 = await workflow.createNode({ + title: 'echo', + type: 'echo', + upstream_id: n1.id + }); + await n1.setDownstream(n2); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + const [pending] = await execution.getJobs(); + expect(pending.status).toEqual(JOB_STATUS.PENDING); + expect(pending.result).toEqual(null); + + pending.set('result', 123); + await execution.resume(pending); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(1); + expect(jobs[0].status).toEqual(JOB_STATUS.REJECTED); + expect(jobs[0].result).toEqual('Error: input failed'); + }); }); - describe('condition node', () => { + describe('branch: condition', () => { it('condition node link to different downstreams', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'condition', type: 'condition', @@ -222,15 +224,6 @@ describe('execution', () => { }); it('suspend downstream in condition branch, then go on', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); - const n1 = await workflow.createNode({ title: 'condition', type: 'condition', @@ -264,5 +257,41 @@ describe('execution', () => { const jobs = await execution.getJobs(); expect(jobs.length).toEqual(3); }); + + it('resume error downstream in condition branch, should reject', async () => { + const n1 = await workflow.createNode({ + title: 'condition', + type: 'condition', + // no config means always true + }); + + const n2 = await workflow.createNode({ + title: 'manual', + type: 'prompt->error', + linkType: LINK_TYPE.ON_TRUE, + upstream_id: n1.id + }); + + const n3 = await workflow.createNode({ + title: 'echo input value', + type: 'echo', + upstream_id: n1.id + }); + + await n1.setDownstream(n3); + + const post = await PostModel.create({ title: 't1' }); + + const [execution] = await workflow.getExecutions(); + expect(execution.status).toEqual(EXECUTION_STATUS.STARTED); + + const [pending] = await execution.getJobs({ node_id: n2.id }); + pending.set('result', 123); + await execution.resume(pending); + expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED); + + const jobs = await execution.getJobs(); + expect(jobs.length).toEqual(2); + }); }); }); diff --git a/packages/plugin-workflow/src/__tests__/index.ts b/packages/plugin-workflow/src/__tests__/index.ts index 15e40b047..d799652b9 100644 --- a/packages/plugin-workflow/src/__tests__/index.ts +++ b/packages/plugin-workflow/src/__tests__/index.ts @@ -2,7 +2,7 @@ import path from 'path'; import { MockServer, mockServer } from '@nocobase/test'; import plugin from '../server'; -import { InstructionResult, registerInstruction } from '../instructions'; +import { registerInstruction } from '../instructions'; import { JOB_STATUS } from '../constants'; export async function getApp(options = {}): Promise { @@ -26,6 +26,17 @@ export async function getApp(options = {}): Promise { } }); + registerInstruction('prompt->error', { + run(this, input, execution) { + return { + status: JOB_STATUS.PENDING + }; + }, + resume(this, input, execution) { + throw new Error('input failed'); + } + }); + await app.load(); app.db.import({ diff --git a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts index 1581bfb12..b9a1c1c4a 100644 --- a/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts +++ b/packages/plugin-workflow/src/__tests__/instructions/condition.test.ts @@ -1,7 +1,6 @@ import { Application } from '@nocobase/server'; -import Database, { Model, ModelCtor } from '@nocobase/database'; +import Database from '@nocobase/database'; import { getApp } from '..'; -import { WorkflowModel } from '../../models/Workflow'; import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; @@ -9,15 +8,25 @@ import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; describe('workflow > instructions > condition', () => { let app: Application; let db: Database; - let PostModel: ModelCtor; - let WorkflowModel: ModelCtor; + let PostModel; + let WorkflowModel; + let workflow; beforeEach(async () => { app = await getApp(); db = app.db; - WorkflowModel = db.getModel('workflows') as any; + WorkflowModel = db.getModel('workflows'); PostModel = db.getModel('posts'); + + workflow = await WorkflowModel.create({ + title: 'condition workflow', + enabled: true, + type: 'afterCreate', + config: { + collection: 'posts' + } + }); }); afterEach(() => db.close()); @@ -28,14 +37,6 @@ describe('workflow > instructions > condition', () => { describe('single calculation', () => { it('calculation to true downstream', async () => { - const workflow = await WorkflowModel.create({ - title: 'condition workflow', - enabled: true, - type: 'afterCreate', - config: { - collection: 'posts' - } - }); const n1 = await workflow.createNode({ title: 'condition', diff --git a/packages/plugin-workflow/src/models/Execution.ts b/packages/plugin-workflow/src/models/Execution.ts index 78ee0f29a..2398b8b57 100644 --- a/packages/plugin-workflow/src/models/Execution.ts +++ b/packages/plugin-workflow/src/models/Execution.ts @@ -73,15 +73,21 @@ export class ExecutionModel extends Model { // call instruction to get result and status job = await instruction.call(node, prevJob, this); } catch (err) { - console.error(err); // for uncaught error, set to rejected job = { - result: err, + result: err instanceof Error ? err.toString() : err, status: JOB_STATUS.REJECTED }; + // if previous job is from resuming + if (prevJob && prevJob.node_id === node.id) { + prevJob.set(job); + job = prevJob; + } } let savedJob; + // TODO(optimize): many checking of resuming or new could be improved + // could be implemented separately in exec() / resume() if (job instanceof Sequelize.Model) { savedJob = await job.save(); } else {