test(NODE-3049): drivers atlas testing

Author: durran

package.json

@@ -143,6 +143,7 @@
     "check:unit": "mocha test/unit",
     "check:ts": "node ./node_modules/typescript/bin/tsc -v && node ./node_modules/typescript/bin/tsc --noEmit",
     "check:atlas": "mocha --config test/manual/mocharc.json test/manual/atlas_connectivity.test.js",
+    "check:drivers-atlas-testing": "mocha --config test/mocha_mongodb.json test/atlas/drivers_atlas_testing.test.ts",
     "check:adl": "mocha --config test/mocha_mongodb.json test/manual/atlas-data-lake-testing",
     "check:aws": "nyc mocha --config test/mocha_mongodb.json test/integration/auth/mongodb_aws.test.ts",
     "check:oidc": "mocha --config test/mocha_mongodb.json test/manual/mongodb_oidc.prose.test.ts",

test/atlas/drivers_atlas_testing.test.ts

@@ -0,0 +1,7 @@
+import { runUnifiedSuite } from '../tools/unified-spec-runner/runner';
+
+describe('Node Driver Atlas Testing', async function () {
+  // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+  const spec = JSON.parse(process.env.WORKLOAD_SPECIFICATION!);
+  runUnifiedSuite([spec]);
+});

test/tools/reporter/mongodb_reporter.js

@@ -103,7 +103,7 @@ class MongoDBMochaReporter extends mocha.reporters.Spec {
       catchErr(test => this.testEnd(test))
     );
 
-    process.on('SIGINT', () => this.end(true));
+    process.prependListener('SIGINT', () => this.end(true));
   }
 
   start() {}
@@ -183,7 +183,11 @@ class MongoDBMochaReporter extends mocha.reporters.Spec {
     } catch (error) {
       console.error(chalk.red(`Failed to output xunit report! ${error}`));
     } finally {
-      if (ctrlC) process.exit(1);
+      // Dont exit the process on Astrolabe testing, let it interrupt and
+      // finish naturally.
+      if (!process.env.WORKLOAD_SPECIFICATION) {
+        process.exit(1);
+      }
     }
   }
 

test/tools/runner/config.ts

