Add crons service tests

express/src/__tests__/crons/crons.service.test.ts

@@ -0,0 +1,146 @@
+import mockMilvusClient from '../../__mocks__/milvus/milvusClient';
+import { schedule } from 'node-cron';
+import { CollectionsService } from '../../collections/collections.service';
+import { CronsService, SchedulerRegistry } from '../../crons/crons.service';
+import { MilvusService } from '../../milvus/milvus.service';
+import { mockAddress } from '../utils/constants';
+import { WS_EVENTS, WS_EVENTS_TYPE } from '../../utils/Const';
+
+// mock Milvus client
+jest.mock('@zilliz/milvus2-sdk-node', () => {
+  return {
+    MilvusClient: mockMilvusClient,
+  };
+});
+
+// mock node-cron
+jest.mock('node-cron', () => {
+  return {
+    schedule: jest.fn(),
+  };
+});
+
+// mock variable
+const mockCronFrequency = '30 00 * * *';
+const mockCronEverySec = '* * * * * *';
+const mockCb = jest.fn();
+const mockName = 'j1';
+const mockSecName = 'everySec';
+
+describe('test crons service', () => {
+  let milvusService: any;
+  let collectionService: any;
+  let cronsService: any;
+  let schedulerRegistry: any;
+
+  const handleStartTask = jest.fn();
+  const handleEndTask = jest.fn();
+
+  beforeAll(async () => {
+    // setup Milvus service and connect to mock Milvus client
+    milvusService = new MilvusService();
+    await milvusService.connectMilvus(mockAddress);
+    collectionService = new CollectionsService(milvusService);
+
+    schedulerRegistry = new SchedulerRegistry([]);
+    cronsService = new CronsService(collectionService, schedulerRegistry);
+  });
+
+  beforeEach(() => {
+    // mock schedule
+    (schedule as jest.Mock).mockImplementationOnce((frequency, callback) => {
+      callback();
+      return {
+        start: handleStartTask,
+        stop: handleEndTask,
+      };
+    });
+  });
+
+  afterAll(() => {
+    milvusService = null;
+    collectionService = null;
+    schedulerRegistry = null;
+    cronsService = null;
+  });
+
+  test('test SchedulerRegistry related methods', async () => {
+    const logSpy = jest.spyOn(console, 'log');
+    schedulerRegistry.setCronJob(mockName, mockCronFrequency, () => mockCb());
+    expect(logSpy).toBeCalledWith(`${mockName}: running a task every seconds`);
+    expect(mockCb).toBeCalledTimes(1);
+    expect(schedule).toBeCalledWith(mockCronFrequency, expect.any(Function));
+
+    const job = schedulerRegistry.getCronJob(mockName);
+    expect(job).toEqual({
+      start: handleStartTask,
+      stop: handleEndTask,
+    });
+
+    schedulerRegistry.setCronJob(mockName, mockCronFrequency, () => mockCb());
+    expect(handleEndTask).toBeCalled();
+
+    schedulerRegistry.setCronJobEverySecond(mockSecName, () => mockCb());
+    expect(schedule).toBeCalledWith(mockCronEverySec, expect.any(Function));
+
+    schedulerRegistry.setCronJob(mockName, mockCronFrequency, () => mockCb());
+    expect(handleEndTask).toBeCalled();
+  });
+
+  test('test CronService related methods', async () => {
+    try {
+      await cronsService.toggleCronJobByName({
+        name: WS_EVENTS.COLLECTION,
+        type: WS_EVENTS_TYPE.STOP,
+      });
+    } catch (err) {
+      expect(err.message).toBe('No existed job entity');
+    }
+
+    await cronsService.toggleCronJobByName({
+      name: WS_EVENTS.COLLECTION,
+      type: WS_EVENTS_TYPE.START,
+    });
+    expect(schedule).toBeCalledWith(mockCronEverySec, expect.any(Function));
+
+    schedulerRegistry.setCronJob(WS_EVENTS.COLLECTION, mockCronFrequency, () =>
+      mockCb()
+    );
+    await cronsService.toggleCronJobByName({
+      name: WS_EVENTS.COLLECTION,
+      type: WS_EVENTS_TYPE.START,
+    });
+
+    expect(handleStartTask).toBeCalled();
+    await cronsService.toggleCronJobByName({
+      name: WS_EVENTS.COLLECTION,
+      type: WS_EVENTS_TYPE.STOP,
+    });
+    expect(handleStartTask).toBeCalled();
+
+    try {
+      await cronsService.toggleCronJobByName({
+        name: mockName,
+        type: WS_EVENTS_TYPE.STOP,
+      });
+    } catch (err) {
+      expect(err.message).toBe('Unsupported event type');
+    }
+  });
+
+  test('test getCollections error', async () => {
+    // reset setup to trigger error
+    cronsService = new CronsService(null, schedulerRegistry);
+
+    try {
+      await cronsService.toggleCronJobByName({
+        name: WS_EVENTS.COLLECTION,
+        type: WS_EVENTS_TYPE.START,
+      });
+    } catch (err) {
+      expect(err).toBeDefined();
+    }
+
+    expect(schedule).toBeCalledWith(mockCronEverySec, expect.any(Function));
+  });
+});

