feat(NODE-7441): add ChangeStream.bufferedCount tests

Author: typesafe

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

@@ -1165,6 +1165,82 @@ describe('Change Streams', function () {
         );
       });
     });
+
+    describe('#bufferedCount()', function () {
+      it(
+        'should return 0 before starting to consume the change stream (empty collection)',
+        { requires: { topology: 'replicaset' } },
+        async function () {
+          changeStream = collection.watch([]);
+          expect(changeStream.bufferedCount()).to.equal(0);
+          await changeStream.close();
+        }
+      );
+
+      it(
+        'should return 0 before starting to consume the change stream (non-empty collection)',
+        { requires: { topology: 'replicaset' } },
+        async function () {
+          // existing documents
+          await collection.insertOne({ a: 1 });
+          await collection.insertOne({ a: 2 });
+
+          changeStream = collection.watch([]);
+
+          // docs inserted after the change stream was created
+          await collection.insertOne({ a: 3 });
+          await collection.insertOne({ a: 4 });
+
+          expect(changeStream.bufferedCount()).to.equal(0);
+          await changeStream.close();
+        }
+      );
+
+      it('should return the underlying cursor buffered document count', {
+        metadata: { requires: { topology: 'replicaset' } },
+        async test() {
+          changeStream = collection.watch([]);
+          await initIteratorMode(changeStream);
+          await collection.insertOne({ a: 1 });
+          await collection.insertOne({ a: 2 });
+
+          expect(changeStream.bufferedCount()).to.equal(0);
+
+          // hasNext will trigger a batch fetch, buffering the documents
+          const hasNext = await changeStream.hasNext();
+          expect(hasNext).to.be.true;
+
+          expect(changeStream.bufferedCount()).to.equal(2);
+        }
+      });
+
+      it(`decreases as buffered documents are consumed`, {
+        metadata: { requires: { topology: 'replicaset' } },
+        async test() {
+          changeStream = collection.watch([]);
+          await initIteratorMode(changeStream);
+          await collection.insertOne({ a: 1 });
+          await collection.insertOne({ a: 2 });
+
+          expect(changeStream.bufferedCount()).to.equal(0);
+          // `next` triggers a batch fetch, buffering the documents
+          // and then consumes one document from that buffer
+          await changeStream.next();
+          expect(changeStream.bufferedCount()).to.equal(1);
+          await changeStream.next();
+          expect(changeStream.bufferedCount()).to.equal(0);
+
+          await collection.insertOne({ a: 1 });
+          await collection.insertOne({ a: 2 });
+
+          // `tryNext` also triggers a batch fetch
+          await changeStream.tryNext();
+          expect(changeStream.bufferedCount()).to.equal(1);
+          await changeStream.tryNext();
+          expect(changeStream.bufferedCount()).to.equal(0);
+        }
+      });
+    });
   });
 
   describe('startAfter', function () {