events-backend: add event bus integration tests

Signed-off-by: Patrik Oldsberg <poldsberg@gmail.com>

plugins/events-backend/src/service/EventsPlugin.test.ts

@@ -14,11 +14,17 @@
  * limitations under the License.
  */
 
+/* eslint-disable jest/expect-expect */
+
 import {
   createBackendModule,
   createServiceFactory,
 } from '@backstage/backend-plugin-api';
-import { mockServices, startTestBackend } from '@backstage/backend-test-utils';
+import {
+  mockCredentials,
+  mockServices,
+  startTestBackend,
+} from '@backstage/backend-test-utils';
 import { eventsServiceRef } from '@backstage/plugin-events-node';
 import { eventsExtensionPoint } from '@backstage/plugin-events-node/alpha';
 import { TestEventsService } from '@backstage/plugin-events-backend-test-utils';
@@ -92,4 +98,151 @@ describe('eventsPlugin', () => {
       test: 'fake-ext',
     });
   });
+
+  describe('event bus', () => {
+    it('should be possible to publish events as a service', async () => {
+      const { server } = await startTestBackend({
+        features: [eventsPlugin()],
+      });
+
+      await request(server)
+        .post('/api/events/bus/v1/events')
+        .set('authorization', mockCredentials.none.header())
+        .send({ event: { topic: 'test', payload: { n: 1 } } })
+        .expect(401);
+
+      await request(server)
+        .post('/api/events/bus/v1/events')
+        .set('authorization', mockCredentials.user.header())
+        .send({ event: { topic: 'test', payload: { n: 1 } } })
+        .expect(403);
+
+      await request(server)
+        .post('/api/events/bus/v1/events')
+        .set('authorization', mockCredentials.service.header())
+        .send({ event: { topic: 'test', payload: { n: 1 } } })
+        .expect(204); // 204, since there are no subscribers
+    });
+
+    it('should be possible to subscribe as a service and receive an event', async () => {
+      const { server } = await startTestBackend({
+        features: [eventsPlugin()],
+      });
+
+      await request(server)
+        .put('/api/events/bus/v1/subscriptions/tester')
+        .set('authorization', mockCredentials.none.header())
+        .send({ topics: ['test'] })
+        .expect(401);
+
+      await request(server)
+        .put('/api/events/bus/v1/subscriptions/tester')
+        .set('authorization', mockCredentials.user.header())
+        .send({ topics: ['test'] })
+        .expect(403);
+
+      await request(server)
+        .put('/api/events/bus/v1/subscriptions/tester')
+        .set('authorization', mockCredentials.service.header())
+        .send({ topics: ['test'] })
+        .expect(201);
+
+      await request(server)
+        .get('/api/events/bus/v1/subscriptions/tester/events')
+        .set('authorization', mockCredentials.none.header())
+        .send({ topics: ['test'] })
+        .expect(401);
+
+      await request(server)
+        .get('/api/events/bus/v1/subscriptions/tester/events')
+        .set('authorization', mockCredentials.user.header())
+        .send({ topics: ['test'] })
+        .expect(403);
+
+      await request(server)
+        .post('/api/events/bus/v1/events')
+        .set('authorization', mockCredentials.service.header())
+        .send({ event: { topic: 'test', payload: { n: 1 } } })
+        .expect(201); // 201, since there is a subscriber
+
+      await request(server)
+        .get('/api/events/bus/v1/subscriptions/tester/events')
+        .set('authorization', mockCredentials.service.header())
+        .send({ topics: ['test'] })
+        .expect(200, {
+          events: [{ topic: 'test', payload: { n: 1 } }],
+        });
+    });
+  });
+
+  it('should only send an event for each subscriber once', async () => {
+    const { server } = await startTestBackend({
+      features: [eventsPlugin()],
+    });
+
+    // 2 subscribers
+    await request(server)
+      .put('/api/events/bus/v1/subscriptions/tester-1')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] })
+      .expect(201);
+    await request(server)
+      .put('/api/events/bus/v1/subscriptions/tester-2')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] })
+      .expect(201);
+
+    // A single event
+    await request(server)
+      .post('/api/events/bus/v1/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ event: { topic: 'test', payload: { n: 1 } } })
+      .expect(201);
+
+    // Single client for subscriber 1 gets the event
+    await request(server)
+      .get('/api/events/bus/v1/subscriptions/tester-1/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] })
+      .expect(200, {
+        events: [{ topic: 'test', payload: { n: 1 } }],
+      });
+
+    // Two clients for subscriber 2, only one  gets the event
+    const res1 = request(server)
+      .get('/api/events/bus/v1/subscriptions/tester-2/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] });
+    const res2 = request(server)
+      .get('/api/events/bus/v1/subscriptions/tester-2/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] });
+
+    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 request(server)
+      .post('/api/events/bus/v1/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ event: { topic: 'test', payload: { n: 2 } } })
+      .expect(201);
+
+    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 request(server)
+      .get('/api/events/bus/v1/subscriptions/tester-2/events')
+      .set('authorization', mockCredentials.service.header())
+      .send({ topics: ['test'] })
+      .expect(200, {
+        events: [{ topic: 'test', payload: { n: 2 } }],
+      });
+  });
 });