express/src/crons/crons.service.ts

@@ -1,7 +1,7 @@
-import { CollectionsService } from "../collections/collections.service";
-import { WS_EVENTS, WS_EVENTS_TYPE } from "../utils/Const";
-import { schedule, ScheduledTask } from "node-cron";
-import { pubSub } from "../events";
+import { CollectionsService } from '../collections/collections.service';
+import { WS_EVENTS, WS_EVENTS_TYPE } from '../utils/Const';
+import { schedule, ScheduledTask } from 'node-cron';
+import { pubSub } from '../events';
 
 export class CronsService {
   constructor(
@@ -11,13 +11,22 @@ export class CronsService {
 
   async toggleCronJobByName(data: { name: string; type: WS_EVENTS_TYPE }) {
     const { name, type } = data;
-    const cronJobEntity = this.schedulerRegistry.getCronJob(name);
-    if (!cronJobEntity && Number(type) === WS_EVENTS_TYPE.START) {
-      return this.getCollections(WS_EVENTS.COLLECTION);
+
+    switch (name) {
+      case WS_EVENTS.COLLECTION:
+        const cronJobEntity = this.schedulerRegistry.getCronJob(name);
+        if (!cronJobEntity && Number(type) === WS_EVENTS_TYPE.START) {
+          return this.getCollections(WS_EVENTS.COLLECTION);
+        }
+        if (!cronJobEntity) {
+          throw new Error('No existed job entity');
+        }
+        return Number(type) === WS_EVENTS_TYPE.STOP
+          ? cronJobEntity.stop()
+          : cronJobEntity.start();
+      default:
+        throw new Error('Unsupported event type');
     }
-    return Number(type) === WS_EVENTS_TYPE.STOP
-      ? cronJobEntity.stop()
-      : cronJobEntity.start();
   }
 
   async getCollections(name: string) {
@@ -26,8 +35,8 @@ export class CronsService {
         const res = await this.collectionService.getAllCollections();
         // TODO
         // this.eventService.server.emit("COLLECTION", res);
-        pubSub.emit("ws_pubsub", {
-          event: WS_EVENTS.COLLECTION + "",
+        pubSub.emit('ws_pubsub', {
+          event: WS_EVENTS.COLLECTION + '',
           data: res,
         });
         return res;
@@ -54,7 +63,7 @@ export class SchedulerRegistry {
 
   setCronJobEverySecond(name: string, func: () => {}) {
     // The cron job will run every second
-    this.setCronJob(name, "* * * * * *", func);
+    this.setCronJob(name, '* * * * * *', func);
   }
 
   // ┌────────────── second (optional)

express/src/events/index.ts

@@ -20,6 +20,7 @@ export class PubSub {
     // const handlerArgs = Array.prototype.slice.call(arguments, 1);
     if (!(eventType in this.handlers)) {
       console.warn(`eventType: ${eventType} missing`);
+      // throw new Error(`eventType: ${eventType} missing`);
       return;
     }
     this.handlers[eventType].forEach((handler) => {