Test SSE propertly

package.json

@@ -52,6 +52,7 @@
     "@nestjs/schematics": "^9.0.0",
     "@nestjs/testing": "^9.0.0",
     "@types/amqplib": "^0.10.1",
+    "@types/eventsource": "^1.1.11",
     "@types/express": "^4.17.13",
     "@types/jest": "^29.5.3",
     "@types/node": "18.15.11",
@@ -61,6 +62,7 @@
     "eslint": "^8.45.0",
     "eslint-config-prettier": "^8.3.0",
     "eslint-plugin-prettier": "^4.0.0",
+    "eventsource": "^2.0.2",
     "husky": "^8.0.3",
     "jest": "^29.6.1",
     "prettier": "^2.3.2",

src/routes/events/events.controller.spec.ts

@@ -12,7 +12,7 @@ describe('EventsController', () => {
   let service: EventsService;
 
   beforeEach(async () => {
-    const module = await Test.createTestingModule({
+    const module: TestingModule = await Test.createTestingModule({
       controllers: [EventsController],
       providers: [EventsService, QueueProvider, WebhookService],
     })
@@ -63,7 +63,7 @@ describe('EventsController', () => {
       // Not relevant event must be ignored by Safe filter
       const event = await firstValue;
       expect(event.data).toEqual(txServiceEvents[1]);
-      expect(event.type).toEqual(txServiceEvents[1].type);
+      expect(event.type).toEqual('message');
     });
   });
 });

src/routes/events/events.service.spec.ts

@@ -17,7 +17,7 @@ describe('EventsService', () => {
 
   beforeEach(async () => {
     const webhookServiceMock = {
-      postEveryWebhook: async (_: object) => ({
+      postEveryWebhook: async () => ({
         data: {},
         status: 200,
         statusText: 'OK',
@@ -47,6 +47,10 @@ describe('EventsService', () => {
   describe('processEvent', () => {
     it('should post webhooks', async () => {
       const postEveryWebhook = jest.spyOn(webhookService, 'postEveryWebhook');
+      const pushEventToEventsObservable = jest.spyOn(
+        eventsService,
+        'pushEventToEventsObservable',
+      );
       const msg = {
         chainId: '1',
         type: 'SAFE_CREATED' as TxServiceEventType,
@@ -56,12 +60,15 @@ describe('EventsService', () => {
       await eventsService.processEvent(JSON.stringify(msg));
       expect(postEveryWebhook).toBeCalledTimes(1);
       expect(postEveryWebhook).toBeCalledWith(msg);
+      expect(pushEventToEventsObservable).toBeCalledTimes(1);
+      expect(pushEventToEventsObservable).toBeCalledWith(msg);
     });
   });
 
   describe('processMessageEvents', () => {
     it('should post webhooks', async () => {
       const postEveryWebhook = jest.spyOn(webhookService, 'postEveryWebhook');
+
       const messageCreated = {
         chainId: '1',
         type: 'MESSAGE_CREATED' as TxServiceEventType,

src/routes/events/events.service.ts

@@ -46,7 +46,7 @@ export class EventsService implements OnApplicationBootstrap {
     txServiceEvent: TxServiceEvent,
   ): MessageEvent<TxServiceEvent> {
     const messageEvent: MessageEvent<TxServiceEvent> = new MessageEvent(
-      txServiceEvent.type,
+      'message',
       {
         data: txServiceEvent,
       },

test/app.e2e-spec.ts

@@ -4,6 +4,9 @@ import * as request from 'supertest';
 import { AppModule } from './../src/app.module';
 import { QueueProvider } from '../src/datasources/queue/queue.provider';
 import { EventsService } from '../src/routes/events/events.service';
+import { Server } from 'tls';
+import { TxServiceEventType } from '../src/routes/events/event.dto';
+import EventSource = require('eventsource');
 
 /* eslint-disable */
 const { version } = require('../package.json');
@@ -39,17 +42,42 @@ describe('AppController (e2e)', () => {
   });
 
   describe('/events/sse/:safe (GET)', () => {
-    it('should subscribe to server side events', () => {
+    it('should subscribe and receive Server Side Events', () => {
       const validSafeAddress = '0x8618ce407F169ABB1388348A19632AaFA857CCB9';
-      const url = `/events/sse/${validSafeAddress}`;
-      const expected = {};
+      const msg = {
+        chainId: '1',
+        type: 'SAFE_CREATED' as TxServiceEventType,
+        hero: 'Tatsumaki',
+        address: validSafeAddress,
+      };
+
+      const path = `/events/sse/${validSafeAddress}`;
+
+      // Supertest cannot be used, as it does not support EventSource
+      const server = app.getHttpServer();
+      server.listen();
+      const port = server.address().port;
+      const protocol = server instanceof Server ? 'https' : 'http';
+      const url = protocol + '://127.1.0.1:' + port + path;
+
+      const eventSource = new EventSource(url);
+      // Use an empty promise so test has to wait for it
+      const messageReceived = new Promise((resolve) => {
+        eventSource.onmessage = (event) => {
+          expect(event.type).toBe('message');
+          const parsedData = JSON.parse(event.data);
+          expect(parsedData).toStrictEqual(msg);
+          server.close();
+          resolve(null);
+        };
+      });
+
+      // Wait a little to send the message
+      setTimeout(() => {
+        eventsService.pushEventToEventsObservable(msg);
+      }, 1000);
 
-      const result = request(app.getHttpServer())
-        .get(url)
-        .expect(200)
-        .expect(expected);
-      eventsService.completeEventsObservable();
-      return result;
+      return messageReceived;
     });
     it('should return a 400 if safe address is not EIP55 valid', () => {
       const notValidAddress = '0x8618CE407F169ABB1388348A19632AaFA857CCB9';