tests: use describe.each for database test iteration
Refactors all test files that use TestDatabases/TestCaches with it.each(databases.eachSupportedId()) to instead use describe.each at the outer level. This ensures that all tests for one database engine complete before moving to the next, rather than interleaving engines across individual tests. This reduces the number of concurrent database connections and should help with test timeout issues in CI. The TestDatabases.create() call is hoisted to module scope so the describe.each can iterate over supported IDs at the top level. 40 files changed across packages/backend-defaults, packages/backend-test-utils, and multiple plugins. Signed-off-by: Fredrik Adelöw <freben@gmail.com> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Signed-off-by: Fredrik Adelöw <freben@gmail.com>
This commit is contained in:
Fredrik Adelöw
@@ -45,60 +45,54 @@ const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_9', 'POSTGRES_14', 'POSTGRES_16'],
|
||||
});
|
||||
|
||||
const maybeDescribe =
|
||||
databases.eachSupportedId().length > 0 ? describe : describe.skip;
|
||||
describe.each(databases.eachSupportedId())('migrations, %p', databaseId => {
|
||||
it('20240523100528_init.js', async () => {
|
||||
const knex = await databases.init(databaseId);
|
||||
|
||||
maybeDescribe('migrations', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'20240523100528_init.js, %p',
|
||||
async databaseId => {
|
||||
const knex = await databases.init(databaseId);
|
||||
await migrateUntilBefore(knex, '20240523100528_init.js');
|
||||
await migrateUpOnce(knex);
|
||||
|
||||
await migrateUntilBefore(knex, '20240523100528_init.js');
|
||||
await migrateUpOnce(knex);
|
||||
await knex('event_bus_events').insert({
|
||||
topic: 'test',
|
||||
created_by: 'abc',
|
||||
data_json: JSON.stringify({ message: 'hello' }),
|
||||
notified_subscribers: ['tester'],
|
||||
});
|
||||
await knex('event_bus_subscriptions').insert({
|
||||
id: 'tester',
|
||||
created_by: 'abc',
|
||||
read_until: '5',
|
||||
topics: ['test', 'test2'],
|
||||
});
|
||||
|
||||
await knex('event_bus_events').insert({
|
||||
topic: 'test',
|
||||
await expect(knex('event_bus_events')).resolves.toEqual([
|
||||
{
|
||||
id: '1',
|
||||
created_by: 'abc',
|
||||
topic: 'test',
|
||||
data_json: JSON.stringify({ message: 'hello' }),
|
||||
created_at: expect.anything(),
|
||||
notified_subscribers: ['tester'],
|
||||
});
|
||||
await knex('event_bus_subscriptions').insert({
|
||||
},
|
||||
]);
|
||||
await expect(knex('event_bus_subscriptions')).resolves.toEqual([
|
||||
{
|
||||
id: 'tester',
|
||||
created_by: 'abc',
|
||||
created_at: expect.anything(),
|
||||
updated_at: expect.anything(),
|
||||
read_until: '5',
|
||||
topics: ['test', 'test2'],
|
||||
});
|
||||
},
|
||||
]);
|
||||
|
||||
await expect(knex('event_bus_events')).resolves.toEqual([
|
||||
{
|
||||
id: '1',
|
||||
created_by: 'abc',
|
||||
topic: 'test',
|
||||
data_json: JSON.stringify({ message: 'hello' }),
|
||||
created_at: expect.anything(),
|
||||
notified_subscribers: ['tester'],
|
||||
},
|
||||
]);
|
||||
await expect(knex('event_bus_subscriptions')).resolves.toEqual([
|
||||
{
|
||||
id: 'tester',
|
||||
created_by: 'abc',
|
||||
created_at: expect.anything(),
|
||||
updated_at: expect.anything(),
|
||||
read_until: '5',
|
||||
topics: ['test', 'test2'],
|
||||
},
|
||||
]);
|
||||
await migrateDownOnce(knex);
|
||||
|
||||
await migrateDownOnce(knex);
|
||||
// This looks odd - you might expect a .toThrow at the end but that
|
||||
// actually is flaky for some reason specifically on sqlite when
|
||||
// performing multiple runs in sequence
|
||||
await expect(knex('event_bus_events')).rejects.toEqual(expect.anything());
|
||||
|
||||
// This looks odd - you might expect a .toThrow at the end but that
|
||||
// actually is flaky for some reason specifically on sqlite when
|
||||
// performing multiple runs in sequence
|
||||
await expect(knex('event_bus_events')).rejects.toEqual(expect.anything());
|
||||
|
||||
await knex.destroy();
|
||||
},
|
||||
);
|
||||
await knex.destroy();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -22,7 +22,6 @@ import {
|
||||
} from '@backstage/backend-plugin-api';
|
||||
import {
|
||||
TestBackend,
|
||||
TestDatabaseId,
|
||||
TestDatabases,
|
||||
mockCredentials,
|
||||
mockServices,
|
||||
@@ -105,8 +104,15 @@ describe('eventsPlugin', () => {
|
||||
test: 'fake-ext',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe('event bus', () => {
|
||||
const eventBusDatabases = TestDatabases.create({
|
||||
ids: ['SQLITE_3', 'MYSQL_8', 'POSTGRES_14', 'POSTGRES_18'],
|
||||
});
|
||||
|
||||
describe.each(eventBusDatabases.eachSupportedId())(
|
||||
'event bus, %p',
|
||||
databaseId => {
|
||||
class ReqHelper {
|
||||
private readonly backend: TestBackend;
|
||||
|
||||
@@ -146,12 +152,8 @@ describe('eventsPlugin', () => {
|
||||
}
|
||||
}
|
||||
|
||||
const databases = TestDatabases.create({
|
||||
ids: ['SQLITE_3', 'MYSQL_8', 'POSTGRES_14', 'POSTGRES_18'],
|
||||
});
|
||||
|
||||
async function mockKnexFactory(databaseId: TestDatabaseId) {
|
||||
const knex = await databases.init(databaseId);
|
||||
async function mockKnexFactory() {
|
||||
const knex = await eventBusDatabases.init(databaseId);
|
||||
return mockServices.database.factory({ knex });
|
||||
}
|
||||
|
||||
@@ -163,278 +165,254 @@ describe('eventsPlugin', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should be possible to publish events as a service, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
it('should be possible to publish events as a service', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.service.header())
|
||||
.expect(204); // 204, since there are no subscribers
|
||||
},
|
||||
);
|
||||
await helper
|
||||
.publish('test', { n: 1 })
|
||||
.set('authorization', mockCredentials.service.header())
|
||||
.expect(204); // 204, since there are no subscribers
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should be possible to subscribe as a service and receive an event, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
it('should be possible to subscribe as a service and receive an event', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
await helper
|
||||
.subscribe('tester', ['test'])
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
await helper
|
||||
.subscribe('tester', ['test'])
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
|
||||
await helper
|
||||
.subscribe('tester', ['test'])
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
await helper
|
||||
.subscribe('tester', ['test'])
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
await helper
|
||||
.readEvents('tester')
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
await helper
|
||||
.readEvents('tester')
|
||||
.set('authorization', mockCredentials.none.header())
|
||||
.expect(401);
|
||||
|
||||
await helper
|
||||
.readEvents('tester')
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
await helper
|
||||
.readEvents('tester')
|
||||
.set('authorization', mockCredentials.user.header())
|
||||
.expect(403);
|
||||
|
||||
await helper.publish('test', { n: 1 }).expect(201); // 201, since there is a subscriber
|
||||
await helper.publish('test', { n: 1 }).expect(201); // 201, since there is a subscriber
|
||||
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
},
|
||||
);
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should only send an event for each subscriber once, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
it('should only send an event for each subscriber once', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester-1', ['test']).expect(201);
|
||||
await helper.subscribe('tester-2', ['test']).expect(201);
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester-1', ['test']).expect(201);
|
||||
await helper.subscribe('tester-2', ['test']).expect(201);
|
||||
|
||||
// A single event
|
||||
await helper.publish('test', { n: 1 }).expect(201);
|
||||
// A single event
|
||||
await helper.publish('test', { n: 1 }).expect(201);
|
||||
|
||||
// Single client for subscriber 1 gets the event
|
||||
await helper.readEvents('tester-1').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
// Single client for subscriber 1 gets the event
|
||||
await helper.readEvents('tester-1').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
|
||||
// Two clients for subscriber 2, only one gets the event
|
||||
const res1 = helper.readEvents('tester-2');
|
||||
const res2 = helper.readEvents('tester-2');
|
||||
// Two clients for subscriber 2, only one gets the event
|
||||
const res1 = helper.readEvents('tester-2');
|
||||
const res2 = helper.readEvents('tester-2');
|
||||
|
||||
const res = await Promise.race([res1, res2]);
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
const res = await Promise.race([res1, res2]);
|
||||
expect(res.status).toBe(200);
|
||||
expect(res.body).toEqual({
|
||||
events: [{ topic: 'test', payload: { n: 1 } }],
|
||||
});
|
||||
|
||||
// Post another event, which triggers the other client to return
|
||||
await helper.publish('test', { n: 2 }).expect(201);
|
||||
// Post another event, which triggers the other client to return
|
||||
await helper.publish('test', { n: 2 }).expect(201);
|
||||
|
||||
const otherRes = await Promise.all([res1, res2]).then(rs =>
|
||||
rs.find(r => r !== res),
|
||||
const otherRes = await Promise.all([res1, res2]).then(rs =>
|
||||
rs.find(r => r !== res),
|
||||
);
|
||||
expect(otherRes?.status).toBe(202);
|
||||
|
||||
// Reading subscriber 2 should now return the second event only
|
||||
await helper.readEvents('tester-2').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 2 } }],
|
||||
});
|
||||
});
|
||||
|
||||
it('should not notify subscribers that have already consumed the event', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester-1', ['test']).expect(201);
|
||||
await helper.subscribe('tester-2', ['test']).expect(201);
|
||||
|
||||
// A single event for each subscriber, that should not be sent to the other one
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-2' },
|
||||
{
|
||||
notifiedSubscribers: ['tester-1'],
|
||||
},
|
||||
)
|
||||
.expect(201);
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-1' },
|
||||
{
|
||||
notifiedSubscribers: ['tester-2'],
|
||||
},
|
||||
)
|
||||
.expect(201);
|
||||
|
||||
// Single client for subscriber 1 gets the event
|
||||
await helper.readEvents('tester-1').expect(200, {
|
||||
events: [{ topic: 'test', payload: { for: 'tester-1' } }],
|
||||
});
|
||||
// Single client for subscriber 2 gets the event
|
||||
await helper.readEvents('tester-2').expect(200, {
|
||||
events: [{ topic: 'test', payload: { for: 'tester-2' } }],
|
||||
});
|
||||
});
|
||||
|
||||
it('should return multiple events in order', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
// A sequence of events published one at a time
|
||||
for (let n = 0; n < 15; ++n) {
|
||||
await helper.publish('test', { n }).expect(201);
|
||||
}
|
||||
|
||||
// Batch size it 10
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [
|
||||
{ topic: 'test', payload: { n: 0 } },
|
||||
{ topic: 'test', payload: { n: 1 } },
|
||||
{ topic: 'test', payload: { n: 2 } },
|
||||
{ topic: 'test', payload: { n: 3 } },
|
||||
{ topic: 'test', payload: { n: 4 } },
|
||||
{ topic: 'test', payload: { n: 5 } },
|
||||
{ topic: 'test', payload: { n: 6 } },
|
||||
{ topic: 'test', payload: { n: 7 } },
|
||||
{ topic: 'test', payload: { n: 8 } },
|
||||
{ topic: 'test', payload: { n: 9 } },
|
||||
],
|
||||
});
|
||||
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [
|
||||
{ topic: 'test', payload: { n: 10 } },
|
||||
{ topic: 'test', payload: { n: 11 } },
|
||||
{ topic: 'test', payload: { n: 12 } },
|
||||
{ topic: 'test', payload: { n: 13 } },
|
||||
{ topic: 'test', payload: { n: 14 } },
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
it('should skip publishing if all subscribers have already consumed the event', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-2' },
|
||||
{ notifiedSubscribers: ['tester'] },
|
||||
)
|
||||
.expect(204);
|
||||
});
|
||||
|
||||
it('should time out when no events are available', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
jest.useFakeTimers({
|
||||
advanceTimers: true,
|
||||
});
|
||||
|
||||
try {
|
||||
// Can't use supertest for this one because it can't handle the partially blocking response
|
||||
const res = await fetch(
|
||||
`http://localhost:${backend.server.port()}/api/events/bus/v1/subscriptions/tester/events`,
|
||||
{
|
||||
headers: {
|
||||
authorization: mockCredentials.service.header(),
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(otherRes?.status).toBe(202);
|
||||
|
||||
// Reading subscriber 2 should now return the second event only
|
||||
await helper.readEvents('tester-2').expect(200, {
|
||||
events: [{ topic: 'test', payload: { n: 2 } }],
|
||||
});
|
||||
},
|
||||
);
|
||||
expect(res.status).toBe(202);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should not notify subscribers that have already consumed the event, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
const reader = res.body!.getReader();
|
||||
const isClosed = () =>
|
||||
Promise.race([
|
||||
reader
|
||||
.read()
|
||||
.then(() => reader.closed)
|
||||
.then(() => true),
|
||||
new Promise<boolean>(r => setTimeout(r, 100, false)),
|
||||
]);
|
||||
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester-1', ['test']).expect(201);
|
||||
await helper.subscribe('tester-2', ['test']).expect(201);
|
||||
await expect(isClosed()).resolves.toBe(false);
|
||||
await jest.advanceTimersByTimeAsync(30000);
|
||||
await expect(isClosed()).resolves.toBe(false);
|
||||
await jest.advanceTimersByTimeAsync(30000);
|
||||
await expect(isClosed()).resolves.toBe(true);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
// A single event for each subscriber, that should not be sent to the other one
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-2' },
|
||||
{
|
||||
notifiedSubscribers: ['tester-1'],
|
||||
},
|
||||
)
|
||||
.expect(201);
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-1' },
|
||||
{
|
||||
notifiedSubscribers: ['tester-2'],
|
||||
},
|
||||
)
|
||||
.expect(201);
|
||||
it('should refuse listen without a subscription', async () => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory()],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
// Single client for subscriber 1 gets the event
|
||||
await helper.readEvents('tester-1').expect(200, {
|
||||
events: [{ topic: 'test', payload: { for: 'tester-1' } }],
|
||||
});
|
||||
// Single client for subscriber 2 gets the event
|
||||
await helper.readEvents('tester-2').expect(200, {
|
||||
events: [{ topic: 'test', payload: { for: 'tester-2' } }],
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should return multiple events in order, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
// 2 subscribers
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
// A sequence of events published one at a time
|
||||
for (let n = 0; n < 15; ++n) {
|
||||
await helper.publish('test', { n }).expect(201);
|
||||
}
|
||||
|
||||
// Batch size it 10
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [
|
||||
{ topic: 'test', payload: { n: 0 } },
|
||||
{ topic: 'test', payload: { n: 1 } },
|
||||
{ topic: 'test', payload: { n: 2 } },
|
||||
{ topic: 'test', payload: { n: 3 } },
|
||||
{ topic: 'test', payload: { n: 4 } },
|
||||
{ topic: 'test', payload: { n: 5 } },
|
||||
{ topic: 'test', payload: { n: 6 } },
|
||||
{ topic: 'test', payload: { n: 7 } },
|
||||
{ topic: 'test', payload: { n: 8 } },
|
||||
{ topic: 'test', payload: { n: 9 } },
|
||||
],
|
||||
});
|
||||
|
||||
await helper.readEvents('tester').expect(200, {
|
||||
events: [
|
||||
{ topic: 'test', payload: { n: 10 } },
|
||||
{ topic: 'test', payload: { n: 11 } },
|
||||
{ topic: 'test', payload: { n: 12 } },
|
||||
{ topic: 'test', payload: { n: 13 } },
|
||||
{ topic: 'test', payload: { n: 14 } },
|
||||
],
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should skip publishing if all subscribers have already consumed the event, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
await helper
|
||||
.publish(
|
||||
'test',
|
||||
{ for: 'tester-2' },
|
||||
{ notifiedSubscribers: ['tester'] },
|
||||
)
|
||||
.expect(204);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should time out when no events are available, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
await helper.subscribe('tester', ['test']).expect(201);
|
||||
|
||||
jest.useFakeTimers({
|
||||
advanceTimers: true,
|
||||
});
|
||||
|
||||
try {
|
||||
// Can't use supertest for this one because it can't handle the partially blocking response
|
||||
const res = await fetch(
|
||||
`http://localhost:${backend.server.port()}/api/events/bus/v1/subscriptions/tester/events`,
|
||||
{
|
||||
headers: {
|
||||
authorization: mockCredentials.service.header(),
|
||||
},
|
||||
},
|
||||
);
|
||||
|
||||
expect(res.status).toBe(202);
|
||||
|
||||
const reader = res.body!.getReader();
|
||||
const isClosed = () =>
|
||||
Promise.race([
|
||||
reader
|
||||
.read()
|
||||
.then(() => reader.closed)
|
||||
.then(() => true),
|
||||
new Promise<boolean>(r => setTimeout(r, 100, false)),
|
||||
]);
|
||||
|
||||
await expect(isClosed()).resolves.toBe(false);
|
||||
await jest.advanceTimersByTimeAsync(30000);
|
||||
await expect(isClosed()).resolves.toBe(false);
|
||||
await jest.advanceTimersByTimeAsync(30000);
|
||||
await expect(isClosed()).resolves.toBe(true);
|
||||
} finally {
|
||||
jest.useRealTimers();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should refuse listen without a subscription, %p',
|
||||
async databaseId => {
|
||||
backend = await startTestBackend({
|
||||
features: [eventsPlugin, await mockKnexFactory(databaseId)],
|
||||
});
|
||||
const helper = new ReqHelper(backend);
|
||||
|
||||
await helper.readEvents('nonexistent').expect(404);
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
await helper.readEvents('nonexistent').expect(404);
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
@@ -29,13 +29,10 @@ const databases = TestDatabases.create({
|
||||
ids: ['POSTGRES_14', 'POSTGRES_18'],
|
||||
});
|
||||
|
||||
const maybeDescribe =
|
||||
databases.eachSupportedId().length > 0 ? describe : describe.skip;
|
||||
|
||||
maybeDescribe('DatabaseEventBusStore', () => {
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should clean up old events, %p',
|
||||
async databaseId => {
|
||||
describe.each(databases.eachSupportedId())(
|
||||
'DatabaseEventBusStore, %p',
|
||||
databaseId => {
|
||||
it('should clean up old events', async () => {
|
||||
const db = await databases.init(databaseId);
|
||||
const store = await DatabaseEventBusStore.forTest({ logger, db });
|
||||
|
||||
@@ -79,12 +76,9 @@ maybeDescribe('DatabaseEventBusStore', () => {
|
||||
|
||||
const { events: events3 } = await store.readSubscription('tester-3');
|
||||
expect(events3.length).toBe(5);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should always clean up events outside the max age window, %p',
|
||||
async databaseId => {
|
||||
it('should always clean up events outside the max age window', async () => {
|
||||
const db = await databases.init(databaseId);
|
||||
const store = await DatabaseEventBusStore.forTest({
|
||||
logger,
|
||||
@@ -132,12 +126,9 @@ maybeDescribe('DatabaseEventBusStore', () => {
|
||||
|
||||
const { events: events3 } = await store.readSubscription('tester-3');
|
||||
expect(events3.length).toBe(0);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should not clean up events within the min age window, %p',
|
||||
async databaseId => {
|
||||
it('should not clean up events within the min age window', async () => {
|
||||
const db = await databases.init(databaseId);
|
||||
const store = await DatabaseEventBusStore.forTest({
|
||||
logger,
|
||||
@@ -162,12 +153,9 @@ maybeDescribe('DatabaseEventBusStore', () => {
|
||||
|
||||
const { events: events1 } = await store.readSubscription('tester-1');
|
||||
expect(events1.length).toBe(10);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should clean up a large number of events, %p',
|
||||
async databaseId => {
|
||||
it('should clean up a large number of events', async () => {
|
||||
const db = await databases.init(databaseId);
|
||||
const store = await DatabaseEventBusStore.forTest({
|
||||
logger,
|
||||
@@ -197,12 +185,9 @@ maybeDescribe('DatabaseEventBusStore', () => {
|
||||
await expect(db('event_bus_events').count()).resolves.toEqual([
|
||||
{ count: '5' },
|
||||
]);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it.each(databases.eachSupportedId())(
|
||||
'should perform well when looking up events by topic, %p',
|
||||
async databaseId => {
|
||||
it('should perform well when looking up events by topic', async () => {
|
||||
const db = await databases.init(databaseId);
|
||||
const store = await DatabaseEventBusStore.forTest({
|
||||
logger,
|
||||
@@ -245,6 +230,6 @@ maybeDescribe('DatabaseEventBusStore', () => {
|
||||
]);
|
||||
|
||||
expect(duration).toBeLessThan(20);
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
Reference in New Issue
Block a user