test: refactor notification tests on the basis of promises rather than polling

web3.js/test/connection.test.ts

@@ -20,7 +20,6 @@ import {
   Message,
 } from '../src';
 import invariant from '../src/util/assert';
-import {DEFAULT_TICKS_PER_SLOT, NUM_TICKS_PER_SECOND} from '../src/timing';
 import {MOCK_PORT, url} from './url';
 import {
   AccountInfo,
@@ -29,8 +28,10 @@ import {
   BlockSignatures,
   Commitment,
   ConfirmedBlock,
+  Context,
   EpochInfo,
   InflationGovernor,
+  Logs,
   SlotInfo,
 } from '../src/connection';
 import {sleep} from '../src/util/sleep';
@@ -3551,17 +3552,28 @@ describe('Connection', function () {
       );
     });
 
-    it('account change notification', async () => {
+    describe('given an open websocket connection', () => {
-      if (mockServer) {
+      beforeEach(async () => {
-        console.log('non-live test skipped');
+        // Open the socket connection and wait for it to become pingable.
-        return;
+        connection._rpcWebSocket.connect();
+        // eslint-disable-next-line no-constant-condition
+        while (true) {
+          try {
+            await connection._rpcWebSocket.notify('ping');
+            break;
+            // eslint-disable-next-line no-empty
+          } catch {}
+          await sleep(100);
         }
+      });
+
+      it('account change notification', async () => {
         const connection = new Connection(url, 'confirmed');
         const owner = Keypair.generate();
 
-      let subscriptionId;
+        let subscriptionId: number | undefined;
         try {
-        const notificationPromise = new Promise<AccountInfo<Buffer>>(
+          const accountInfoPromise = new Promise<AccountInfo<Buffer>>(
             resolve => {
               subscriptionId = connection.onAccountChange(
                 owner.publicKey,
@@ -3571,13 +3583,12 @@ describe('Connection', function () {
             },
           );
           connection.requestAirdrop(owner.publicKey, LAMPORTS_PER_SOL);
-        const notificationPayload = await notificationPromise;
+          const accountInfo = await accountInfoPromise;
-        expect(notificationPayload.lamports).to.eq(LAMPORTS_PER_SOL);
+          expect(accountInfo.lamports).to.eq(LAMPORTS_PER_SOL);
-        expect(notificationPayload.owner.equals(SystemProgram.programId)).to.be
+          expect(accountInfo.owner.equals(SystemProgram.programId)).to.be.true;
-          .true;
         } finally {
           if (subscriptionId != null) {
-          connection.removeAccountChangeListener(subscriptionId);
+            await connection.removeAccountChangeListener(subscriptionId);
           }
         }
       });
@@ -3587,23 +3598,17 @@ describe('Connection', function () {
 
         const owner = Keypair.generate();
         const programAccount = Keypair.generate();
-      const balanceNeeded = await connection.getMinimumBalanceForRentExemption(
+        const balanceNeeded =
-        0,
+          await connection.getMinimumBalanceForRentExemption(0);
-      );
 
-      let notified = false;
+        let subscriptionId: number | undefined;
-      const subscriptionId = connection.onProgramAccountChange(
+        try {
+          const keyedAccountInfoPromise = new Promise<KeyedAccountInfo>(
+            resolve => {
+              subscriptionId = connection.onProgramAccountChange(
                 SystemProgram.programId,
-        (keyedAccountInfo: KeyedAccountInfo) => {
+                resolve,
-          if (keyedAccountInfo.accountId.equals(programAccount.publicKey)) {
+              );
-            expect(keyedAccountInfo.accountInfo.lamports).to.eq(balanceNeeded);
-            expect(
-              keyedAccountInfo.accountInfo.owner.equals(
-                SystemProgram.programId,
-              ),
-            ).to.be.true;
-            notified = true;
-          }
             },
           );
 
@@ -3613,7 +3618,6 @@ describe('Connection', function () {
             amount: LAMPORTS_PER_SOL,
           });
 
-      try {
           const transaction = new Transaction().add(
             SystemProgram.transfer({
               fromPubkey: owner.publicKey,
@@ -3624,97 +3628,83 @@ describe('Connection', function () {
           await sendAndConfirmTransaction(connection, transaction, [owner], {
             commitment: 'confirmed',
           });
-      } catch (err) {
-        await connection.removeProgramAccountChangeListener(subscriptionId);
-        throw err;
-      }
 
-      let i = 0;
+          const keyedAccountInfo = await keyedAccountInfoPromise;
-      while (!notified) {
+          if (keyedAccountInfo.accountId.equals(programAccount.publicKey)) {
-        if (++i === 30) {
+            expect(keyedAccountInfo.accountInfo.lamports).to.eq(balanceNeeded);
-          throw new Error('Program change notification not observed');
+            expect(
+              keyedAccountInfo.accountInfo.owner.equals(
+                SystemProgram.programId,
+              ),
+            ).to.be.true;
           }
-        // Sleep for a 1/4 of a slot, notifications only occur after a block is
+        } finally {
-        // processed
+          if (subscriptionId != null) {
-        await sleep((250 * DEFAULT_TICKS_PER_SLOT) / NUM_TICKS_PER_SECOND);
-      }
-
             await connection.removeProgramAccountChangeListener(subscriptionId);
+          }
+        }
       });
 
       it('slot notification', async () => {
-      let notifiedSlotInfo: SlotInfo | undefined;
+        let subscriptionId: number | undefined;
-      const subscriptionId = connection.onSlotChange(slotInfo => {
+        try {
-        notifiedSlotInfo = slotInfo;
+          const notifiedSlotInfo = await new Promise<SlotInfo>(resolve => {
+            subscriptionId = connection.onSlotChange(resolve);
           });
-
-      // Wait for notification
-      let i = 0;
-      while (!notifiedSlotInfo) {
-        if (++i === 30) {
-          throw new Error('Slot change notification not observed');
-        }
-        // Sleep for a 1/4 of a slot, notifications only occur after a block is
-        // processed
-        await sleep((250 * DEFAULT_TICKS_PER_SLOT) / NUM_TICKS_PER_SECOND);
-      }
-
           expect(notifiedSlotInfo.parent).to.be.at.least(0);
           expect(notifiedSlotInfo.root).to.be.at.least(0);
           expect(notifiedSlotInfo.slot).to.be.at.least(1);
-
+        } finally {
+          if (subscriptionId != null) {
             await connection.removeSlotChangeListener(subscriptionId);
+          }
+        }
       });
 
       it('root notification', async () => {
-      let roots: number[] = [];
+        let subscriptionId: number | undefined;
-      const subscriptionId = connection.onRootChange(root => {
+        try {
+          const atLeastTwoRoots = await new Promise<number[]>(resolve => {
+            const roots: number[] = [];
+            subscriptionId = connection.onRootChange(root => {
+              if (roots.length === 2) {
+                return;
+              }
               roots.push(root);
+              if (roots.length === 2) {
+                // Collect at least two, then resolve.
+                resolve(roots);
+              }
             });
-
+          });
-      // Wait for mockCallback to receive a call
+          expect(atLeastTwoRoots[1]).to.be.greaterThan(atLeastTwoRoots[0]);
-      let i = 0;
+        } finally {
-      while (roots.length < 2) {
+          if (subscriptionId != null) {
-        if (++i === 30) {
-          throw new Error('Root change notification not observed');
-        }
-        // Sleep for a 1/4 of a slot, notifications only occur after a block is
-        // processed
-        await sleep((250 * DEFAULT_TICKS_PER_SLOT) / NUM_TICKS_PER_SECOND);
-      }
-
-      expect(roots[1]).to.be.greaterThan(roots[0]);
             await connection.removeRootChangeListener(subscriptionId);
+          }
+        }
       });
 
       it('logs notification', async () => {
-      let listener: number | undefined;
+        let subscriptionId: number | undefined;
         const owner = Keypair.generate();
-      const [logsRes, ctx] = await new Promise(resolve => {
+        try {
-        let received = false;
+          const logPromise = new Promise<[Logs, Context]>(resolve => {
-        listener = connection.onLogs(
+            subscriptionId = connection.onLogs(
               owner.publicKey,
               (logs, ctx) => {
                 if (!logs.err) {
-              received = true;
                   resolve([logs, ctx]);
                 }
               },
               'processed',
             );
-
-        // Send transactions until the log subscription receives an event
-        (async () => {
-          while (!received) {
-            // Execute a transaction so that we can pickup its logs.
-            await connection.requestAirdrop(
-              owner.publicKey,
-              1 * LAMPORTS_PER_SOL,
-            );
-            await sleep(1000);
-          }
-        })();
           });
+
+          // Execute a transaction so that we can pickup its logs.
+          await connection.requestAirdrop(owner.publicKey, LAMPORTS_PER_SOL);
+
+          const [logsRes, ctx] = await logPromise;
           expect(ctx.slot).to.be.greaterThan(0);
           expect(logsRes.logs.length).to.eq(2);
           expect(logsRes.logs[0]).to.eq(
@@ -3723,8 +3713,13 @@ describe('Connection', function () {
           expect(logsRes.logs[1]).to.eq(
             'Program 11111111111111111111111111111111 success',
           );
-      await connection.removeOnLogsListener(listener!);
+        } finally {
-    }).timeout(60 * 1000);
+          if (subscriptionId != null) {
+            await connection.removeOnLogsListener(subscriptionId);
+          }
+        }
+      });
+    });
 
     it('https request', async () => {
       const connection = new Connection('https://api.mainnet-beta.solana.com');