houseassassin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Run integration tests against the real BullMQ driver (#22551)

Browse Source 
Migrate whole suite to real BullMQ

Shard times unchanged, still 5-6 min.
This commit is contained in:
neo773
2026-07-05 04:00:21 +05:30
committed by GitHub
parent 4e43a0fb4e
commit f14e62ec0c
10 changed files with 252 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/twenty-server/jest-integration.config.ts
+1 -3
View File
@@ -35,6 +35,7 @@ const jestConfig: JestConfigWithTsJest = {
modulePathIgnorePatterns: ['<rootDir>/dist'],
globalSetup: '<rootDir>/test/integration/utils/setup-test.ts',
globalTeardown: '<rootDir>/test/integration/utils/teardown-test.ts',
setupFilesAfterEnv: ['<rootDir>/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts'],
testTimeout: 20000,
maxWorkers: 1,
// jsdom 29 pulls ESM-only transitive deps (parse5, entities, tough-cookie,
@@ -81,9 +82,6 @@ const jestConfig: JestConfigWithTsJest = {
}),
'^test/(.*)$': '<rootDir>/test/$1',
},
fakeTimers: {
enableGlobally: true,
},
globals: {
APP_PORT: 4000,
NODE_ENV: NodeEnvironment.TEST,
packages/twenty-server/test/integration/graphql/suites/auth/jwt-key-rotation/rotate-signing-keys-cron-job.integration-spec.ts
+23 -23
View File
@@ -6,6 +6,7 @@ import { findManyApplications } from 'test/integration/graphql/utils/find-many-a
import { generateApiKeyToken } from 'test/integration/graphql/utils/generate-api-key-token.util';
import { deleteConfigVariable } from 'test/integration/twenty-config/utils/delete-config-variable.util';
import { updateConfigVariable } from 'test/integration/twenty-config/utils/update-config-variable.util';
import { expectEventually } from 'test/integration/utils/expect-eventually.util';
import { RotateSigningKeysCronJob } from 'src/engine/core-modules/jwt/crons/jobs/rotate-signing-keys.cron.job';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
@@ -14,7 +15,6 @@ import { getQueueToken } from 'src/engine/core-modules/message-queue/utils/get-q
import { API_KEY_DATA_SEED_IDS } from 'src/engine/workspace-manager/dev-seeder/data/constants/api-key-data-seeds.constant';
const SIGNING_KEY_ROTATION_DAYS_KEY = 'SIGNING_KEY_ROTATION_DAYS';
const ROTATE_SIGNING_KEYS_CRON_PATTERN = '15 3 * * *';
describe('RotateSigningKeysCronJob (integration)', () => {
const seededApiKeyId = API_KEY_DATA_SEED_IDS.ID_1;
@@ -62,30 +62,30 @@ describe('RotateSigningKeysCronJob (integration)', () => {
input: { key: SIGNING_KEY_ROTATION_DAYS_KEY, value: 0 },
});
await cronQueue.addCron({
jobName: RotateSigningKeysCronJob.name,
data: undefined,
options: { repeat: { pattern: ROTATE_SIGNING_KEYS_CRON_PATTERN } },
await cronQueue.add(RotateSigningKeysCronJob.name, {});
let rotatedApiKeyToken = '';
await expectEventually(async () => {
const rotatedTokenResponse = await generateApiKeyToken({
apiKeyId: seededApiKeyId,
accessToken: APPLE_JANE_ADMIN_ACCESS_TOKEN,
});
expect(rotatedTokenResponse.body.errors).toBeUndefined();
rotatedApiKeyToken =
rotatedTokenResponse.body.data?.generateApiKeyToken.token ?? '';
expect(isNonEmptyString(rotatedApiKeyToken)).toBe(true);
const rotatedKid = decodeJwtCompleteOrThrow(rotatedApiKeyToken).header
.kid as string;
expect(isNonEmptyString(rotatedKid)).toBe(true);
expect(rotatedKid).not.toBe(initialKid);
});
const rotatedTokenResponse = await generateApiKeyToken({
apiKeyId: seededApiKeyId,
accessToken: APPLE_JANE_ADMIN_ACCESS_TOKEN,
});
expect(rotatedTokenResponse.body.errors).toBeUndefined();
const rotatedApiKeyToken: string =
rotatedTokenResponse.body.data?.generateApiKeyToken.token ?? '';
expect(isNonEmptyString(rotatedApiKeyToken)).toBe(true);
const rotatedKid = decodeJwtCompleteOrThrow(rotatedApiKeyToken).header
.kid as string;
expect(isNonEmptyString(rotatedKid)).toBe(true);
expect(rotatedKid).not.toBe(initialKid);
const callWithPreviousToken = await findManyApplications({
accessToken: initialApiKeyToken,
expectToFail: false,
packages/twenty-server/test/integration/graphql/suites/nested-relation-per-parent-limit.integration-spec.ts
+2 -2
View File
@@ -59,7 +59,7 @@ describe('Nested relation per-parent limit (e2e)', () => {
{ id: SOFT_DELETED_COMPANY_ID, name: 'Soft-deleted relation company' },
{ id: EMPTY_COMPANY_ID, name: 'Empty relation company' },
],
upsert: true,
upsert: false,
});
await makeGraphqlAPIRequest(createCompanies);
@@ -76,7 +76,7 @@ describe('Nested relation per-parent limit (e2e)', () => {
companyId: SOFT_DELETED_COMPANY_ID,
})),
],
upsert: true,
upsert: false,
});
await makeGraphqlAPIRequest(createPeople);
packages/twenty-server/test/integration/graphql/suites/search/search-resolver.integration-spec.ts
+4
View File
@@ -26,6 +26,7 @@ import {
import { createManyOperation } from 'test/integration/graphql/utils/create-many-operation.util';
import { search } from 'test/integration/graphql/utils/search.util';
import { deleteAllRecords } from 'test/integration/utils/delete-all-records';
import { waitForAllJobsToFinish } from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
import {
eachTestingContextFilter,
type EachTestingContext,
@@ -192,6 +193,9 @@ describe('SearchResolver', () => {
data: persons,
});
await waitForAllJobsToFinish();
await deleteAllRecords('company');
await createManyOperation({
objectMetadataSingularName: 'company',
objectMetadataPluralName: 'companies',
packages/twenty-server/test/integration/graphql/suites/workflow/quick-lead-workflow.integration-spec.ts
+13 -3
View File
@@ -3,6 +3,8 @@ import {
destroyWorkflowRun,
getWorkflowRun,
runWorkflowVersion,
waitForWorkflowCompletion,
waitForWorkflowRunStatus,
} from 'test/integration/graphql/suites/workflow/utils/workflow-run-test.util';
import { v4 as uuidv4 } from 'uuid';
@@ -125,7 +127,10 @@ describe('Quick Lead Workflow (e2e)', () => {
createdWorkflowRunId = workflowRunId;
const workflowRun = await getWorkflowRun(workflowRunId);
const workflowRun = await waitForWorkflowRunStatus(
workflowRunId,
'RUNNING',
);
expect(workflowRun).toBeDefined();
expect(workflowRun?.workflowVersionId).toBe(
@@ -238,7 +243,10 @@ describe('Quick Lead Workflow (e2e)', () => {
expect(testWorkflowRunId).toBeDefined();
let workflowRun = await getWorkflowRun(testWorkflowRunId as string);
let workflowRun = await waitForWorkflowRunStatus(
testWorkflowRunId as string,
'RUNNING',
);
expect(workflowRun?.status).toBe('RUNNING');
expect(workflowRun?.state?.stepInfos?.[FORM_STEP_ID]?.status).toBe(
@@ -276,7 +284,9 @@ describe('Quick Lead Workflow (e2e)', () => {
expect(submitFormResponse.body.errors).toBeUndefined();
expect(submitFormResponse.body.data.submitFormStep).toBe(true);
workflowRun = await getWorkflowRun(testWorkflowRunId as string);
workflowRun = await waitForWorkflowCompletion(
testWorkflowRunId as string,
);
expect(workflowRun?.status).toBe('COMPLETED');
expect(workflowRun?.state?.stepInfos?.trigger?.status).toBe('SUCCESS');
expect(workflowRun?.state?.stepInfos?.[FORM_STEP_ID]?.status).toBe(
packages/twenty-server/test/integration/graphql/suites/workflow/utils/workflow-run-test.util.ts
+33 -3
View File
@@ -117,6 +117,13 @@ export const destroyWorkflowRun = async (
});
};
const PENDING_WORKFLOW_RUN_STATUSES: WorkflowRunStatusType[] = [
'NOT_STARTED',
'ENQUEUED',
'RUNNING',
'STOPPING',
];
export const waitForWorkflowCompletion = async (
workflowRunId: string,
maxAttempts = 30,
@@ -126,9 +133,32 @@ export const waitForWorkflowCompletion = async (
let attempts = 0;
while (
workflowRun?.status === 'RUNNING' &&
attempts < maxAttempts &&
workflowRun !== null
workflowRun !== null &&
PENDING_WORKFLOW_RUN_STATUSES.includes(workflowRun.status) &&
attempts < maxAttempts
) {
await new Promise((resolve) => setTimeout(resolve, intervalMs));
workflowRun = await getWorkflowRun(workflowRunId);
attempts++;
}
return workflowRun;
};
export const waitForWorkflowRunStatus = async (
workflowRunId: string,
expectedStatus: WorkflowRunStatusType,
maxAttempts = 30,
intervalMs = 500,
): Promise<WorkflowRunResponse | null> => {
let workflowRun = await getWorkflowRun(workflowRunId);
let attempts = 0;
while (
attempts < maxAttempts &&
workflowRun?.status !== expectedStatus &&
(workflowRun === null ||
PENDING_WORKFLOW_RUN_STATUSES.includes(workflowRun.status))
) {
await new Promise((resolve) => setTimeout(resolve, intervalMs));
workflowRun = await getWorkflowRun(workflowRunId);
packages/twenty-server/test/integration/utils/create-app.ts
+1 -9
View File
@@ -17,9 +17,7 @@ import { CaptchaDriverFactory } from 'src/engine/core-modules/captcha/captcha-dr
import { ExceptionHandlerService } from 'src/engine/core-modules/exception-handler/exception-handler.service';
import { ExceptionHandlerMockService } from 'src/engine/core-modules/exception-handler/mocks/exception-handler-mock.service';
import { MockedUnhandledExceptionFilter } from 'src/engine/core-modules/exception-handler/mocks/mock-unhandled-exception.filter';
import { SyncDriver } from 'src/engine/core-modules/message-queue/drivers/sync.driver';
import { JobsModule } from 'src/engine/core-modules/message-queue/jobs.module';
import { QUEUE_DRIVER } from 'src/engine/core-modules/message-queue/message-queue.constants';
import { MessageQueueModule } from 'src/engine/core-modules/message-queue/message-queue.module';
interface TestingModuleCreatePreHook {
@@ -33,10 +31,6 @@ export type TestingAppCreatePreHook = (
app: NestExpressApplication,
) => Promise<void>;
// Shared SyncDriver instance for all queues in tests
// This enables synchronous processing of jobs during integration tests
const syncDriver = new SyncDriver();
/**
* Sets basic integration testing module of app
*/
@@ -66,9 +60,7 @@ export const createApp = async (
getCurrentDriver: () => ({
validate: async () => ({ success: true }),
}),
})
.overrideProvider(QUEUE_DRIVER)
.useValue(syncDriver);
});
if (config.moduleBuilderHook) {
moduleBuilder = config.moduleBuilderHook(moduleBuilder);
packages/twenty-server/test/integration/utils/expect-eventually.util.ts
+25
View File
@@ -0,0 +1,25 @@
type ExpectEventuallyOptions = {
timeoutMs?: number;
intervalMs?: number;
};
export const expectEventually = async (
assertion: () => Promise<void> | void,
{ timeoutMs = 10_000, intervalMs = 100 }: ExpectEventuallyOptions = {},
): Promise<void> => {
const startedAt = Date.now();
for (;;) {
try {
await assertion();
return;
} catch (error) {
if (Date.now() - startedAt > timeoutMs) {
throw error;
}
await new Promise((resolve) => setTimeout(resolve, intervalMs));
}
}
};
packages/twenty-server/test/integration/utils/setup-wait-for-all-jobs-between-tests.ts
+19
View File
@@ -0,0 +1,19 @@
import {
closeQueueConnections,
waitForAllJobsToFinish,
} from 'test/integration/utils/wait-for-all-jobs-to-finish.util';
const WAIT_FOR_JOBS_HOOK_TIMEOUT_MS = 150_000;
beforeAll(async () => {
await waitForAllJobsToFinish();
}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS);
afterEach(async () => {
jest.useRealTimers();
await waitForAllJobsToFinish();
}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS);
afterAll(async () => {
await closeQueueConnections();
}, WAIT_FOR_JOBS_HOOK_TIMEOUT_MS);
packages/twenty-server/test/integration/utils/wait-for-all-jobs-to-finish.util.ts
+131
View File
@@ -0,0 +1,131 @@
import { Queue } from 'bullmq';
import IORedis from 'ioredis';
import { MessageQueue } from 'src/engine/core-modules/message-queue/message-queue.constants';
const POLL_INTERVAL_MS = 25;
const REQUIRED_CONSECUTIVE_QUIET_CHECKS = 2;
const STALL_TIMEOUT_MS = 15_000;
const HARD_TIMEOUT_MS = 120_000;
const PENDING_JOB_STATES = [
'waiting',
'active',
'prioritized',
'waiting-children',
'delayed',
] as const;
let redisConnection: IORedis | null = null;
let queues: Queue[] | null = null;
const getQueues = (): Queue[] => {
if (!queues) {
redisConnection = new IORedis(
process.env.REDIS_QUEUE_URL ??
process.env.REDIS_URL ??
'redis://localhost:6379',
{ maxRetriesPerRequest: null },
);
queues = Object.values(MessageQueue).map(
(queueName) => new Queue(queueName, { connection: redisConnection! }),
);
}
return queues;
};
const getPendingJobCountsByQueue = async (): Promise<
Record<string, number>
> => {
const countsByQueue = await Promise.all(
getQueues().map(async (queue) => {
const jobCounts = await queue.getJobCounts(...PENDING_JOB_STATES);
const pendingCount = Object.values(jobCounts).reduce(
(sum, count) => sum + count,
0,
);
return [queue.name, pendingCount] as const;
}),
);
return Object.fromEntries(
countsByQueue.filter(([, pendingCount]) => pendingCount > 0),
);
};
const getActiveJobsFingerprint = async (
busyQueueNames: string[],
): Promise<string> => {
const activeJobIdsByQueue = await Promise.all(
getQueues()
.filter((queue) => busyQueueNames.includes(queue.name))
.map(async (queue) => {
const activeJobs = await queue.getActive(0, 50);
return `${queue.name}:${activeJobs.map((job) => job.id).join(',')}`;
}),
);
return activeJobIdsByQueue.join('|');
};
export const waitForAllJobsToFinish = async (): Promise<void> => {
const startedAt = Date.now();
let lastProgressAt = startedAt;
let lastBusyFingerprint = '';
let consecutiveQuietChecks = 0;
while (consecutiveQuietChecks < REQUIRED_CONSECUTIVE_QUIET_CHECKS) {
const pendingJobCountsByQueue = await getPendingJobCountsByQueue();
const pendingTotal = Object.values(pendingJobCountsByQueue).reduce(
(sum, count) => sum + count,
0,
);
if (pendingTotal === 0) {
consecutiveQuietChecks += 1;
} else {
consecutiveQuietChecks = 0;
const now = Date.now();
const activeJobsFingerprint = await getActiveJobsFingerprint(
Object.keys(pendingJobCountsByQueue),
);
const busyFingerprint = `${pendingTotal}|${activeJobsFingerprint}`;
if (busyFingerprint !== lastBusyFingerprint) {
lastBusyFingerprint = busyFingerprint;
lastProgressAt = now;
}
if (now - lastProgressAt > STALL_TIMEOUT_MS) {
throw new Error(
`Message queues stalled, no progress for ${STALL_TIMEOUT_MS}ms, pending jobs: ${JSON.stringify(pendingJobCountsByQueue)}`,
);
}
if (now - startedAt > HARD_TIMEOUT_MS) {
throw new Error(
`Message queues still busy after ${HARD_TIMEOUT_MS}ms, pending jobs: ${JSON.stringify(pendingJobCountsByQueue)}`,
);
}
}
if (consecutiveQuietChecks < REQUIRED_CONSECUTIVE_QUIET_CHECKS) {
await new Promise((resolve) => setTimeout(resolve, POLL_INTERVAL_MS));
}
}
};
export const closeQueueConnections = async (): Promise<void> => {
if (queues) {
await Promise.allSettled(queues.map((queue) => queue.close()));
queues = null;
}
if (redisConnection) {
await redisConnection.quit().catch(() => redisConnection?.disconnect());
redisConnection = null;
}
};