mirror of
https://github.com/n8n-io/n8n.git
synced 2026-05-28 07:17:04 +02:00
391 lines
13 KiB
TypeScript
391 lines
13 KiB
TypeScript
import { mockInstance } from '@n8n/backend-test-utils';
|
|
import { GLOBAL_OWNER_ROLE, type User } from '@n8n/db';
|
|
import { Container } from '@n8n/di';
|
|
import axios from 'axios';
|
|
import { mock } from 'jest-mock-extended';
|
|
import type {
|
|
MessageEventBusDestinationSentryOptions,
|
|
MessageEventBusDestinationSyslogOptions,
|
|
MessageEventBusDestinationWebhookOptions,
|
|
} from 'n8n-workflow';
|
|
import {
|
|
defaultMessageEventBusDestinationSentryOptions,
|
|
defaultMessageEventBusDestinationSyslogOptions,
|
|
defaultMessageEventBusDestinationWebhookOptions,
|
|
} from 'n8n-workflow';
|
|
import { v4 as uuid } from 'uuid';
|
|
|
|
import type { EventNamesTypes } from '@/eventbus/event-message-classes';
|
|
import { EventMessageAudit } from '@/eventbus/event-message-classes/event-message-audit';
|
|
import { EventMessageGeneric } from '@/eventbus/event-message-classes/event-message-generic';
|
|
import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
|
|
import { ExecutionRecoveryService } from '@/executions/execution-recovery.service';
|
|
import type { MessageEventBusDestinationSentry } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-sentry.ee';
|
|
import type { MessageEventBusDestinationSyslog } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-syslog.ee';
|
|
import type { MessageEventBusDestinationWebhook } from '@/modules/log-streaming.ee/destinations/message-event-bus-destination-webhook.ee';
|
|
import { LogStreamingDestinationService } from '@/modules/log-streaming.ee/log-streaming-destination.service';
|
|
import { Publisher } from '@/scaling/pubsub/publisher.service';
|
|
|
|
import { createUser } from './shared/db/users';
|
|
import type { SuperAgentTest } from './shared/types';
|
|
import * as utils from './shared/utils';
|
|
|
|
jest.unmock('@/eventbus/message-event-bus/message-event-bus');
|
|
jest.mock('axios');
|
|
|
|
const mockAxiosInstance = mock<ReturnType<typeof axios.create>>();
|
|
const mockedAxios = axios as jest.Mocked<typeof axios>;
|
|
mockedAxios.create.mockReturnValue(mockAxiosInstance);
|
|
|
|
mockInstance(Publisher);
|
|
|
|
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'],
|
|
};
|
|
|
|
let eventBus: MessageEventBus;
|
|
let destinationService: LogStreamingDestinationService;
|
|
|
|
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();
|
|
}
|
|
|
|
mockInstance(ExecutionRecoveryService);
|
|
const testServer = utils.setupTestServer({
|
|
endpointGroups: ['eventBus'],
|
|
enabledFeatures: ['feat:logStreaming'],
|
|
modules: ['log-streaming'],
|
|
});
|
|
|
|
beforeAll(async () => {
|
|
owner = await createUser({ role: GLOBAL_OWNER_ROLE });
|
|
authOwnerAgent = testServer.authAgentFor(owner);
|
|
|
|
eventBus = Container.get(MessageEventBus);
|
|
await eventBus.initialize();
|
|
|
|
destinationService = Container.get(LogStreamingDestinationService);
|
|
await destinationService.initialize();
|
|
});
|
|
|
|
afterAll(async () => {
|
|
jest.mock('@/eventbus/message-event-bus/message-event-bus');
|
|
await eventBus?.close();
|
|
});
|
|
|
|
test('should have a running logwriter process', () => {
|
|
const thread = eventBus.logWriter.worker;
|
|
expect(thread).toBeDefined();
|
|
});
|
|
|
|
describe('message confirmation', () => {
|
|
afterEach(async () => {
|
|
// Restore the log-streaming destination listener for subsequent tests
|
|
destinationService['isListening'] = false;
|
|
await destinationService.initialize();
|
|
});
|
|
|
|
test('should confirm messages immediately when no listener is registered', async () => {
|
|
// Simulate an unlicensed instance: remove all message listeners
|
|
eventBus.removeAllListeners('message');
|
|
|
|
const testMessage = new EventMessageGeneric({
|
|
eventName: 'n8n.test.message' as EventNamesTypes,
|
|
id: uuid(),
|
|
});
|
|
|
|
await eventBus.send(testMessage);
|
|
await new Promise((resolve) => {
|
|
eventBus.logWriter.worker?.on(
|
|
'message',
|
|
async function handler(msg: { command: string; data: any }) {
|
|
if (msg.command === 'confirmMessageSent') {
|
|
await confirmIdSent(testMessage.id);
|
|
eventBus.logWriter.worker?.removeListener('message', handler);
|
|
resolve(true);
|
|
}
|
|
},
|
|
);
|
|
});
|
|
});
|
|
|
|
test('should delegate confirmation to listener when one is registered', async () => {
|
|
const testMessage = new EventMessageGeneric({
|
|
eventName: 'n8n.test.message' as EventNamesTypes,
|
|
id: uuid(),
|
|
});
|
|
|
|
await eventBus.send(testMessage);
|
|
// The first worker message should be appendMessageToLog, not confirmMessageSent.
|
|
// This proves the event bus delegated to the handler instead of auto-confirming.
|
|
await new Promise((resolve) => {
|
|
const workerMessages: string[] = [];
|
|
eventBus.logWriter.worker?.on(
|
|
'message',
|
|
function handler(msg: { command: string; data: unknown }) {
|
|
workerMessages.push(msg.command);
|
|
if (
|
|
workerMessages.includes('appendMessageToLog') &&
|
|
workerMessages.includes('confirmMessageSent')
|
|
) {
|
|
expect(workerMessages[0]).toBe('appendMessageToLog');
|
|
eventBus.logWriter.worker?.removeListener('message', handler);
|
|
resolve(true);
|
|
}
|
|
},
|
|
);
|
|
});
|
|
});
|
|
});
|
|
|
|
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 destinationService.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);
|
|
});
|
|
});
|
|
|
|
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 = destinationService['destinations'][
|
|
testSyslogDestination.id!
|
|
] as MessageEventBusDestinationSyslog;
|
|
|
|
syslogDestination.enabled = true;
|
|
|
|
const mockedSyslogClientLog = jest.spyOn(syslogDestination.client, 'log');
|
|
mockedSyslogClientLog.mockImplementation(async (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');
|
|
});
|
|
|
|
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') {
|
|
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') {
|
|
await eventBus.getEventsAll();
|
|
await confirmIdInAll(testAuditMessage.id);
|
|
expect(mockedSyslogClientLog).toHaveBeenCalled();
|
|
syslogDestination.enabled = false;
|
|
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 = destinationService['destinations'][
|
|
testWebhookDestination.id!
|
|
] as MessageEventBusDestinationWebhook;
|
|
|
|
webhookDestination.enabled = true;
|
|
|
|
mockAxiosInstance.post.mockResolvedValue({ status: 200, data: { msg: 'OK' } });
|
|
mockAxiosInstance.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(mockAxiosInstance.request).toHaveBeenCalled();
|
|
webhookDestination.enabled = false;
|
|
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 = destinationService['destinations'][
|
|
testSentryDestination.id!
|
|
] as MessageEventBusDestinationSentry;
|
|
|
|
sentryDestination.enabled = true;
|
|
|
|
const mockedSentryCaptureMessage = jest.spyOn(sentryDestination.sentryClient!, 'captureMessage');
|
|
mockedSentryCaptureMessage.mockImplementation((_m, _level, _hint, _scope) => {
|
|
eventBus.confirmMessageDelivered(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.enabled = false;
|
|
eventBus.logWriter.worker?.removeListener('message', handler004);
|
|
resolve(true);
|
|
}
|
|
},
|
|
);
|
|
});
|
|
});
|
|
|
|
test('DELETE /eventbus/destination delete all destinations by id', async () => {
|
|
const existingDestinations = await destinationService.findDestination();
|
|
const existingDestinationIds = existingDestinations.reduce<string[]>((acc, d) => {
|
|
if (d.id) {
|
|
acc.push(d.id);
|
|
}
|
|
return acc;
|
|
}, []);
|
|
|
|
// Delete sequentially to avoid race conditions
|
|
for (const id of existingDestinationIds) {
|
|
const response = await authOwnerAgent.del('/eventbus/destination').query({ id });
|
|
expect(response.statusCode).toBe(200);
|
|
}
|
|
|
|
const remainingDestinations = await destinationService.findDestination();
|
|
expect(remainingDestinations.length).toBe(0);
|
|
});
|