import config from '@/config'; import axios from 'axios'; import syslog from 'syslog-client'; import { v4 as uuid } from 'uuid'; import type { SuperAgentTest } from 'supertest'; import type { Role } from '@db/entities/Role'; import type { User } from '@db/entities/User'; import type { MessageEventBusDestinationSentryOptions, MessageEventBusDestinationSyslogOptions, MessageEventBusDestinationWebhookOptions, } from 'n8n-workflow'; import { defaultMessageEventBusDestinationSentryOptions, defaultMessageEventBusDestinationSyslogOptions, defaultMessageEventBusDestinationWebhookOptions, } from 'n8n-workflow'; import { eventBus } from '@/eventbus'; import { EventMessageGeneric } from '@/eventbus/EventMessageClasses/EventMessageGeneric'; import type { MessageEventBusDestinationSyslog } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationSyslog.ee'; import type { MessageEventBusDestinationWebhook } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationWebhook.ee'; import type { MessageEventBusDestinationSentry } from '@/eventbus/MessageEventBusDestination/MessageEventBusDestinationSentry.ee'; import { EventMessageAudit } from '@/eventbus/EventMessageClasses/EventMessageAudit'; import type { EventNamesTypes } from '@/eventbus/EventMessageClasses'; import { EventMessageWorkflow } from '@/eventbus/EventMessageClasses/EventMessageWorkflow'; import { EventMessageNode } from '@/eventbus/EventMessageClasses/EventMessageNode'; import * as utils from './shared/utils'; import { getGlobalOwnerRole } from './shared/db/roles'; import { createUser } from './shared/db/users'; jest.unmock('@/eventbus/MessageEventBus/MessageEventBus'); jest.mock('axios'); const mockedAxios = axios as jest.Mocked; jest.mock('syslog-client'); const mockedSyslog = syslog as jest.Mocked; let globalOwnerRole: Role; let owner: User; let authOwnerAgent: SuperAgentTest; const testSyslogDestination: MessageEventBusDestinationSyslogOptions = { ...defaultMessageEventBusDestinationSyslogOptions, id: 'b88038f4-0a89-4e94-89a9-658dfdb74539', protocol: 'udp', label: 'Test Syslog', enabled: false, subscribedEvents: ['n8n.test.message', 'n8n.audit.user.updated'], }; const testWebhookDestination: MessageEventBusDestinationWebhookOptions = { ...defaultMessageEventBusDestinationWebhookOptions, id: '88be6560-bfb4-455c-8aa1-06971e9e5522', url: 'http://localhost:3456', method: 'POST', label: 'Test Webhook', enabled: false, subscribedEvents: ['n8n.test.message', 'n8n.audit.user.updated'], }; const testSentryDestination: MessageEventBusDestinationSentryOptions = { ...defaultMessageEventBusDestinationSentryOptions, id: '450ca04b-87dd-4837-a052-ab3a347a00e9', dsn: 'http://localhost:3000', label: 'Test Sentry', enabled: false, subscribedEvents: ['n8n.test.message', 'n8n.audit.user.updated'], }; async function confirmIdInAll(id: string) { const sent = await eventBus.getEventsAll(); expect(sent.length).toBeGreaterThan(0); expect(sent.find((msg) => msg.id === id)).toBeTruthy(); } async function confirmIdSent(id: string) { const sent = await eventBus.getEventsSent(); expect(sent.length).toBeGreaterThan(0); expect(sent.find((msg) => msg.id === id)).toBeTruthy(); } const testServer = utils.setupTestServer({ endpointGroups: ['eventBus'], enabledFeatures: ['feat:logStreaming'], }); beforeAll(async () => { globalOwnerRole = await getGlobalOwnerRole(); owner = await createUser({ globalRole: globalOwnerRole }); authOwnerAgent = testServer.authAgentFor(owner); mockedSyslog.createClient.mockImplementation(() => new syslog.Client()); config.set('eventBus.logWriter.logBaseName', 'n8n-test-logwriter'); config.set('eventBus.logWriter.keepLogCount', 1); await eventBus.initialize({}); }); afterAll(async () => { jest.mock('@/eventbus/MessageEventBus/MessageEventBus'); await eventBus.close(); }); test('should have a running logwriter process', () => { const thread = eventBus.logWriter.worker; expect(thread).toBeDefined(); }); test('should have logwriter log messages', async () => { const testMessage = new EventMessageGeneric({ eventName: 'n8n.test.message' as EventNamesTypes, id: uuid(), }); await eventBus.send(testMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.once('message', async (msg: { command: string; data: any }) => { expect(msg.command).toBe('appendMessageToLog'); expect(msg.data).toBe(true); await confirmIdInAll(testMessage.id); resolve(true); }); }); }); describe('GET /eventbus/destination', () => { test('should fail due to missing authentication', async () => { const response = await testServer.authlessAgent.get('/eventbus/destination'); expect(response.statusCode).toBe(401); }); test('all returned destinations should exist in eventbus', async () => { const response = await authOwnerAgent.get('/eventbus/destination'); expect(response.statusCode).toBe(200); const data = response.body.data; expect(data).toBeTruthy(); expect(Array.isArray(data)).toBeTruthy(); for (let index = 0; index < data.length; index++) { const destination = data[index]; const foundDestinations = await eventBus.findDestination(destination.id); expect(Array.isArray(foundDestinations)).toBeTruthy(); expect(foundDestinations.length).toBe(1); expect(foundDestinations[0].label).toBe(destination.label); } }); }); describe('POST /eventbus/destination', () => { test('create syslog destination', async () => { const response = await authOwnerAgent.post('/eventbus/destination').send(testSyslogDestination); expect(response.statusCode).toBe(200); }); test('create sentry destination', async () => { const response = await authOwnerAgent.post('/eventbus/destination').send(testSentryDestination); expect(response.statusCode).toBe(200); }); test('create webhook destination', async () => { const response = await authOwnerAgent .post('/eventbus/destination') .send(testWebhookDestination); expect(response.statusCode).toBe(200); }); }); // this test (presumably the mocking) is causing the test suite to randomly fail // eslint-disable-next-line n8n-local-rules/no-skipped-tests test.skip('should send message to syslog', async () => { const testMessage = new EventMessageGeneric({ eventName: 'n8n.test.message' as EventNamesTypes, id: uuid(), }); const syslogDestination = eventBus.destinations[ testSyslogDestination.id! ] as MessageEventBusDestinationSyslog; syslogDestination.enable(); const mockedSyslogClientLog = jest.spyOn(syslogDestination.client, 'log'); mockedSyslogClientLog.mockImplementation((_m, _options, _cb) => { eventBus.confirmSent(testMessage, { id: syslogDestination.id, name: syslogDestination.label, }); return syslogDestination.client; }); await eventBus.send(testMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler001(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { await confirmIdInAll(testMessage.id); } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessage.id); expect(mockedSyslogClientLog).toHaveBeenCalled(); syslogDestination.disable(); eventBus.logWriter.worker?.removeListener('message', handler001); resolve(true); } }, ); }); }); // eslint-disable-next-line n8n-local-rules/no-skipped-tests test.skip('should confirm send message if there are no subscribers', async () => { const testMessageUnsubscribed = new EventMessageGeneric({ eventName: 'n8n.test.unsub' as EventNamesTypes, id: uuid(), }); const syslogDestination = eventBus.destinations[ testSyslogDestination.id! ] as MessageEventBusDestinationSyslog; syslogDestination.enable(); await eventBus.send(testMessageUnsubscribed); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler002(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { await confirmIdInAll(testMessageUnsubscribed.id); } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessageUnsubscribed.id); syslogDestination.disable(); eventBus.logWriter.worker?.removeListener('message', handler002); resolve(true); } }, ); }); }); test('should anonymize audit message to syslog ', async () => { const testAuditMessage = new EventMessageAudit({ eventName: 'n8n.audit.user.updated', payload: { _secret: 'secret', public: 'public', }, id: uuid(), }); const syslogDestination = eventBus.destinations[ testSyslogDestination.id! ] as MessageEventBusDestinationSyslog; syslogDestination.enable(); const mockedSyslogClientLog = jest.spyOn(syslogDestination.client, 'log'); mockedSyslogClientLog.mockImplementation((m, _options, _cb) => { const o = JSON.parse(m); expect(o).toHaveProperty('payload'); expect(o.payload).toHaveProperty('_secret'); syslogDestination.anonymizeAuditMessages ? expect(o.payload._secret).toBe('*') : expect(o.payload._secret).toBe('secret'); expect(o.payload).toHaveProperty('public'); expect(o.payload.public).toBe('public'); return syslogDestination.client; }); syslogDestination.anonymizeAuditMessages = true; await eventBus.send(testAuditMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler005(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { const sent = await eventBus.getEventsAll(); await confirmIdInAll(testAuditMessage.id); expect(mockedSyslogClientLog).toHaveBeenCalled(); eventBus.logWriter.worker?.removeListener('message', handler005); resolve(true); } }, ); }); syslogDestination.anonymizeAuditMessages = false; await eventBus.send(testAuditMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler006(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { const sent = await eventBus.getEventsAll(); await confirmIdInAll(testAuditMessage.id); expect(mockedSyslogClientLog).toHaveBeenCalled(); syslogDestination.disable(); eventBus.logWriter.worker?.removeListener('message', handler006); resolve(true); } }, ); }); }); test('should send message to webhook ', async () => { const testMessage = new EventMessageGeneric({ eventName: 'n8n.test.message' as EventNamesTypes, id: uuid(), }); const webhookDestination = eventBus.destinations[ testWebhookDestination.id! ] as MessageEventBusDestinationWebhook; webhookDestination.enable(); mockedAxios.post.mockResolvedValue({ status: 200, data: { msg: 'OK' } }); mockedAxios.request.mockResolvedValue({ status: 200, data: { msg: 'OK' } }); await eventBus.send(testMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler003(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { await confirmIdInAll(testMessage.id); } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessage.id); expect(mockedAxios.request).toHaveBeenCalled(); webhookDestination.disable(); eventBus.logWriter.worker?.removeListener('message', handler003); resolve(true); } }, ); }); }); test('should send message to sentry ', async () => { const testMessage = new EventMessageGeneric({ eventName: 'n8n.test.message' as EventNamesTypes, id: uuid(), }); const sentryDestination = eventBus.destinations[ testSentryDestination.id! ] as MessageEventBusDestinationSentry; sentryDestination.enable(); const mockedSentryCaptureMessage = jest.spyOn(sentryDestination.sentryClient!, 'captureMessage'); mockedSentryCaptureMessage.mockImplementation((_m, _level, _hint, _scope) => { eventBus.confirmSent(testMessage, { id: sentryDestination.id, name: sentryDestination.label, }); return testMessage.id; }); await eventBus.send(testMessage); await new Promise((resolve) => { eventBus.logWriter.worker?.on( 'message', async function handler004(msg: { command: string; data: any }) { if (msg.command === 'appendMessageToLog') { await confirmIdInAll(testMessage.id); } else if (msg.command === 'confirmMessageSent') { await confirmIdSent(testMessage.id); expect(mockedSentryCaptureMessage).toHaveBeenCalled(); sentryDestination.disable(); eventBus.logWriter.worker?.removeListener('message', handler004); resolve(true); } }, ); }); }); test('DELETE /eventbus/destination delete all destinations by id', async () => { const existingDestinationIds = [...Object.keys(eventBus.destinations)]; await Promise.all( existingDestinationIds.map(async (id) => { const response = await authOwnerAgent.del('/eventbus/destination').query({ id }); expect(response.statusCode).toBe(200); }), ); expect(Object.keys(eventBus.destinations).length).toBe(0); }); // These two tests are running very flaky on CI due to the logwriter working in a worker // Mocking everything on the other would defeat the purpose of even testing them... so, skipping in CI for now. // eslint-disable-next-line n8n-local-rules/no-skipped-tests test.skip('should not find unfinished executions in recovery process', async () => { eventBus.logWriter?.putMessage( new EventMessageWorkflow({ eventName: 'n8n.workflow.started', payload: { executionId: '509', isManual: false }, }), ); eventBus.logWriter?.putMessage( new EventMessageNode({ eventName: 'n8n.node.started', payload: { executionId: '509', nodeName: 'Set', workflowName: 'test' }, }), ); eventBus.logWriter?.putMessage( new EventMessageNode({ eventName: 'n8n.node.finished', payload: { executionId: '509', nodeName: 'Set', workflowName: 'test' }, }), ); eventBus.logWriter?.putMessage( new EventMessageWorkflow({ eventName: 'n8n.workflow.success', payload: { executionId: '509', success: true }, }), ); const unfinishedExecutions = await eventBus.getUnfinishedExecutions(); expect(Object.keys(unfinishedExecutions)).toHaveLength(0); }); // eslint-disable-next-line n8n-local-rules/no-skipped-tests test.skip('should not find unfinished executions in recovery process', async () => { eventBus.logWriter?.putMessage( new EventMessageWorkflow({ eventName: 'n8n.workflow.started', payload: { executionId: '510', isManual: false }, }), ); eventBus.logWriter?.putMessage( new EventMessageNode({ eventName: 'n8n.node.started', payload: { executionId: '510', nodeName: 'Set', workflowName: 'test' }, }), ); const unfinishedExecutions = await eventBus.getUnfinishedExecutions(); expect(Object.keys(unfinishedExecutions)).toHaveLength(1); expect(Object.keys(unfinishedExecutions)).toContain('510'); });