test: implement unified thread entities

Author: dariakp

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

@@ -51,6 +51,35 @@ export interface UnifiedChangeStream extends ChangeStream {
   eventCollector: InstanceType<typeof import('../../tools/utils')['EventCollector']>;
 }
 
+export class UnifiedThread {
+  #promise: Promise<void>;
+  #error: Error;
+  #killed = false;
+
+  id: string;
+
+  constructor(id) {
+    this.id = id;
+    this.#promise = Promise.resolve();
+  }
+
+  queue(functionToQueue: () => Promise<any>) {
+    if (this.#killed || this.#error) {
+      return;
+    }
+
+    this.#promise = this.#promise.then(functionToQueue).catch(e => (this.#error = e));
+  }
+
+  async finish() {
+    this.#killed = true;
+    await this.#promise;
+    if (this.#error) {
+      throw this.#error;
+    }
+  }
+}
+
 export type CommandEvent = CommandStartedEvent | CommandSucceededEvent | CommandFailedEvent;
 export type CmapEvent =
   | ConnectionPoolCreatedEvent
@@ -285,6 +314,7 @@ export type EntityCtor =
   | typeof ChangeStream
   | typeof AbstractCursor
   | typeof GridFSBucket
+  | typeof UnifiedThread
   | ClientEncryption;
 
 export type EntityTypeId =
@@ -293,6 +323,7 @@ export type EntityTypeId =
   | 'collection'
   | 'session'
   | 'bucket'
+  | 'thread'
   | 'cursor'
   | 'stream'
   | 'clientEncryption';
@@ -303,6 +334,7 @@ ENTITY_CTORS.set('db', Db);
 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);
 
@@ -335,6 +367,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: 'thread', key: string, assertExists?: boolean): UnifiedThread;
   getEntity(type: 'cursor', key: string, assertExists?: boolean): AbstractCursor;
   getEntity(type: 'stream', key: string, assertExists?: boolean): UnifiedChangeStream;
   getEntity(type: 'clientEncryption', key: string, assertExists?: boolean): ClientEncryption;
@@ -474,6 +507,8 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
         }
 
         map.set(entity.bucket.id, new GridFSBucket(db, options));
+      } else if ('thread' in entity) {
+        map.set(entity.thread.id, new UnifiedThread(entity.thread.id));
       } else if ('stream' in entity) {
         throw new Error(`Unsupported Entity ${JSON.stringify(entity)}`);
       } else if ('clientEncryption' in entity) {

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

@@ -572,6 +572,31 @@ operations.set('waitForPrimaryChange', async ({ entities, operation }) => {
   ]);
 });
 
+operations.set('runOnThread', async ({ entities, operation, client, testConfig }) => {
+  const threadId: string = operation.arguments!.thread;
+  expect(threadId).to.be.a('string');
+  const thread = entities.getEntity('thread', threadId, true);
+  const operationToQueue = operation.arguments!.operation;
+  const executeFn = executeOperationAndCheck.bind(
+    null,
+    operationToQueue,
+    entities,
+    client,
+    testConfig
+  );
+  thread.queue(executeFn);
+});
+
+operations.set('waitForThread', async ({ entities, operation }) => {
+  const threadId: string = operation.arguments!.thread;
+  expect(threadId).to.be.a('string');
+  const thread = entities.getEntity('thread', threadId, true);
+  await Promise.race([
+    thread.finish(),
+    sleep(10000).then(() => Promise.reject(new Error(`Timed out waiting for thread: ${threadId}`)))
+  ]);
+});
+
 operations.set('withTransaction', async ({ entities, operation, client, testConfig }) => {
   const session = entities.getEntity('session', operation.object);
 

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

@@ -5,6 +5,7 @@ import type { ReadPreferenceMode } from '../../../src/read_preference';
 import type { TagSet } from '../../../src/sdam/server_description';
 import type { W } from '../../../src/write_concern';
 import { TestConfiguration } from '../runner/config';
+import { UnifiedThread } from './entities';
 
 export const SupportedVersion = '^1.0';
 
@@ -199,6 +200,7 @@ export type EntityDescription =
   | { database: DatabaseEntity }
   | { collection: CollectionEntity }
   | { bucket: BucketEntity }
+  | { thread: UnifiedThread }
   | { stream: StreamEntity }
   | { session: SessionEntity }
   | { clientEncryption: ClientEncryptionEntity };