test(NODE-4783): add half integration test

Author: durran

test/tools/utils.ts

@@ -354,6 +354,18 @@ export class TestBuilder {
   }
 }
 
+export function bufferToStream(buffer) {
+  const stream = new Readable();
+  if (Array.isArray(buffer)) {
+    buffer.forEach(b => stream.push(b));
+  } else {
+    stream.push(buffer);
+  }
+
+  stream.push(null);
+  return stream;
+}
+
 export function generateOpMsgBuffer(document: Document): Buffer {
   const header = Buffer.alloc(4 * 4 + 4);
 

test/unit/cmap/connection.test.ts

@@ -1,7 +1,8 @@
 import { expect } from 'chai';
-import { EventEmitter } from 'events';
+import { EventEmitter, on } from 'events';
 import { Socket } from 'net';
 import * as sinon from 'sinon';
+import { Readable } from 'stream';
 import { setTimeout } from 'timers';
 
 import { BinMsg } from '../../../src/cmap/commands';
@@ -165,11 +166,44 @@ describe('new Connection()', function () {
 
       beforeEach(function () {
         driverSocket = sinon.spy(new FakeSocket());
-        // @ts-expect-error: driverSocket does not fully satisfy the stream type, but that's okay
-        connection = sinon.spy(new Connection(driverSocket, connectionOptionsDefaults));
-        connection.isMonitoringConnection = true;
-        const queueSymbol = getSymbolFrom(connection, 'queue');
-        queue = connection[queueSymbol];
+      });
+
+      context('when multiple hellos exist on the stream', function () {
+        let callbackSpy;
+        const inputStream = new Readable();
+        const document = { ok: 1 };
+
+        beforeEach(function () {
+          callbackSpy = sinon.spy();
+          const firstHello = generateOpMsgBuffer(document);
+          const secondHello = generateOpMsgBuffer(document);
+          const thirdHello = generateOpMsgBuffer(document);
+          const buffer = Buffer.concat([firstHello, secondHello, thirdHello]);
+
+          // @ts-expect-error: driverSocket does not fully satisfy the stream type, but that's okay
+          connection = sinon.spy(new Connection(inputStream, connectionOptionsDefaults));
+          connection.isMonitoringConnection = true;
+          const queueSymbol = getSymbolFrom(connection, 'queue');
+          queue = connection[queueSymbol];
+
+          // Create the operation description.
+          const operationDescription: OperationDescription = {
+            requestId: 1,
+            cb: callbackSpy
+          };
+
+          // Stick an operation description in the queue.
+          queue.set(1, operationDescription);
+
+          // Push the buffer of 3 hellos to the input stream
+          inputStream.push(buffer);
+          inputStream.push(null);
+        });
+
+        it('calls the operation description callback with the document', async function () {
+          await on(inputStream, 'message');
+          expect(callbackSpy).to.be.calledOnceWith(undefined, document);
+        });
       });
 
       context('when requestId/responseTo do not match', function () {
@@ -178,6 +212,13 @@ describe('new Connection()', function () {
 
         beforeEach(function () {
           callbackSpy = sinon.spy();
+
+          // @ts-expect-error: driverSocket does not fully satisfy the stream type, but that's okay
+          connection = sinon.spy(new Connection(driverSocket, connectionOptionsDefaults));
+          connection.isMonitoringConnection = true;
+          const queueSymbol = getSymbolFrom(connection, 'queue');
+          queue = connection[queueSymbol];
+
           // Create the operation description.
           const operationDescription: OperationDescription = {
             requestId: 1,
@@ -201,7 +242,7 @@ describe('new Connection()', function () {
         });
 
         it('calls the operation description callback with the document', function () {
-          expect(callbackSpy).to.be.calledExactlyOnceWith(undefined, document);
+          expect(callbackSpy).to.be.calledOnceWith(undefined, document);
         });
       });
 
@@ -211,6 +252,13 @@ describe('new Connection()', function () {
 
         beforeEach(function () {
           callbackSpy = sinon.spy();
+
+          // @ts-expect-error: driverSocket does not fully satisfy the stream type, but that's okay
+          connection = sinon.spy(new Connection(driverSocket, connectionOptionsDefaults));
+          connection.isMonitoringConnection = true;
+          const queueSymbol = getSymbolFrom(connection, 'queue');
+          queue = connection[queueSymbol];
+
           // Create the operation description.
           const operationDescription: OperationDescription = {
             requestId: 1,
@@ -234,7 +282,7 @@ describe('new Connection()', function () {
         });
 
         it('calls the operation description callback with the document', function () {
-          expect(callbackSpy).to.be.calledExactlyOnceWith(undefined, document);
+          expect(callbackSpy).to.be.calledOnceWith(undefined, document);
         });
       });
     });

test/unit/cmap/message_stream.test.js

@@ -6,19 +6,7 @@ const { MessageStream } = require('../../../src/cmap/message_stream');
 const { Msg } = require('../../../src/cmap/commands');
 const expect = require('chai').expect;
 const { LEGACY_HELLO_COMMAND } = require('../../../src/constants');
-const { generateOpMsgBuffer } = require('../../tools/utils');
-
-function bufferToStream(buffer) {
-  const stream = new Readable();
-  if (Array.isArray(buffer)) {
-    buffer.forEach(b => stream.push(b));
-  } else {
-    stream.push(buffer);
-  }
-
-  stream.push(null);
-  return stream;
-}
+const { bufferToStream, generateOpMsgBuffer } = require('../../tools/utils');
 
 describe('MessageStream', function () {
   context('when the stream is for a monitoring connection', function () {