test(repeat): fix flaky tests (#1578)

taskforcesh · Dec 9, 2022 · 8f9eefd · 8f9eefd
1 parent 6d9dada
commit 8f9eefd
Show file tree

Hide file tree

Showing 4 changed files with 112 additions and 66 deletions.
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
@@ -214,25 +214,6 @@ export class Worker<
 
     this.id = v4();
 
-    this.blockingConnection = new RedisConnection(
-      isRedisInstance(opts.connection)
-        ? (<Redis>opts.connection).duplicate()
-        : opts.connection,
-    );
-    this.blockingConnection.on('error', error => this.emit('error', error));
-
-    this.blockingConnection.on('ready', async () => {
-      try {
-        const client = await this.blockingConnection.client;
-        await client.client('SETNAME', this.clientName(WORKER_SUFFIX));
-      } catch (error) {
-        if (!clientCommandMessageReg.test((<Error>error).message)) {
-          this.emit('error', <Error>error);
-        }
-      }
-      this.emit('ready');
-    });
-
     if (processor) {
       if (typeof processor === 'function') {
         this.processFn = processor;
@@ -267,6 +248,25 @@ export class Worker<
         this.run().catch(error => this.emit('error', error));
       }
     }
+
+    this.blockingConnection = new RedisConnection(
+      isRedisInstance(opts.connection)
+        ? (<Redis>opts.connection).duplicate()
+        : opts.connection,
+    );
+    this.blockingConnection.on('error', error => this.emit('error', error));
+
+    this.blockingConnection.on('ready', async () => {
+      try {
+        const client = await this.blockingConnection.client;
+        await client.client('SETNAME', this.clientName(WORKER_SUFFIX));
+      } catch (error) {
+        if (!clientCommandMessageReg.test((<Error>error).message)) {
+          this.emit('error', <Error>error);
+        }
+      }
+      this.emit('ready');
+    });
   }
 
   emit<U extends keyof WorkerListener<DataType, ResultType, NameType>>(
@@ -456,7 +456,7 @@ export class Worker<
     } else {
       if (this.limitUntil) {
         // TODO: We need to be able to break this delay when we are closing the worker.
-        await delay(this.limitUntil);
+        await this.delay(this.limitUntil);
       }
       return this.moveToActive(token);
     }
@@ -539,8 +539,8 @@ export class Worker<
    *
    * This function is exposed only for testing purposes.
    */
-  async delay(): Promise<void> {
-    await delay(DELAY_TIME_1);
+  async delay(milliseconds?: number): Promise<void> {
+    await delay(milliseconds || DELAY_TIME_1);
   }
 
   protected async nextJobFromJobData(
@@ -789,7 +789,7 @@ export class Worker<
       } catch (err) {
         this.emit('error', <Error>err);
         if (delayInMs) {
-          await delay(delayInMs);
+          await this.delay(delayInMs);
         } else {
           return;
         }

diff --git a/tests/test_flow.ts b/tests/test_flow.ts
@@ -809,6 +809,7 @@ describe('flows', () => {
   describe('when continually adding jobs', async () => {
     it('adds jobs that do not exists', async () => {
       const worker = new Worker(queueName, async () => {}, {
+        autorun: false,
         connection,
       });
 
@@ -855,6 +856,8 @@ describe('flows', () => {
         ],
       });
 
+      worker.run();
+
       await completing1;
 
       const completing2 = new Promise<void>(resolve => {

diff --git a/tests/test_getters.ts b/tests/test_getters.ts
@@ -47,7 +47,10 @@ describe('Jobs getters', function () {
 
   describe('.getWorkers', () => {
     it('gets all workers for this queue only', async function () {
-      const worker = new Worker(queueName, async () => {}, { connection });
+      const worker = new Worker(queueName, async () => {}, {
+        autorun: false,
+        connection,
+      });
       await new Promise<void>(resolve => {
         worker.on('ready', () => {
           resolve();
@@ -57,7 +60,10 @@ describe('Jobs getters', function () {
       const workers = await queue.getWorkers();
       expect(workers).to.have.length(1);
 
-      const worker2 = new Worker(queueName, async () => {}, { connection });
+      const worker2 = new Worker(queueName, async () => {}, {
+        autorun: false,
+        connection,
+      });
       await new Promise<void>(resolve => {
         worker2.on('ready', () => {
           resolve();
@@ -74,13 +80,19 @@ describe('Jobs getters', function () {
     it('gets only workers related only to one queue', async function () {
       const queueName2 = `${queueName}2`;
       const queue2 = new Queue(queueName2, { connection });
-      const worker = new Worker(queueName, async () => {}, { connection });
+      const worker = new Worker(queueName, async () => {}, {
+        autorun: false,
+        connection,
+      });
       await new Promise<void>(resolve => {
         worker.on('ready', () => {
           resolve();
         });
       });
-      const worker2 = new Worker(queueName2, async () => {}, { connection });
+      const worker2 = new Worker(queueName2, async () => {}, {
+        autorun: false,
+        connection,
+      });
       await new Promise<void>(resolve => {
         worker2.on('ready', () => {
           resolve();
@@ -103,10 +115,14 @@ describe('Jobs getters', function () {
       it('gets same reference for all workers for same queue', async function () {
         const ioredisConnection = new IORedis({ maxRetriesPerRequest: null });
         const worker = new Worker(queueName, async () => {}, {
+          autorun: false,
           connection: ioredisConnection,
         });
-        await worker.waitUntilReady();
-        await delay(10);
+        await new Promise<void>(resolve => {
+          worker.on('ready', () => {
+            resolve();
+          });
+        });
 
         const workers = await queue.getWorkers();
         expect(workers).to.have.length(1);
@@ -115,7 +131,6 @@ describe('Jobs getters', function () {
           connection: ioredisConnection,
         });
         await worker2.waitUntilReady();
-        await delay(10);
 
         const nextWorkers = await queue.getWorkers();
         expect(nextWorkers).to.have.length(1);
@@ -129,10 +144,15 @@ describe('Jobs getters', function () {
     describe('when disconnection happens', () => {
       it('gets all workers even after reconnection', async function () {
         const worker = new Worker(queueName, async () => {}, {
+          autorun: false,
           connection,
         });
+        await new Promise<void>(resolve => {
+          worker.on('ready', () => {
+            resolve();
+          });
+        });
         const client = await worker.waitUntilReady();
-        await delay(10);
 
         const workers = await queue.getWorkers();
         expect(workers).to.have.length(1);