test(ChangeStream): update changeStream spec tests / test runner

- update changeStream spec tests
- update spec-test runner to comply with tests
- Adds a default value for expectations array in case there are no expectations
- Parses expected values through Extended JSON to ensure that they are in canonical format
- fix result assertion in change stream spec test
- get rid of EJSONToJSON
- add fail point to spec tests. 
- Skips a test that will not pass until later features are added
- fix flakey test in Node 10.16
- fix race condition in test runner

Fixes NODE-2017

Author: daprahamian

test/functional/change_stream_spec_tests.js

@@ -9,12 +9,12 @@ const delay = require('./shared').delay;
 const expect = chai.expect;
 
 describe('Change Stream Spec', function() {
-  const EJSONToJSON = x => JSON.parse(EJSON.stringify(x));
-
   let globalClient;
   let ctx;
   let events;
 
+  const TESTS_TO_SKIP = new Set(['Test consecutive resume']);
+
   before(function() {
     const configuration = this.configuration;
     return setupDatabase(configuration).then(() => {
@@ -34,7 +34,7 @@ describe('Change Stream Spec', function() {
     .filter(filename => filename.match(/\.json$/))
     .forEach(filename => {
       const specString = fs.readFileSync(`${__dirname}/spec/change-stream/${filename}`, 'utf8');
-      const specData = JSON.parse(specString);
+      const specData = EJSON.parse(specString, { relaxed: true });
 
       const ALL_DBS = [specData.database_name, specData.database2_name];
 
@@ -64,11 +64,14 @@ describe('Change Stream Spec', function() {
           ctx = undefined;
           events = undefined;
 
+          client.removeAllListeners('commandStarted');
+
           return client && client.close();
         });
 
         specData.tests.forEach(test => {
-          const itFn = test.skip ? it.skip : test.only ? it.only : it;
+          const shouldSkip = test.skip || TESTS_TO_SKIP.has(test.description);
+          const itFn = shouldSkip ? it.skip : test.only ? it.only : it;
           const metadata = generateMetadata(test);
           const testFn = generateTestFn(test);
 
@@ -94,18 +97,30 @@ describe('Change Stream Spec', function() {
   }
 
   function generateTestFn(test) {
+    const configureFailPoint = makeFailPointCommand(test);
     const testFnRunOperations = makeTestFnRunOperations(test);
     const testSuccess = makeTestSuccess(test);
     const testFailure = makeTestFailure(test);
     const testAPM = makeTestAPM(test);
 
     return function testFn() {
-      return testFnRunOperations(ctx)
+      return configureFailPoint(ctx)
+        .then(() => testFnRunOperations(ctx))
         .then(testSuccess, testFailure)
         .then(() => testAPM(ctx, events));
     };
   }
 
+  function makeFailPointCommand(test) {
+    if (!test.failPoint) {
+      return () => Promise.resolve();
+    }
+
+    return function(ctx) {
+      return ctx.gc.db('admin').command(test.failPoint);
+    };
+  }
+
   function makeTestSuccess(test) {
     const result = test.result;
 
@@ -115,7 +130,7 @@ describe('Change Stream Spec', function() {
       }
 
       if (result.success) {
-        value = EJSONToJSON(value);
+        expect(value).to.have.a.lengthOf(result.success.length);
         assertEquality(value, result.success);
       }
     };
@@ -134,7 +149,7 @@ describe('Change Stream Spec', function() {
   }
 
   function makeTestAPM(test) {
-    const expectedEvents = test.expectations;
+    const expectedEvents = test.expectations || [];
 
     return function testAPM(ctx, events) {
       expectedEvents
@@ -146,12 +161,21 @@ describe('Change Stream Spec', function() {
               `Expected there to be an APM event at index ${idx}, but there was none`
             );
           }
-          const actual = EJSONToJSON(events[idx]);
-          assertEquality(actual, expected);
+          assertEquality(events[idx], expected);
         });
     };
   }
 
+  function allSettled(promises) {
+    let err;
+    return Promise.all(promises.map(p => p.catch(x => (err = err || x)))).then(args => {
+      if (err) {
+        throw err;
+      }
+      return args;
+    });
+  }
+
   function makeTestFnRunOperations(test) {
     const target = test.target;
     const operations = test.operations;
@@ -165,7 +189,7 @@ describe('Change Stream Spec', function() {
       const changeStreamPromise = readAndCloseChangeStream(ctx.changeStream, success.length);
       const operationsPromise = runOperations(ctx.gc, operations);
 
-      return Promise.all([changeStreamPromise, operationsPromise]).then(args => args[0]);
+      return allSettled([changeStreamPromise, operationsPromise]).then(args => args[0]);
     };
   }
 

test/functional/change_stream_tests.js

@@ -1272,7 +1272,7 @@ describe('Change Streams', function() {
     // The actual test we wish to run
     test: function(done) {
       var configuration = this.configuration;
-      var fs = require('fs');
+      const stream = require('stream');
       const client = configuration.newClient();
 
       client.connect(function(err, client) {
@@ -1282,30 +1282,32 @@ describe('Change Streams', function() {
         var theCollection = theDatabase.collection('pipeTest');
         var thisChangeStream = theCollection.watch(pipeline);
 
-        var filename = '/tmp/_nodemongodbnative_stream_out.txt';
-        var outStream = fs.createWriteStream(filename);
+        var outStream = new stream.PassThrough({ objectMode: true });
 
         // Make a stream transforming to JSON and piping to the file
         thisChangeStream.stream({ transform: JSON.stringify }).pipe(outStream);
 
+        function close(_err) {
+          thisChangeStream.close(err => client.close(cErr => done(_err || err || cErr)));
+        }
+
+        outStream
+          .on('data', data => {
+            try {
+              var parsedEvent = JSON.parse(data);
+              assert.equal(parsedEvent.fullDocument.a, 1);
+              close();
+            } catch (e) {
+              close(e);
+            }
+          })
+          .on('error', close);
+
         setTimeout(() => {
           theCollection.insert({ a: 1 }, function(err) {
             assert.ifError(err);
           });
         });
-
-        // Listen for changes to the file
-        var watcher = fs.watch(filename, function(eventType) {
-          assert.equal(eventType, 'change');
-
-          var fileContents = fs.readFileSync(filename, 'utf8');
-          var parsedFileContents = JSON.parse(fileContents);
-          assert.equal(parsedFileContents.fullDocument.a, 1);
-
-          watcher.close();
-
-          thisChangeStream.close(err => client.close(cErr => done(err || cErr)));
-        });
       });
     }
   });

test/functional/spec/change-stream/README.rst

@@ -33,7 +33,7 @@ Each YAML file has the following keys:
   - ``description``: The name of the test.
   - ``minServerVersion``: The minimum server version to run this test against. If not present, assume there is no minimum server version.
   - ``maxServerVersion``: Reserved for later use
-  - ``failPoint``: Reserved for later use
+  - ``failPoint``(optional): The configureFailPoint command document to run to configure a fail point on the primary server.
   - ``target``: The entity on which to run the change stream. Valid values are:
 
     - ``collection``: Watch changes on collection ``database_name.collection_name``
@@ -44,10 +44,11 @@ Each YAML file has the following keys:
   - ``changeStreamPipeline``: An array of additional aggregation pipeline stages to add to the change stream
   - ``changeStreamOptions``: Additional options to add to the changeStream
   - ``operations``: Array of documents, each describing an operation. Each document has the following fields:
+
     - ``database``: Database against which to run the operation
     - ``collection``: Collection against which to run the operation
-    - ``commandName``: Name of the command to run
-    - ``arguments``: Object of arguments for the command (ex: document to insert)
+    - ``name``: Name of the command to run
+    - ``arguments`` (optional): Object of arguments for the command (ex: document to insert)
 
   - ``expectations``: Optional list of command-started events in Extended JSON format
   - ``result``: Document with ONE of the following fields:
@@ -103,6 +104,9 @@ For each YAML file, for each element in ``tests``:
   - Drop the database ``database2_name``
   - Create the database ``database_name`` and the collection ``database_name.collection_name``
   - Create the database ``database2_name`` and the collection ``database2_name.collection2_name``
+  - If the the ``failPoint`` field is present, configure the fail point on the primary server. See
+    `Server Fail Point <../../transactions/tests#server-fail-point>`_ in the
+    Transactions spec test documentation for more information.
 
 - Create a new MongoClient ``client``
 - Begin monitoring all APM events for ``client``. (If the driver uses global listeners, filter out all events that do not originate with ``client``). Filter out any "internal" commands (e.g. ``isMaster``)
@@ -117,12 +121,12 @@ For each YAML file, for each element in ``tests``:
 - If there was an error:
 
   - Assert that an error was expected for the test.
-  - Assert that the error MATCHES ``results.error``
+  - Assert that the error MATCHES ``result.error``
 
 - Else:
 
   - Assert that no error was expected for the test
-  - Assert that the changes received from ``changeStream`` MATCH the results in ``results.success``
+  - Assert that the changes received from ``changeStream`` MATCH the results in ``result.success``
 
 - If there are any ``expectations``
 
@@ -144,12 +148,56 @@ Prose Tests
 
 The following tests have not yet been automated, but MUST still be tested
 
-1. ``ChangeStream`` must continuously track the last seen ``resumeToken``
-2. ``ChangeStream`` will throw an exception if the server response is missing the resume token
-3. ``ChangeStream`` will automatically resume one time on a resumable error (including `not master`) with the initial pipeline and options, except for the addition/update of a ``resumeToken``.
-4. ``ChangeStream`` will not attempt to resume on a server error
-5. ``ChangeStream`` will perform server selection before attempting to resume, using initial ``readPreference``
-6. Ensure that a cursor returned from an aggregate command with a cursor id and an initial empty batch is not closed on the driver side.
-7. The ``killCursors`` command sent during the "Resume Process" must not be allowed to throw an exception.
-8. ``$changeStream`` stage for ``ChangeStream`` against a server ``>=4.0`` that has not received any results yet MUST include a ``startAtOperationTime`` option when resuming a changestream.
-9. ``ChangeStream`` will resume after a ``killCursors`` command is issued for its child cursor.
+#. ``ChangeStream`` must continuously track the last seen ``resumeToken``
+#. ``ChangeStream`` will throw an exception if the server response is missing the resume token (if wire version is < 8, this is a driver-side error; for 8+, this is a server-side error)
+#. ``ChangeStream`` will automatically resume one time on a resumable error (including `not master`) with the initial pipeline and options, except for the addition/update of a ``resumeToken``.
+#. ``ChangeStream`` will not attempt to resume on any error encountered while executing an ``aggregate`` command.
+#. ``ChangeStream`` will not attempt to resume after encountering error code 11601 (Interrupted), 136 (CappedPositionLost), or 237 (CursorKilled) while executing a ``getMore`` command.
+#. ``ChangeStream`` will perform server selection before attempting to resume, using initial ``readPreference``
+#. Ensure that a cursor returned from an aggregate command with a cursor id and an initial empty batch is not closed on the driver side.
+#. The ``killCursors`` command sent during the "Resume Process" must not be allowed to throw an exception.
+#. ``$changeStream`` stage for ``ChangeStream`` against a server ``>=4.0`` and ``<4.0.7`` that has not received any results yet MUST include a ``startAtOperationTime`` option when resuming a changestream.
+#. ``ChangeStream`` will resume after a ``killCursors`` command is issued for its child cursor.
+#. - For a ``ChangeStream`` under these conditions:
+      - Running against a server ``>=4.0.7``.
+      - The batch is empty or has been iterated to the last document.
+   - Expected result: 
+       - ``getResumeToken`` must return the ``postBatchResumeToken`` from the current command response.
+#. - For a ``ChangeStream`` under these conditions:
+      - Running against a server ``<4.0.7``.
+      - The batch is empty or has been iterated to the last document.
+   - Expected result: 
+      - ``getResumeToken`` must return the ``_id`` of the last document returned if one exists.
+      - ``getResumeToken`` must return ``startAfter`` from the initial aggregate if the option was specified.
+      - ``getResumeToken`` must return ``resumeAfter`` from the initial aggregate if the option was specified.
+      - If neither the ``startAfter`` nor ``resumeAfter`` options were specified, the ``getResumeToken`` result must be empty.
+#. - For a ``ChangeStream`` under these conditions:
+      - The batch is not empty.
+      - The batch has been iterated up to but not including the last element.
+   - Expected result:
+      - ``getResumeToken`` must return the ``_id`` of the previous document returned.
+#. - For a ``ChangeStream`` under these conditions:
+      - The batch is not empty.
+      - The batch hasn’t been iterated at all.
+      - Only the initial ``aggregate`` command has been executed.
+   - Expected result:
+      - ``getResumeToken`` must return ``startAfter`` from the initial aggregate if the option was specified.
+      - ``getResumeToken`` must return ``resumeAfter`` from the initial aggregate if the option was specified.
+      - If neither the ``startAfter`` nor ``resumeAfter`` options were specified, the ``getResumeToken`` result must be empty.
+#. - For a ``ChangeStream`` under these conditions:
+      - Running against a server ``>=4.0.7``.
+      - The batch is not empty.
+      - The batch hasn’t been iterated at all.
+      - The stream has iterated beyond a previous batch and a ``getMore`` command has just been executed.
+   - Expected result:
+      - ``getResumeToken`` must return the ``postBatchResumeToken`` from the previous command response.
+#. - For a ``ChangeStream`` under these conditions:
+      - Running against a server ``<4.0.7``.
+      - The batch is not empty.
+      - The batch hasn’t been iterated at all.
+      - The stream has iterated beyond a previous batch and a ``getMore`` command has just been executed.
+   - Expected result:
+      - ``getResumeToken`` must return the ``_id`` of the previous document returned if one exists.
+      - ``getResumeToken`` must return ``startAfter`` from the initial aggregate if the option was specified.
+      - ``getResumeToken`` must return ``resumeAfter`` from the initial aggregate if the option was specified.
+      - If neither the ``startAfter`` nor ``resumeAfter`` options were specified, the ``getResumeToken`` result must be empty.

test/functional/spec/change-stream/change-streams-errors.json

@@ -73,6 +73,43 @@
           "code": 40324
         }
       }
+    },
+    {
+      "description": "Change Stream should error when _id is projected out",
+      "minServerVersion": "4.1.11",
+      "target": "collection",
+      "topology": [
+        "replicaset",
+        "sharded"
+      ],
+      "changeStreamPipeline": [
+        {
+          "$project": {
+            "_id": 0
+          }
+        }
+      ],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "z": 3
+            }
+          }
+        }
+      ],
+      "result": {
+        "error": {
+          "code": 280,
+          "errorLabels": [
+            "NonResumableChangeStreamError"
+          ]
+        }
+      }
     }
   ]
 }

test/functional/spec/change-stream/change-streams-errors.yml

@@ -50,4 +50,27 @@ tests:
           database_name: *database_name
     result:
       error:
-        code: 40324
+        code: 40324
+  -
+    description: Change Stream should error when _id is projected out
+    minServerVersion: "4.1.11"
+    target: collection
+    topology:
+      - replicaset
+      - sharded
+    changeStreamPipeline:
+      -
+        $project: { _id: 0 }
+    changeStreamOptions: {}
+    operations:
+      -
+        database: *database_name
+        collection: *collection_name
+        name: insertOne
+        arguments:
+          document:
+            z: 3
+    result:
+      error:
+        code: 280
+        errorLabels: [ "NonResumableChangeStreamError" ]