mirror of
https://github.com/n8n-io/n8n.git
synced 2024-12-23 11:44:06 -08:00
test: Add tests for ActiveWorkflowRunner class (#5278)
This commit is contained in:
parent
538984dc2f
commit
a2e2ec5442
|
@ -1,7 +1,6 @@
|
|||
/* eslint-disable prefer-spread */
|
||||
/* eslint-disable @typescript-eslint/no-non-null-assertion */
|
||||
/* eslint-disable no-param-reassign */
|
||||
/* eslint-disable no-console */
|
||||
/* eslint-disable no-await-in-loop */
|
||||
/* eslint-disable no-restricted-syntax */
|
||||
/* eslint-disable @typescript-eslint/no-floating-promises */
|
||||
|
@ -93,7 +92,7 @@ export class ActiveWorkflowRunner {
|
|||
// so instead of pulling all the active webhooks just pull the actives that have a trigger
|
||||
const workflowsData: IWorkflowDb[] = (await Db.collections.Workflow.find({
|
||||
where: { active: true },
|
||||
relations: ['shared', 'shared.user', 'shared.user.globalRole'],
|
||||
relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
|
||||
})) as IWorkflowDb[];
|
||||
|
||||
if (!config.getEnv('endpoints.skipWebhooksDeregistrationOnShutdown')) {
|
||||
|
@ -109,12 +108,12 @@ export class ActiveWorkflowRunner {
|
|||
}
|
||||
|
||||
if (workflowsData.length !== 0) {
|
||||
console.info(' ================================');
|
||||
console.info(' Start Active Workflows:');
|
||||
console.info(' ================================');
|
||||
Logger.info(' ================================');
|
||||
Logger.info(' Start Active Workflows:');
|
||||
Logger.info(' ================================');
|
||||
|
||||
for (const workflowData of workflowsData) {
|
||||
console.log(` - ${workflowData.name} (ID: ${workflowData.id})`);
|
||||
Logger.info(` - ${workflowData.name} (ID: ${workflowData.id})`);
|
||||
Logger.debug(`Initializing active workflow "${workflowData.name}" (startup)`, {
|
||||
workflowName: workflowData.name,
|
||||
workflowId: workflowData.id,
|
||||
|
@ -125,14 +124,14 @@ export class ActiveWorkflowRunner {
|
|||
workflowName: workflowData.name,
|
||||
workflowId: workflowData.id,
|
||||
});
|
||||
console.log(' => Started');
|
||||
Logger.info(' => Started');
|
||||
} catch (error) {
|
||||
ErrorReporter.error(error);
|
||||
console.log(
|
||||
Logger.info(
|
||||
' => ERROR: Workflow could not be activated on first try, keep on trying',
|
||||
);
|
||||
// eslint-disable-next-line @typescript-eslint/restrict-template-expressions
|
||||
console.log(` ${error.message}`);
|
||||
Logger.info(` ${error.message}`);
|
||||
Logger.error(
|
||||
`Issue on intital workflow activation try "${workflowData.name}" (startup)`,
|
||||
{
|
||||
|
@ -166,7 +165,10 @@ export class ActiveWorkflowRunner {
|
|||
}
|
||||
|
||||
const activeWorkflows = await this.getActiveWorkflows();
|
||||
activeWorkflowIds = [...activeWorkflowIds, ...activeWorkflows.map((workflow) => workflow.id)];
|
||||
activeWorkflowIds = [
|
||||
...activeWorkflowIds,
|
||||
...activeWorkflows.map((workflow) => workflow.id.toString()),
|
||||
];
|
||||
|
||||
// Make sure IDs are unique
|
||||
activeWorkflowIds = Array.from(new Set(activeWorkflowIds));
|
||||
|
@ -480,7 +482,7 @@ export class ActiveWorkflowRunner {
|
|||
try {
|
||||
await this.removeWorkflowWebhooks(workflow.id as string);
|
||||
} catch (error) {
|
||||
console.error(
|
||||
Logger.error(
|
||||
// eslint-disable-next-line @typescript-eslint/restrict-template-expressions
|
||||
`Could not remove webhooks of workflow "${workflow.id}" because of error: "${error.message}"`,
|
||||
);
|
||||
|
@ -649,7 +651,7 @@ export class ActiveWorkflowRunner {
|
|||
.catch(donePromise.reject);
|
||||
});
|
||||
} else {
|
||||
executePromise.catch(console.error);
|
||||
executePromise.catch(Logger.error);
|
||||
}
|
||||
};
|
||||
|
||||
|
@ -706,7 +708,7 @@ export class ActiveWorkflowRunner {
|
|||
.catch(donePromise.reject);
|
||||
});
|
||||
} else {
|
||||
executePromise.catch(console.error);
|
||||
executePromise.catch(Logger.error);
|
||||
}
|
||||
};
|
||||
returnFunctions.emitError = async (error: Error): Promise<void> => {
|
||||
|
@ -783,7 +785,7 @@ export class ActiveWorkflowRunner {
|
|||
if (workflowData === undefined) {
|
||||
workflowData = (await Db.collections.Workflow.findOne({
|
||||
where: { id: workflowId },
|
||||
relations: ['shared', 'shared.user', 'shared.user.globalRole'],
|
||||
relations: ['shared', 'shared.user', 'shared.user.globalRole', 'shared.role'],
|
||||
})) as IWorkflowDb;
|
||||
}
|
||||
|
||||
|
@ -811,9 +813,13 @@ export class ActiveWorkflowRunner {
|
|||
}
|
||||
|
||||
const mode = 'trigger';
|
||||
const additionalData = await WorkflowExecuteAdditionalData.getBase(
|
||||
(workflowData as WorkflowEntity).shared[0].user.id,
|
||||
const workflowOwner = (workflowData as WorkflowEntity).shared.find(
|
||||
(shared) => shared.role.name === 'owner',
|
||||
);
|
||||
if (!workflowOwner) {
|
||||
throw new Error('Workflow cannot be activated because it has no owner');
|
||||
}
|
||||
const additionalData = await WorkflowExecuteAdditionalData.getBase(workflowOwner.user.id);
|
||||
const getTriggerFunctions = this.getExecuteTriggerFunctions(
|
||||
workflowData,
|
||||
additionalData,
|
||||
|
@ -977,7 +983,7 @@ export class ActiveWorkflowRunner {
|
|||
await this.removeWorkflowWebhooks(workflowId);
|
||||
} catch (error) {
|
||||
ErrorReporter.error(error);
|
||||
console.error(
|
||||
Logger.error(
|
||||
// eslint-disable-next-line @typescript-eslint/restrict-template-expressions
|
||||
`Could not remove webhooks of workflow "${workflowId}" because of error: "${error.message}"`,
|
||||
);
|
||||
|
|
265
packages/cli/test/unit/ActiveWorkflowRunner.test.ts
Normal file
265
packages/cli/test/unit/ActiveWorkflowRunner.test.ts
Normal file
|
@ -0,0 +1,265 @@
|
|||
import { v4 as uuid } from 'uuid';
|
||||
import { mocked } from 'jest-mock';
|
||||
|
||||
import { ICredentialTypes, LoggerProxy, NodeOperationError, Workflow } from 'n8n-workflow';
|
||||
|
||||
import { ActiveWorkflowRunner } from '@/ActiveWorkflowRunner';
|
||||
import * as Db from '@/Db';
|
||||
import { WorkflowEntity } from '@/databases/entities/WorkflowEntity';
|
||||
import { SharedWorkflow } from '@/databases/entities/SharedWorkflow';
|
||||
import { Role } from '@/databases/entities/Role';
|
||||
import { User } from '@/databases/entities/User';
|
||||
import { getLogger } from '@/Logger';
|
||||
import { NodeTypes } from '@/NodeTypes';
|
||||
import { CredentialTypes } from '@/CredentialTypes';
|
||||
import { randomEmail, randomName } from '../integration/shared/random';
|
||||
import * as Helpers from './Helpers';
|
||||
import { WorkflowExecuteAdditionalData } from '@/index';
|
||||
import { WorkflowRunner } from '@/WorkflowRunner';
|
||||
|
||||
/**
|
||||
* TODO:
|
||||
* - test workflow webhooks activation (that trigger `executeWebhook`and other webhook methods)
|
||||
* - test activation error catching and getters such as `getActivationError` (requires building a workflow that fails to activate)
|
||||
* - test queued workflow activation functions (might need to create a non-working workflow to test this)
|
||||
*/
|
||||
|
||||
let databaseActiveWorkflowsCount = 0;
|
||||
let databaseActiveWorkflowsList: WorkflowEntity[] = [];
|
||||
|
||||
const generateWorkflows = (count: number): WorkflowEntity[] => {
|
||||
const workflows: WorkflowEntity[] = [];
|
||||
const ownerRole = new Role();
|
||||
ownerRole.scope = 'workflow';
|
||||
ownerRole.name = 'owner';
|
||||
ownerRole.id = '1';
|
||||
|
||||
const owner = new User();
|
||||
owner.id = uuid();
|
||||
owner.firstName = randomName();
|
||||
owner.lastName = randomName();
|
||||
owner.email = randomEmail();
|
||||
|
||||
for (let i = 0; i < count; i++) {
|
||||
const workflow = new WorkflowEntity();
|
||||
Object.assign(workflow, {
|
||||
id: i + 1,
|
||||
name: randomName(),
|
||||
active: true,
|
||||
createdAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
nodes: [
|
||||
{
|
||||
parameters: {
|
||||
rule: {
|
||||
interval: [{}],
|
||||
},
|
||||
},
|
||||
id: uuid(),
|
||||
name: 'Schedule Trigger',
|
||||
type: 'n8n-nodes-base.scheduleTrigger',
|
||||
typeVersion: 1,
|
||||
position: [900, 460],
|
||||
},
|
||||
],
|
||||
connections: {},
|
||||
tags: [],
|
||||
});
|
||||
const sharedWorkflow = new SharedWorkflow();
|
||||
sharedWorkflow.workflowId = workflow.id;
|
||||
sharedWorkflow.role = ownerRole;
|
||||
sharedWorkflow.user = owner;
|
||||
|
||||
workflow.shared = [sharedWorkflow];
|
||||
|
||||
workflows.push(workflow);
|
||||
}
|
||||
databaseActiveWorkflowsList = workflows;
|
||||
return workflows;
|
||||
};
|
||||
|
||||
const MOCK_NODE_TYPES_DATA = Helpers.mockNodeTypesData(['scheduleTrigger'], {
|
||||
addTrigger: true,
|
||||
});
|
||||
|
||||
jest.mock('@/Db', () => {
|
||||
return {
|
||||
collections: {
|
||||
Workflow: {
|
||||
find: jest.fn(async () => Promise.resolve(generateWorkflows(databaseActiveWorkflowsCount))),
|
||||
findOne: jest.fn(async (searchParams) => {
|
||||
const foundWorkflow = databaseActiveWorkflowsList.find(
|
||||
(workflow) => workflow.id.toString() === searchParams.where.id.toString(),
|
||||
);
|
||||
return Promise.resolve(foundWorkflow);
|
||||
}),
|
||||
update: jest.fn(),
|
||||
createQueryBuilder: jest.fn(() => {
|
||||
const fakeQueryBuilder = {
|
||||
update: () => fakeQueryBuilder,
|
||||
set: () => fakeQueryBuilder,
|
||||
where: () => fakeQueryBuilder,
|
||||
execute: () => Promise.resolve(),
|
||||
};
|
||||
return fakeQueryBuilder;
|
||||
}),
|
||||
},
|
||||
Webhook: {
|
||||
clear: jest.fn(),
|
||||
delete: jest.fn(),
|
||||
},
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const mockExternalHooksRunFunction = jest.fn();
|
||||
|
||||
jest.mock('@/ExternalHooks', () => {
|
||||
return {
|
||||
ExternalHooks: () => {
|
||||
return {
|
||||
run: () => mockExternalHooksRunFunction(),
|
||||
init: () => Promise.resolve(),
|
||||
};
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const workflowCheckIfCanBeActivated = jest.fn(() => true);
|
||||
|
||||
jest
|
||||
.spyOn(Workflow.prototype, 'checkIfWorkflowCanBeActivated')
|
||||
.mockImplementation(workflowCheckIfCanBeActivated);
|
||||
|
||||
const removeFunction = jest.spyOn(ActiveWorkflowRunner.prototype, 'remove');
|
||||
const removeWebhooksFunction = jest.spyOn(ActiveWorkflowRunner.prototype, 'removeWorkflowWebhooks');
|
||||
const workflowRunnerRun = jest.spyOn(WorkflowRunner.prototype, 'run');
|
||||
const workflowExecuteAdditionalDataExecuteErrorWorkflowSpy = jest.spyOn(
|
||||
WorkflowExecuteAdditionalData,
|
||||
'executeErrorWorkflow',
|
||||
);
|
||||
|
||||
describe('ActiveWorkflowRunner', () => {
|
||||
let activeWorkflowRunner: ActiveWorkflowRunner;
|
||||
|
||||
beforeAll(async () => {
|
||||
LoggerProxy.init(getLogger());
|
||||
NodeTypes({
|
||||
loaded: {
|
||||
nodes: MOCK_NODE_TYPES_DATA,
|
||||
credentials: {},
|
||||
},
|
||||
known: { nodes: {}, credentials: {} },
|
||||
credentialTypes: {} as ICredentialTypes,
|
||||
});
|
||||
CredentialTypes({
|
||||
loaded: {
|
||||
nodes: MOCK_NODE_TYPES_DATA,
|
||||
credentials: {},
|
||||
},
|
||||
known: { nodes: {}, credentials: {} },
|
||||
credentialTypes: {} as ICredentialTypes,
|
||||
});
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
activeWorkflowRunner = new ActiveWorkflowRunner();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await activeWorkflowRunner.removeAll();
|
||||
databaseActiveWorkflowsCount = 0;
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
test('Should initialize activeWorkflowRunner with empty list of active workflows and call External Hooks', async () => {
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.getActiveWorkflows()).toHaveLength(0);
|
||||
expect(mocked(Db.collections.Workflow.find)).toHaveBeenCalled();
|
||||
expect(mocked(Db.collections.Webhook.clear)).toHaveBeenCalled();
|
||||
expect(mockExternalHooksRunFunction).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('Should initialize activeWorkflowRunner with one active workflow', async () => {
|
||||
databaseActiveWorkflowsCount = 1;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.getActiveWorkflows()).toHaveLength(
|
||||
databaseActiveWorkflowsCount,
|
||||
);
|
||||
expect(mocked(Db.collections.Workflow.find)).toHaveBeenCalled();
|
||||
expect(mocked(Db.collections.Webhook.clear)).toHaveBeenCalled();
|
||||
expect(mockExternalHooksRunFunction).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('Should make sure function checkIfWorkflowCanBeActivated was called for every workflow', async () => {
|
||||
databaseActiveWorkflowsCount = 2;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(workflowCheckIfCanBeActivated).toHaveBeenCalledTimes(databaseActiveWorkflowsCount);
|
||||
});
|
||||
|
||||
test('Call to removeAll should remove every workflow', async () => {
|
||||
databaseActiveWorkflowsCount = 2;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.getActiveWorkflows()).toHaveLength(
|
||||
databaseActiveWorkflowsCount,
|
||||
);
|
||||
void (await activeWorkflowRunner.removeAll());
|
||||
expect(removeFunction).toHaveBeenCalledTimes(databaseActiveWorkflowsCount);
|
||||
});
|
||||
|
||||
test('Call to remove should also call removeWorkflowWebhooks', async () => {
|
||||
databaseActiveWorkflowsCount = 1;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.getActiveWorkflows()).toHaveLength(
|
||||
databaseActiveWorkflowsCount,
|
||||
);
|
||||
void (await activeWorkflowRunner.remove('1'));
|
||||
expect(removeWebhooksFunction).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('Call to isActive should return true for valid workflow', async () => {
|
||||
databaseActiveWorkflowsCount = 1;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.isActive('1')).toBe(true);
|
||||
});
|
||||
|
||||
test('Call to isActive should return false for invalid workflow', async () => {
|
||||
databaseActiveWorkflowsCount = 1;
|
||||
void (await activeWorkflowRunner.init());
|
||||
expect(await activeWorkflowRunner.isActive('2')).toBe(false);
|
||||
});
|
||||
|
||||
test('Calling add should call checkIfWorkflowCanBeActivated', async () => {
|
||||
// Initialize with default (0) workflows
|
||||
void (await activeWorkflowRunner.init());
|
||||
generateWorkflows(1);
|
||||
void (await activeWorkflowRunner.add('1', 'activate'));
|
||||
expect(workflowCheckIfCanBeActivated).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('runWorkflow should call run method in WorkflowRunner', async () => {
|
||||
void (await activeWorkflowRunner.init());
|
||||
const workflow = generateWorkflows(1);
|
||||
const additionalData = await WorkflowExecuteAdditionalData.getBase('fake-user-id');
|
||||
|
||||
workflowRunnerRun.mockImplementationOnce(() => Promise.resolve('invalid-execution-id'));
|
||||
|
||||
void (await activeWorkflowRunner.runWorkflow(
|
||||
workflow[0],
|
||||
workflow[0].nodes[0],
|
||||
[[]],
|
||||
additionalData,
|
||||
'trigger',
|
||||
));
|
||||
|
||||
expect(workflowRunnerRun).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('executeErrorWorkflow should call function with same name in WorkflowExecuteAdditionalData', async () => {
|
||||
const workflowData = generateWorkflows(1)[0];
|
||||
const error = new NodeOperationError(workflowData.nodes[0], 'Fake error message');
|
||||
void (await activeWorkflowRunner.init());
|
||||
activeWorkflowRunner.executeErrorWorkflow(error, workflowData, 'trigger');
|
||||
expect(workflowExecuteAdditionalDataExecuteErrorWorkflowSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
|
@ -3,6 +3,8 @@ import {
|
|||
INodeType,
|
||||
INodeTypeData,
|
||||
INodeTypes,
|
||||
ITriggerFunctions,
|
||||
ITriggerResponse,
|
||||
IVersionedNodeType,
|
||||
NodeHelpers,
|
||||
} from 'n8n-workflow';
|
||||
|
@ -41,6 +43,41 @@ class NodeTypesClass implements INodeTypes {
|
|||
},
|
||||
},
|
||||
},
|
||||
'fake-scheduler': {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: {
|
||||
displayName: 'Schedule',
|
||||
name: 'set',
|
||||
group: ['input'],
|
||||
version: 1,
|
||||
description: 'Schedules execuitons',
|
||||
defaults: {
|
||||
name: 'Set',
|
||||
color: '#0000FF',
|
||||
},
|
||||
inputs: ['main'],
|
||||
outputs: ['main'],
|
||||
properties: [
|
||||
{
|
||||
displayName: 'Value1',
|
||||
name: 'value1',
|
||||
type: 'string',
|
||||
default: 'default-value1',
|
||||
},
|
||||
{
|
||||
displayName: 'Value2',
|
||||
name: 'value2',
|
||||
type: 'string',
|
||||
default: 'default-value2',
|
||||
},
|
||||
],
|
||||
},
|
||||
trigger: () => {
|
||||
return Promise.resolve(undefined);
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
constructor(nodesAndCredentials?: INodesAndCredentials) {
|
||||
|
@ -74,3 +111,33 @@ export function NodeTypes(nodesAndCredentials?: INodesAndCredentials): NodeTypes
|
|||
* after all promises in the microtask queue have settled first.
|
||||
*/
|
||||
export const flushPromises = async () => new Promise(setImmediate);
|
||||
|
||||
export function mockNodeTypesData(
|
||||
nodeNames: string[],
|
||||
options?: {
|
||||
addTrigger?: boolean;
|
||||
},
|
||||
) {
|
||||
return nodeNames.reduce<INodeTypeData>((acc, nodeName) => {
|
||||
return (
|
||||
(acc[`n8n-nodes-base.${nodeName}`] = {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: {
|
||||
displayName: nodeName,
|
||||
name: nodeName,
|
||||
group: [],
|
||||
description: '',
|
||||
version: 1,
|
||||
defaults: {},
|
||||
inputs: [],
|
||||
outputs: [],
|
||||
properties: [],
|
||||
},
|
||||
trigger: options?.addTrigger ? () => Promise.resolve(undefined) : undefined,
|
||||
},
|
||||
}),
|
||||
acc
|
||||
);
|
||||
}, {});
|
||||
}
|
||||
|
|
|
@ -10,7 +10,7 @@ import {
|
|||
import config from '@/config';
|
||||
import * as Db from '@/Db';
|
||||
import * as testDb from '../integration/shared/testDb';
|
||||
import { NodeTypes as MockNodeTypes } from './Helpers';
|
||||
import { mockNodeTypesData, NodeTypes as MockNodeTypes } from './Helpers';
|
||||
import { UserService } from '@/user/user.service';
|
||||
import { PermissionChecker } from '@/UserManagement/PermissionChecker';
|
||||
import * as UserManagementHelper from '@/UserManagement/UserManagementHelper';
|
||||
|
@ -388,24 +388,4 @@ describe('PermissionChecker.checkSubworkflowExecutePolicy', () => {
|
|||
});
|
||||
});
|
||||
|
||||
const MOCK_NODE_TYPES_DATA = ['start', 'actionNetwork'].reduce<INodeTypeData>((acc, nodeName) => {
|
||||
return (
|
||||
(acc[`n8n-nodes-base.${nodeName}`] = {
|
||||
sourcePath: '',
|
||||
type: {
|
||||
description: {
|
||||
displayName: nodeName,
|
||||
name: nodeName,
|
||||
group: [],
|
||||
description: '',
|
||||
version: 1,
|
||||
defaults: {},
|
||||
inputs: [],
|
||||
outputs: [],
|
||||
properties: [],
|
||||
},
|
||||
},
|
||||
}),
|
||||
acc
|
||||
);
|
||||
}, {});
|
||||
const MOCK_NODE_TYPES_DATA = mockNodeTypesData(['start', 'actionNetwork']);
|
||||
|
|
Loading…
Reference in a new issue