fix(plugin-workflow): test for error job

This commit is contained in:
mytharcher 2022-01-26 17:46:22 +08:00
parent 6018013195
commit 4249047318
4 changed files with 141 additions and 94 deletions

View File

@ -1,7 +1,6 @@
import { Application } from '@nocobase/server'; import { Application } from '@nocobase/server';
import Database, { Model, ModelCtor } from '@nocobase/database'; import Database from '@nocobase/database';
import { getApp } from '.'; import { getApp } from '.';
import { WorkflowModel } from '../models/Workflow';
import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants'; import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../constants';
jest.setTimeout(300000); jest.setTimeout(300000);
@ -9,32 +8,32 @@ jest.setTimeout(300000);
describe('execution', () => { describe('execution', () => {
let app: Application; let app: Application;
let db: Database; let db: Database;
let PostModel: ModelCtor<Model>; let PostModel;
// let Target: ModelCtor<Model>; let WorkflowModel;
let WorkflowModel: ModelCtor<WorkflowModel>; let workflow;
beforeEach(async () => { beforeEach(async () => {
app = await getApp(); app = await getApp();
db = app.db; db = app.db;
WorkflowModel = db.getModel('workflows') as any; WorkflowModel = db.getModel('workflows');
PostModel = db.getModel('posts'); PostModel = db.getModel('posts');
// Target = db.getModel('targets'); // Target = db.getModel('targets');
workflow = await WorkflowModel.create({
title: 'test workflow',
enabled: true,
type: 'afterCreate',
config: {
collection: 'posts'
}
});
}); });
afterEach(() => db.close()); afterEach(() => db.close());
describe('base', () => { describe('base', () => {
it('empty workflow without any nodes', async () => { 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 post = await PostModel.create({ title: 't1' });
const [execution] = await workflow.getExecutions(); const [execution] = await workflow.getExecutions();
@ -42,16 +41,24 @@ describe('execution', () => {
expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
}); });
it('workflow with single simple node', async () => { it('execute resolved workflow', async () => {
const workflow = await WorkflowModel.create({ await workflow.createNode({
title: 'simple workflow', title: 'echo',
enabled: true, type: 'echo'
type: 'afterCreate',
config: {
collection: 'posts'
}
}); });
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({ await workflow.createNode({
title: 'echo', title: 'echo',
type: 'echo' type: 'echo'
@ -71,15 +78,6 @@ describe('execution', () => {
}); });
it('workflow with multiple simple nodes', async () => { 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({ const n1 = await workflow.createNode({
title: 'echo 1', title: 'echo 1',
type: 'echo' type: 'echo'
@ -106,44 +104,27 @@ describe('execution', () => {
expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) }); expect(result).toMatchObject({ data: JSON.parse(JSON.stringify(post.toJSON())) });
}); });
it('execute resolved workflow', async () => { it('workflow with error node', async () => {
const workflow = await WorkflowModel.create({
title: 'simple workflow',
enabled: true,
type: 'afterCreate',
config: {
collection: 'posts'
}
});
await workflow.createNode({ await workflow.createNode({
title: 'echo', title: 'error',
type: 'echo' type: 'error'
}); });
const post = await PostModel.create({ title: 't1' }); const post = await PostModel.create({ title: 't1' });
const [execution] = await workflow.getExecutions(); const [execution] = await workflow.getExecutions();
expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED); expect(execution.status).toEqual(EXECUTION_STATUS.REJECTED);
expect(execution.start()).rejects.toThrow();
expect(execution.status).toEqual(EXECUTION_STATUS.RESOLVED);
const jobs = await execution.getJobs(); const jobs = await execution.getJobs();
expect(jobs.length).toEqual(1); 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', () => { describe('manual nodes', () => {
it('manual node should pause execution, and could be manually resume', async () => { it('manual node should suspend execution, and could be manually resume', async () => {
const workflow = await WorkflowModel.create({
title: 'manual workflow',
enabled: true,
type: 'afterCreate',
config: {
collection: 'posts'
}
});
const n1 = await workflow.createNode({ const n1 = await workflow.createNode({
title: 'prompt', title: 'prompt',
type: 'prompt', type: 'prompt',
@ -176,19 +157,40 @@ describe('execution', () => {
expect(jobs[1].status).toEqual(JOB_STATUS.RESOLVED); expect(jobs[1].status).toEqual(JOB_STATUS.RESOLVED);
expect(jobs[1].result).toEqual(123); 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 () => { 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({ const n1 = await workflow.createNode({
title: 'condition', title: 'condition',
type: 'condition', type: 'condition',
@ -222,15 +224,6 @@ describe('execution', () => {
}); });
it('suspend downstream in condition branch, then go on', async () => { 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({ const n1 = await workflow.createNode({
title: 'condition', title: 'condition',
type: 'condition', type: 'condition',
@ -264,5 +257,41 @@ describe('execution', () => {
const jobs = await execution.getJobs(); const jobs = await execution.getJobs();
expect(jobs.length).toEqual(3); 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);
});
}); });
}); });

View File

@ -2,7 +2,7 @@ import path from 'path';
import { MockServer, mockServer } from '@nocobase/test'; import { MockServer, mockServer } from '@nocobase/test';
import plugin from '../server'; import plugin from '../server';
import { InstructionResult, registerInstruction } from '../instructions'; import { registerInstruction } from '../instructions';
import { JOB_STATUS } from '../constants'; import { JOB_STATUS } from '../constants';
export async function getApp(options = {}): Promise<MockServer> { export async function getApp(options = {}): Promise<MockServer> {
@ -26,6 +26,17 @@ export async function getApp(options = {}): Promise<MockServer> {
} }
}); });
registerInstruction('prompt->error', {
run(this, input, execution) {
return {
status: JOB_STATUS.PENDING
};
},
resume(this, input, execution) {
throw new Error('input failed');
}
});
await app.load(); await app.load();
app.db.import({ app.db.import({

View File

@ -1,7 +1,6 @@
import { Application } from '@nocobase/server'; import { Application } from '@nocobase/server';
import Database, { Model, ModelCtor } from '@nocobase/database'; import Database from '@nocobase/database';
import { getApp } from '..'; import { getApp } from '..';
import { WorkflowModel } from '../../models/Workflow';
import { EXECUTION_STATUS, JOB_STATUS, LINK_TYPE } from '../../constants'; 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', () => { describe('workflow > instructions > condition', () => {
let app: Application; let app: Application;
let db: Database; let db: Database;
let PostModel: ModelCtor<Model>; let PostModel;
let WorkflowModel: ModelCtor<WorkflowModel>; let WorkflowModel;
let workflow;
beforeEach(async () => { beforeEach(async () => {
app = await getApp(); app = await getApp();
db = app.db; db = app.db;
WorkflowModel = db.getModel('workflows') as any; WorkflowModel = db.getModel('workflows');
PostModel = db.getModel('posts'); PostModel = db.getModel('posts');
workflow = await WorkflowModel.create({
title: 'condition workflow',
enabled: true,
type: 'afterCreate',
config: {
collection: 'posts'
}
});
}); });
afterEach(() => db.close()); afterEach(() => db.close());
@ -28,14 +37,6 @@ describe('workflow > instructions > condition', () => {
describe('single calculation', () => { describe('single calculation', () => {
it('calculation to true downstream', async () => { 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({ const n1 = await workflow.createNode({
title: 'condition', title: 'condition',

View File

@ -73,15 +73,21 @@ export class ExecutionModel extends Model {
// call instruction to get result and status // call instruction to get result and status
job = await instruction.call(node, prevJob, this); job = await instruction.call(node, prevJob, this);
} catch (err) { } catch (err) {
console.error(err);
// for uncaught error, set to rejected // for uncaught error, set to rejected
job = { job = {
result: err, result: err instanceof Error ? err.toString() : err,
status: JOB_STATUS.REJECTED status: JOB_STATUS.REJECTED
}; };
// if previous job is from resuming
if (prevJob && prevJob.node_id === node.id) {
prevJob.set(job);
job = prevJob;
}
} }
let savedJob; 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) { if (job instanceof Sequelize.Model) {
savedJob = await job.save(); savedJob = await job.save();
} else { } else {