test: move existing int tests to unit

Author: andymina

src/change_stream.ts

@@ -752,7 +752,7 @@ export class ChangeStream<
       try {
         await this.close();
       } catch (error) {
-        // we're not concerned with errors from close()
+        // We're not concerned with errors from close()
       }
     }
   }

test/integration/change-streams/change_stream.test.ts

@@ -10,12 +10,12 @@ import { promisify } from 'util';
 import {
   AbstractCursor,
   ChangeStream,
+  ChangeStreamDocument,
   ChangeStreamOptions,
   Collection,
   CommandStartedEvent,
   Db,
   Long,
-  MongoAPIError,
   MongoChangeStreamError,
   MongoClient,
   MongoServerError,
@@ -989,137 +989,45 @@ describe('Change Streams', function () {
         async function () {
           changeStream = collection.watch([]);
           await initIteratorMode(changeStream);
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
 
           const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
           await collection.insertMany(docs);
 
-          for (const doc of docs) {
-            const change = await changeStreamIterator.next();
-            const { fullDocument } = change.value;
-            expect(fullDocument.city).to.equal(doc.city);
+          for await (const change of changeStream) {
+            const { fullDocument } = change;
+            const expectedDoc = docs.shift();
+            expect(fullDocument.city).to.equal(expectedDoc.city);
+            if (docs.length === 0) {
+              break;
+            }
           }
+          expect(docs).to.have.length(0, 'expected to find all docs before exiting loop');
         }
       );
 
       it(
-        'should close the change stream when return is called',
+        'cannot resume from partial iteration',
         { requires: { topology: '!single' } },
         async function () {
           changeStream = collection.watch([]);
           await initIteratorMode(changeStream);
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
 
           const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
           await collection.insertMany(docs);
 
-          await changeStreamIterator.next();
-          await changeStreamIterator.return();
-          expect(changeStream.closed).to.be.true;
-          expect(changeStream.cursor.closed).to.be.true;
-        }
-      );
-
-      it(
-        'should close the change stream when an error is thrown',
-        { requires: { topology: '!single', mongodb: '>=4.2' } },
-        async function () {
-          changeStream = collection.watch([]);
-          await initIteratorMode(changeStream);
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
-
-          const unresumableErrorCode = 1000;
-          await client.db('admin').command({
-            configureFailPoint: is4_2Server(this.configuration.version)
-              ? 'failCommand'
-              : 'failGetMoreAfterCursorCheckout',
-            mode: { times: 1 },
-            data: {
-              failCommands: ['getMore'],
-              errorCode: unresumableErrorCode
-            }
-          } as FailPoint);
-
-          await collection.insertOne({ city: 'New York City' });
-          try {
-            await changeStreamIterator.next();
-            expect.fail(
-              'Change stream did not throw unresumable error and did not produce any events'
-            );
-          } catch (error) {
-            expect(changeStream.closed).to.be.true;
-            expect(changeStream.cursor.closed).to.be.true;
+          for await (const change of changeStream) {
+            const { fullDocument } = change;
+            const expectedDoc = docs.shift();
+            expect(fullDocument.city).to.equal(expectedDoc.city);
+            break;
           }
-        }
-      );
-
-      it(
-        'should not produce events on closed stream',
-        { requires: { topology: '!single' } },
-        async function () {
-          changeStream = collection.watch([]);
-          changeStream.close();
 
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
-          const change = await changeStreamIterator.next();
-
-          expect(change.value).to.be.undefined;
-        }
-      );
-
-      it(
-        'cannot be used with emitter-based iteration',
-        { requires: { topology: '!single' } },
-        async function () {
-          changeStream = collection.watch([]);
-          changeStream.on('change', sinon.stub());
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
-
-          const error = await changeStreamIterator.next().catch(e => e);
-          expect(error).to.be.instanceOf(MongoAPIError);
-        }
-      );
-
-      it(
-        'can be used with raw iterator API',
-        { requires: { topology: '!single' } },
-        async function () {
-          changeStream = collection.watch([]);
-          await initIteratorMode(changeStream);
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
-
-          const docs = [{ city: 'Los Angeles' }, { city: 'Miami' }];
-          await collection.insertMany(docs);
-
-          await changeStream.next();
-
-          try {
-            const change = await changeStreamIterator.next();
-            expect(change.value).to.not.be.undefined;
-
-            const { fullDocument } = change.value;
-            expect(fullDocument.city).to.equal(docs[1].city);
-          } catch (error) {
-            expect.fail('Async could not be used with raw iterator API');
+          // eslint-disable-next-line @typescript-eslint/no-unused-vars
+          for await (const change of changeStream) {
+            expect.fail('Change stream resumed from partial iteration');
           }
-        }
-      );
 
-      it(
-        'ignores errors thrown from close',
-        { requires: { topology: '!single' } },
-        async function () {
-          changeStream = collection.watch([]);
-          await initIteratorMode(changeStream);
-          const changeStreamIterator = changeStream[Symbol.asyncIterator]();
-
-          sinon.stub(changeStream.cursor, 'close').throws(new MongoAPIError('testing'));
-
-          try {
-            await changeStreamIterator.return();
-          } catch (error) {
-            expect.fail('Async iterator threw an error on close');
-          }
+          expect(docs).to.have.length(2, 'expected to find remaining docs after partial iteration');
         }
       );
     });
@@ -2352,7 +2260,6 @@ describe('ChangeStream resumability', function () {
           async function () {
             changeStream = collection.watch([]);
             await initIteratorMode(changeStream);
-            const changeStreamIterator = changeStream[Symbol.asyncIterator]();
 
             await client.db('admin').command({
               configureFailPoint: is4_2Server(this.configuration.version)
@@ -2366,10 +2273,18 @@ describe('ChangeStream resumability', function () {
               }
             } as FailPoint);
 
-            await collection.insertOne({ city: 'New York City' });
-            await changeStreamIterator.next();
+            const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
+            await collection.insertMany(docs);
 
-            expect(aggregateEvents).to.have.lengthOf(2);
+            for await (const change of changeStream) {
+              const { fullDocument } = change;
+              const expectedDoc = docs.shift();
+              expect(fullDocument.city).to.equal(expectedDoc.city);
+              if (docs.length === 0) {
+                break;
+              }
+            }
+            expect(docs).to.have.length(0, 'expected to find all docs before exiting loop');
           }
         );
       }
@@ -2383,12 +2298,15 @@ describe('ChangeStream resumability', function () {
             await initIteratorMode(changeStream);
             const changeStreamIterator = changeStream[Symbol.asyncIterator]();
 
+            const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
+            await collection.insertMany(docs);
+
             // on 3.6 servers, no postBatchResumeToken is sent back in the initial aggregate response.
             // This means that a resume token isn't cached until the first change has been iterated.
             // In order to test the resume, we need to ensure that at least one document has
             // been iterated so we have a resume token to resume on.
-            await collection.insertOne({ city: 'New York City' });
             await changeStreamIterator.next();
+            docs.shift();
 
             const mock = sinon
               .stub(changeStream.cursor, '_getMore')
@@ -2399,10 +2317,16 @@ describe('ChangeStream resumability', function () {
                 callback(error);
               });
 
-            await collection.insertOne({ city: 'New York City' });
-            await changeStreamIterator.next();
+            for await (const change of changeStream) {
+              const { fullDocument } = change;
+              const expectedDoc = docs.shift();
+              expect(fullDocument.city).to.equal(expectedDoc.city);
+              if (docs.length === 0) {
+                break;
+              }
+            }
 
-            expect(aggregateEvents).to.have.lengthOf(2);
+            expect(docs).to.have.length(0, 'expected to find all docs before exiting loop');
           }
         );
       }
@@ -2447,25 +2371,34 @@ describe('ChangeStream resumability', function () {
           async function () {
             changeStream = collection.watch([]);
             await initIteratorMode(changeStream);
-            const changeStreamIterator = changeStream[Symbol.asyncIterator]();
 
-            const unresumableErrorCode = 1000;
-            await client.db('admin').command({
-              configureFailPoint: is4_2Server(this.configuration.version)
-                ? 'failCommand'
-                : 'failGetMoreAfterCursorCheckout',
-              mode: { times: 1 },
-              data: {
-                failCommands: ['getMore'],
-                errorCode: unresumableErrorCode
-              }
-            } as FailPoint);
+            const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
+            await collection.insertMany(docs);
 
-            await collection.insertOne({ city: 'New York City' });
+            try {
+              for await (const change of changeStream) {
+                const { fullDocument } = change;
+                const expectedDoc = docs.shift();
+                expect(fullDocument.city).to.equal(expectedDoc.city);
+
+                const unresumableErrorCode = 1000;
+                await client.db('admin').command({
+                  configureFailPoint: is4_2Server(this.configuration.version)
+                    ? 'failCommand'
+                    : 'failGetMoreAfterCursorCheckout',
+                  mode: { times: 1 },
+                  data: {
+                    failCommands: ['getMore'],
+                    errorCode: unresumableErrorCode
+                  }
+                } as FailPoint);
+              }
 
-            const error = await changeStreamIterator.next().catch(e => e);
-            expect(error).to.be.instanceOf(MongoServerError);
-            expect(aggregateEvents).to.have.lengthOf(1);
+              expect.fail('Async did not throw on an unresumable error');
+            } catch (error) {
+              expect(error).to.be.instanceOf(MongoServerError);
+              expect(aggregateEvents).to.have.lengthOf(1);
+            }
           }
         );
       });