test: Complete all matching operations

test/functional/change_stream.test.js

@@ -5,6 +5,7 @@ const { MongoNetworkError } = require('../../src/error');
 const { delay, setupDatabase, withClient, withCursor } = require('./shared');
 const co = require('co');
 const mock = require('../tools/mock');
+const { EventCollector } = require('../tools/utils');
 const chai = require('chai');
 const expect = chai.expect;
 const sinon = require('sinon');
@@ -233,56 +234,6 @@ describe('Change Streams', function () {
     }
   });
 
-  class EventCollector {
-    constructor(obj, events, options) {
-      this._events = [];
-      this._timeout = options ? options.timeout : 5000;
-
-      events.forEach(eventName => {
-        this._events[eventName] = [];
-        obj.on(eventName, event => this._events[eventName].push(event));
-      });
-    }
-
-    waitForEvent(eventName, count, callback) {
-      if (typeof count === 'function') {
-        callback = count;
-        count = 1;
-      }
-
-      waitForEventImpl(this, Date.now(), eventName, count, callback);
-    }
-
-    reset(eventName) {
-      if (eventName == null) {
-        Object.keys(this._events).forEach(eventName => {
-          this._events[eventName] = [];
-        });
-
-        return;
-      }
-
-      if (this._events[eventName] == null) {
-        throw new TypeError(`invalid event name "${eventName}" specified for reset`);
-      }
-
-      this._events[eventName] = [];
-    }
-  }
-
-  function waitForEventImpl(collector, start, eventName, count, callback) {
-    const events = collector._events[eventName];
-    if (events.length >= count) {
-      return callback(undefined, events);
-    }
-
-    if (Date.now() - start >= collector._timeout) {
-      return callback(new Error(`timed out waiting for event "${eventName}"`));
-    }
-
-    setTimeout(() => waitForEventImpl(collector, start, eventName, count, callback), 10);
-  }
-
   it('should create a ChangeStream on a collection and emit `change` events', {
     metadata: { requires: { topology: 'replicaset', mongodb: '>=3.6' } },
 

test/functional/unified-spec-runner/entities.ts

@@ -9,11 +9,18 @@ import type {
 } from '../../../src/cmap/events';
 import { patchCollectionOptions, patchDbOptions } from './unified-utils';
 import { TestConfiguration } from './unified.test';
+import { expect } from 'chai';
+
+interface UnifiedChangeStream extends ChangeStream {
+  eventCollector: InstanceType<typeof import('../../tools/utils')['EventCollector']>;
+}
 
 export type CommandEvent = CommandStartedEvent | CommandSucceededEvent | CommandFailedEvent;
 
 export class UnifiedMongoClient extends MongoClient {
   events: CommandEvent[];
+  failPoints: Document[];
+  ignoredEvents: string[];
   observedEvents: ('commandStarted' | 'commandSucceeded' | 'commandFailed')[];
 
   static EVENT_NAME_LOOKUP = {
@@ -25,6 +32,11 @@ export class UnifiedMongoClient extends MongoClient {
   constructor(url: string, description: ClientEntity) {
     super(url, { monitorCommands: true, ...description.uriOptions });
     this.events = [];
+    this.failPoints = [];
+    this.ignoredEvents = [
+      ...(description.ignoreCommandMonitoringEvents ?? []),
+      'configureFailPoint'
+    ];
     // apm
     this.observedEvents = (description.observeEvents ?? []).map(
       e => UnifiedMongoClient.EVENT_NAME_LOOKUP[e]
@@ -34,9 +46,11 @@ export class UnifiedMongoClient extends MongoClient {
     }
   }
 
-  // NOTE: this must be an arrow function for `this` to work.
+  // NOTE: pushEvent must be an arrow function
   pushEvent: (e: CommandEvent) => void = e => {
-    this.events.push(e);
+    if (!this.ignoredEvents.includes(e.commandName)) {
+      this.events.push(e);
+    }
   };
 
   /** Disables command monitoring for the client and returns a list of the captured events. */
@@ -46,14 +60,33 @@ export class UnifiedMongoClient extends MongoClient {
     }
     return this.events;
   }
+
+  async enableFailPoint(failPoint: Document): Promise<Document> {
+    const admin = this.db().admin();
+    const result = await admin.command(failPoint);
+    expect(result).to.have.property('ok', 1);
+    this.failPoints.push(failPoint.configureFailPoint);
+    return result;
+  }
+
+  async disableFailPoints(): Promise<Document[]> {
+    return Promise.all(
+      this.failPoints.map(configureFailPoint =>
+        this.db().admin().command({
+          configureFailPoint,
+          mode: 'off'
+        })
+      )
+    );
+  }
 }
 
 export type Entity =
   | UnifiedMongoClient
   | Db
   | Collection
   | ClientSession
-  | ChangeStream
+  | UnifiedChangeStream
   | GridFSBucket
   | Document; // Results from operations
 
@@ -81,7 +114,7 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
   mapOf(type: 'collection'): EntitiesMap<Collection>;
   mapOf(type: 'session'): EntitiesMap<ClientSession>;
   mapOf(type: 'bucket'): EntitiesMap<GridFSBucket>;
-  mapOf(type: 'stream'): EntitiesMap<ChangeStream>;
+  mapOf(type: 'stream'): EntitiesMap<UnifiedChangeStream>;
   mapOf(type: EntityTypeId): EntitiesMap<Entity> {
     const ctor = ENTITY_CTORS.get(type);
     if (!ctor) {
@@ -95,7 +128,7 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
   getEntity(type: 'collection', key: string, assertExists?: boolean): Collection;
   getEntity(type: 'session', key: string, assertExists?: boolean): ClientSession;
   getEntity(type: 'bucket', key: string, assertExists?: boolean): GridFSBucket;
-  getEntity(type: 'stream', key: string, assertExists?: boolean): ChangeStream;
+  getEntity(type: 'stream', key: string, assertExists?: boolean): UnifiedChangeStream;
   getEntity(type: EntityTypeId, key: string, assertExists = true): Entity {
     const entity = this.get(key);
     if (!entity) {
@@ -114,6 +147,7 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
 
   async cleanup(): Promise<void> {
     for (const [, client] of this.mapOf('client')) {
+      await client.disableFailPoints();
       await client.close();
     }
     for (const [, session] of this.mapOf('session')) {