@@ -159,6 +159,11 @@ export class TestConfiguration {
   }
 
   newClient(dbOptions?: string | Record<string, any>, serverOptions?: Record<string, any>) {
+    if (process.env.DRIVERS_ATLAS_TESTING_URI) {
+      console.log('Using drivers Atlas testing URI', process.env.DRIVERS_ATLAS_TESTING_URI);
+      return new MongoClient(process.env.DRIVERS_ATLAS_TESTING_URI);
+    }
+
     serverOptions = Object.assign({}, getEnvironmentalOptions(), serverOptions);
 
     // support MongoClient constructor form (url, options) for `newClient`
@@ -258,6 +263,11 @@ export class TestConfiguration {
       ...options
     };
 
+    if (process.env.DRIVERS_ATLAS_TESTING_URI) {
+      console.log('Using drivers Atlas testing URI', process.env.DRIVERS_ATLAS_TESTING_URI);
+      return process.env.DRIVERS_ATLAS_TESTING_URI;
+    }
+
     const FILLER_HOST = 'fillerHost';
 
     const protocol = this.isServerless ? 'mongodb+srv' : 'mongodb';

test/tools/unified-spec-runner/astrolabe_results_writer.ts

@@ -0,0 +1,44 @@
+import { writeFile } from 'node:fs/promises';
+
+import * as path from 'path';
+
+import type { EntitiesMap } from './entities';
+import { trace } from './runner';
+
+/**
+ * Writes the entities saved from the loop operations run in the
+ * Astrolabe workload executor to the required files.
+ */
+export class AstrolabeResultsWriter {
+  constructor(private entities: EntitiesMap) {
+    this.entities = entities;
+  }
+
+  async write(): Promise<void> {
+    // Write the events.json to the execution directory.
+    const errors = this.entities.getEntity('errors', 'errors', false);
+    const failures = this.entities.getEntity('failures', 'failures', false);
+    const events = this.entities.getEntity('events', 'events', false);
+    const iterations = this.entities.getEntity('iterations', 'iterations', false);
+    const successes = this.entities.getEntity('successes', 'successes', false);
+
+    // Write the events.json to the execution directory.
+    trace('writing events.json');
+    await writeFile(
+      path.join(process.env.OUTPUT_DIRECTORY ?? '', 'events.json'),
+      JSON.stringify({ events: events ?? [], errors: errors ?? [], failures: failures ?? [] })
+    );
+
+    // Write the results.json to the execution directory.
+    trace('writing results.json');
+    await writeFile(
+      path.join(process.env.OUTPUT_DIRECTORY ?? '', 'results.json'),
+      JSON.stringify({
+        numErrors: errors?.length ?? 0,
+        numFailures: failures?.length ?? 0,
+        numSuccesses: successes ?? 0,
+        numIterations: iterations ?? 0
+      })
+    );
+  }
+}

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

@@ -5,6 +5,7 @@ import { EventEmitter } from 'events';
 import {
   AbstractCursor,
   ChangeStream,
+  ClientEncryption,
   ClientSession,
   Collection,
   type CommandFailedEvent,
@@ -37,13 +38,9 @@ import {
 } from '../../mongodb';
 import { ejson, getEnvironmentalOptions } from '../../tools/utils';
 import type { TestConfiguration } from '../runner/config';
+import { EntityEventRegistry } from './entity_event_registry';
 import { trace } from './runner';
-import type {
-  ClientEncryption,
-  ClientEntity,
-  EntityDescription,
-  ExpectedLogMessage
-} from './schema';
+import type { ClientEntity, EntityDescription, ExpectedLogMessage } from './schema';
 import {
   createClientEncryption,
   makeConnectionString,
@@ -357,9 +354,10 @@ export type Entity =
   | AbstractCursor
   | UnifiedChangeStream
   | GridFSBucket
+  | Document
   | ClientEncryption
   | TopologyDescription // From recordTopologyDescription operation
-  | Document; // Results from operations
+  | number;
 
 export type EntityCtor =
   | typeof UnifiedMongoClient
@@ -370,7 +368,7 @@ export type EntityCtor =
   | typeof AbstractCursor
   | typeof GridFSBucket
   | typeof UnifiedThread
-  | ClientEncryption;
+  | typeof ClientEncryption;
 
 export type EntityTypeId =
   | 'client'
@@ -381,18 +379,26 @@ export type EntityTypeId =
   | 'thread'
   | 'cursor'
   | 'stream'
-  | 'clientEncryption';
+  | 'clientEncryption'
+  | 'errors'
+  | 'failures'
+  | 'events'
+  | 'iterations'
+  | 'successes';
 
 const ENTITY_CTORS = new Map<EntityTypeId, EntityCtor>();
 ENTITY_CTORS.set('client', UnifiedMongoClient);
 ENTITY_CTORS.set('db', Db);
+ENTITY_CTORS.set('clientEncryption', ClientEncryption);
 ENTITY_CTORS.set('collection', Collection);
 ENTITY_CTORS.set('session', ClientSession);
 ENTITY_CTORS.set('bucket', GridFSBucket);
 ENTITY_CTORS.set('thread', UnifiedThread);
 ENTITY_CTORS.set('cursor', AbstractCursor);
 ENTITY_CTORS.set('stream', ChangeStream);
 
+const NO_INSTANCE_CHECK = ['errors', 'failures', 'events', 'successes', 'iterations'];
+
 export class EntitiesMap<E = Entity> extends Map<string, E> {
   failPoints: FailPointMap;
 
@@ -435,19 +441,25 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
   getEntity(type: 'thread', key: string, assertExists?: boolean): UnifiedThread;
   getEntity(type: 'cursor', key: string, assertExists?: boolean): AbstractCursor;
   getEntity(type: 'stream', key: string, assertExists?: boolean): UnifiedChangeStream;
+  getEntity(type: 'iterations', key: string, assertExists?: boolean): number;
+  getEntity(type: 'successes', key: string, assertExists?: boolean): number;
+  getEntity(type: 'errors', key: string, assertExists?: boolean): Document[];
+  getEntity(type: 'failures', key: string, assertExists?: boolean): Document[];
+  getEntity(type: 'events', key: string, assertExists?: boolean): Document[];
   getEntity(type: 'clientEncryption', key: string, assertExists?: boolean): ClientEncryption;
   getEntity(type: EntityTypeId, key: string, assertExists = true): Entity | undefined {
     const entity = this.get(key);
     if (!entity) {
       if (assertExists) throw new Error(`Entity '${key}' does not exist`);
       return;
     }
-    if (type === 'clientEncryption') {
-      // we do not have instanceof checking here since csfle might not be installed
+    if (NO_INSTANCE_CHECK.includes(type)) {
+      // Skip constructor checks for interfaces.
       return entity;
     }
     const ctor = ENTITY_CTORS.get(type);
     if (!ctor) {
+      console.log('CTORS', type, ENTITY_CTORS);
       throw new Error(`Unknown type ${type}`);
     }
     if (!(entity instanceof ctor)) {
@@ -499,6 +511,7 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
           entity.client.uriOptions
         );
         const client = new UnifiedMongoClient(uri, entity.client);
+        new EntityEventRegistry(client, entity.client, map).register();
         try {
           await client.connect();
         } catch (error) {

test/tools/unified-spec-runner/entity_event_registry.ts

@@ -0,0 +1,76 @@
+import {
+  COMMAND_FAILED,
+  COMMAND_STARTED,
+  COMMAND_SUCCEEDED,
+  CONNECTION_CHECK_OUT_FAILED,
+  CONNECTION_CHECK_OUT_STARTED,
+  CONNECTION_CHECKED_IN,
+  CONNECTION_CHECKED_OUT,
+  CONNECTION_CLOSED,
+  CONNECTION_CREATED,
+  CONNECTION_POOL_CLEARED,
+  CONNECTION_POOL_CLOSED,
+  CONNECTION_POOL_CREATED,
+  CONNECTION_POOL_READY,
+  CONNECTION_READY
+} from '../../mongodb';
+import { type EntitiesMap, type UnifiedMongoClient } from './entities';
+import { type ClientEntity } from './schema';
+
+/**
+ * Maps the names of the events the unified runner passes and maps
+ * them to the names of the events emitted in the driver.
+ */
+const MAPPINGS = {
+  PoolCreatedEvent: CONNECTION_POOL_CREATED,
+  PoolReadyEvent: CONNECTION_POOL_READY,
+  PoolClearedEvent: CONNECTION_POOL_CLEARED,
+  PoolClosedEvent: CONNECTION_POOL_CLOSED,
+  ConnectionCreatedEvent: CONNECTION_CREATED,
+  ConnectionReadyEvent: CONNECTION_READY,
+  ConnectionClosedEvent: CONNECTION_CLOSED,
+  ConnectionCheckOutStartedEvent: CONNECTION_CHECK_OUT_STARTED,
+  ConnectionCheckOutFailedEvent: CONNECTION_CHECK_OUT_FAILED,
+  ConnectionCheckedOutEvent: CONNECTION_CHECKED_OUT,
+  ConnectionCheckedInEvent: CONNECTION_CHECKED_IN,
+  CommandStartedEvent: COMMAND_STARTED,
+  CommandSucceededEvent: COMMAND_SUCCEEDED,
+  CommandFailedEvent: COMMAND_FAILED
+};
+
+/**
+ * Registers events that need to be stored in the entities map, since
+ * the UnifiedMongoClient does not contain a ciclical dependency on the
+ * entities map itself.
+ */
+export class EntityEventRegistry {
+  constructor(
+    private client: UnifiedMongoClient,
+    private clientEntity: ClientEntity,
+    private entitiesMap: EntitiesMap
+  ) {
+    this.client = client;
+    this.clientEntity = clientEntity;
+    this.entitiesMap = entitiesMap;
+  }
+
+  /**
+   * Connect the event listeners on the client and the entities map.
+   */
+  register(): void {
+    if (this.clientEntity.storeEventsAsEntities) {
+      for (const { id, events } of this.clientEntity.storeEventsAsEntities) {
+        this.entitiesMap.set(id, []);
+        for (const eventName of events) {
+          // Need to map the event names to the Node event names.
+          this.client.on(MAPPINGS[eventName], () => {
+            this.entitiesMap.getEntity('events', id).push({
+              name: eventName,
+              observedAt: Date.now()
+            });
+          });
+        }
+      }
+    }
+  }
+}

test/tools/unified-spec-runner/operations.ts

@@ -2,7 +2,7 @@
 /* eslint-disable @typescript-eslint/no-non-null-assertion */
 import { once, Writable } from 'node:stream';
 
-import { expect } from 'chai';
+import { AssertionError, expect } from 'chai';
 
 import {
   AbstractCursor,
@@ -377,6 +377,79 @@ operations.set('listIndexes', async ({ entities, operation }) => {
   return collection.listIndexes(operation.arguments!).toArray();
 });
 
+operations.set('loop', async ({ entities, operation, client, testConfig }) => {
+  const controller = new AbortController();
+  // We always want the process to exit on SIGINT last, so all other
+  // SIGINT events listeners must be prepended.
+  process.prependListener('SIGINT', () => {
+    console.log('Receiving SIGINT to process.');
+    controller.abort('Process received SIGINT, aborting operation loop.');
+  });
+  const args = operation.arguments!;
+  const storeIterationsAsEntity = args.storeIterationsAsEntity;
+  const storeSuccessesAsEntity = args.storeSuccessesAsEntity;
+  const storeErrorsAsEntity = args.storeErrorsAsEntity;
+  const storeFailuresAsEntity = args.storeFailuresAsEntity;
+
+  if (storeErrorsAsEntity) {
+    entities.set(storeErrorsAsEntity, []);
+  }
+  if (storeFailuresAsEntity) {
+    entities.set(storeFailuresAsEntity, []);
+  }
+
+  let iterations = 0;
+  let successes = 0;
+  while (!controller.signal.aborted) {
+    if (storeIterationsAsEntity) {
+      entities.set(storeIterationsAsEntity, iterations++);
+    }
+    for (const op of args.operations) {
+      console.log('op', op);
+      try {
+        await executeOperationAndCheck(op, entities, client, testConfig);
+        if (storeSuccessesAsEntity) {
+          entities.set(storeSuccessesAsEntity, successes++);
+        }
+      } catch (error) {
+        console.log('error', error);
+        // From the unified spec:
+        // If neither storeErrorsAsEntity nor storeFailuresAsEntity are specified,
+        // the loop MUST terminate and raise the error/failure (i.e. the error/failure
+        // will interrupt the test).
+        if (!storeErrorsAsEntity && !storeFailuresAsEntity) {
+          entities.set('errors', [
+            {
+              error: 'Neither storeErrorsAsEntity or storeFailuresAsEntity specified',
+              time: Date.now()
+            }
+          ]);
+          controller.abort('Neither storeErrorsAsEntity or storeFailuresAsEntity specified');
+          return;
+        }
+
+        // From the unified spec format specification for the loop operation:
+        // A failure is when the result or outcome of an operation executed by the test
+        // runner differs from its expected outcome. For example, an expectResult assertion
+        // failing to match a BSON document or an expectError assertion failing to match
+        // an error message would be considered a failure.
+        // An error is any other type of error raised by the test runner. For example, an
+        // unsupported operation or inability to resolve an entity name would be considered
+        // an error.
+        if (storeFailuresAsEntity && error instanceof AssertionError) {
+          entities
+            .getEntity('failures', storeFailuresAsEntity)
+            .push({ error: error.message, time: Date.now() });
+        } else if (storeErrorsAsEntity) {
+          entities
+            .getEntity('errors', storeErrorsAsEntity)
+            .push({ error: error.message, time: Date.now() });
+        }
+      }
+    }
+  }
+});
+
 operations.set('replaceOne', async ({ entities, operation }) => {
   const collection = entities.getEntity('collection', operation.object);
   const { filter, replacement, ...opts } = operation.arguments!;