PHPLIB-366: Change streams test runner

Support placeholders in DocumentsMatchConstraint and allow CommandExpectations to ignore extra events.

Also removes abstract from base spec test class.

Spec tests synced with mongodb/specifications@a1c9d3c5a7b938f595bb2547c07a238da21263f3

tests/SpecTests/ChangeStreamsSpecTest.php

+<?php
+
+namespace MongoDB\Tests\SpecTests;
+
+use MongoDB\ChangeStream;
+use MongoDB\Client;
+use MongoDB\Driver\Exception\Exception;
+use MongoDB\Model\BSONDocument;
+use ArrayIterator;
+use LogicException;
+use MultipleIterator;
+use stdClass;
+
+/**
+ * Change Streams spec tests.
+ *
+ * @see https://github.com/mongodb/specifications/tree/master/source/change-streams
+ */
+class ChangeStreamsSpecTest extends FunctionalTestCase
+{
+    /* These should all pass before the driver can be considered compatible with
+     * MongoDB 4.2. */
+    private static $incompleteTests = [
+        'change-streams: Test consecutive resume' => 'PHPLIB-442, PHPLIB-416',
+    ];
+
+    /**
+     * Assert that the expected and actual command documents match.
+     *
+     * Note: this method may modify the $expected object.
+     *
+     * @param stdClass $expected Expected command document
+     * @param stdClass $actual   Actual command document
+     */
+    public static function assertCommandMatches(stdClass $expected, stdClass $actual)
+    {
+        static::assertDocumentsMatch($expected, $actual);
+    }
+
+    /**
+     * Assert that the expected and actual documents match.
+     *
+     * @param array $expectedDocuments Expected documents
+     * @param array $actualDocuments   Actual documents
+     */
+    public static function assertResult(array $expectedDocuments, array $actualDocuments)
+    {
+        static::assertCount(count($expectedDocuments), $actualDocuments);
+
+        $mi = new MultipleIterator(MultipleIterator::MIT_NEED_ANY);
+        $mi->attachIterator(new ArrayIterator($expectedDocuments));
+        $mi->attachIterator(new ArrayIterator($actualDocuments));
+
+        foreach ($mi as $documents) {
+            list($expectedDocument, $actualDocument) = $documents;
+
+            $constraint = new DocumentsMatchConstraint($expectedDocument, true, true, ['42']);
+
+            static::assertThat($actualDocument, $constraint);
+        }
+    }
+
+    /**
+     * Execute an individual test case from the specification.
+     *
+     * @dataProvider provideTests
+     * @param string    $name            Test name
+     * @param stdClass  $test            Individual "tests[]" document
+     * @param string    $databaseName    Name of database under test
+     * @param string    $collectionName  Name of collection under test
+     * @param string    $database2Name   Name of alternate database under test
+     * @param string    $collection2Name Name of alternate collection under test
+     */
+    public function testChangeStreams($name, stdClass $test, $databaseName = null, $collectionName = null, $database2Name = null, $collection2Name = null)
+    {
+        $this->setName($name);
+
+        if (isset(self::$incompleteTests[$name])) {
+            $this->markTestIncomplete(self::$incompleteTests[$name]);
+        }
+
+        $this->checkServerRequirements($this->createRunOn($test));
+
+        if (!isset($databaseName, $collectionName, $database2Name, $collection2Name)) {
+            $this->fail('Required database and collection names are unset');
+        }
+
+        $context = Context::fromChangeStreams($test, $databaseName, $collectionName);
+        $this->setContext($context);
+
+        $this->dropDatabasesAndCreateCollection($databaseName, $collectionName);
+        $this->dropDatabasesAndCreateCollection($database2Name, $collection2Name);
+
+        if (isset($test->failPoint)) {
+            $this->configureFailPoint($test->failPoint);
+        }
+
+        if (isset($test->expectations)) {
+            $commandExpectations = CommandExpectations::fromChangeStreams($test->expectations);
+            $commandExpectations->startMonitoring();
+        }
+
+        $errorExpectation = ErrorExpectation::fromChangeStreams($test->result);
+        $resultExpectation = ResultExpectation::fromChangeStreams($test->result, [$this, 'assertResult']);
+
+        $result = null;
+        $exception = null;
+
+        try {
+            $changeStream = $this->createChangeStream($test);
+        } catch (Exception $e) {
+            $exception = $e;
+        }
+
+        if (isset($commandExpectations)) {
+            $commandExpectations->stopMonitoring();
+        }
+
+        foreach ($test->operations as $operation) {
+            Operation::fromChangeStreams($operation)->assert($this, $context);
+        }
+
+        if (isset($commandExpectations)) {
+            $commandExpectations->startMonitoring();
+        }
+
+        /* If the change stream was successfully created (i.e. $exception is
+         * null), attempt to iterate up to the expected number of results. It's
+         * possible that some errors (e.g. projecting out _id) will only be
+         * thrown during iteration, so we must also try/catch here. */
+        try {
+            if (isset($changeStream)) {
+                $limit = isset($test->result->success) ? count($test->result->success) : 0;
+                $result = $this->iterateChangeStream($changeStream, $limit);
+            }
+        } catch (Exception $e) {
+            $this->assertNull($exception);
+            $exception = $e;
+        }
+
+        $errorExpectation->assert($this, $exception);
+        $resultExpectation->assert($this, $result);
+
+        if (isset($commandExpectations)) {
+            $commandExpectations->stopMonitoring();
+            $commandExpectations->assert($this, $context);
+        }
+    }
+
+    public function provideTests()
+    {
+        $testArgs = [];
+
+        foreach (glob(__DIR__ . '/change-streams/*.json') as $filename) {
+            $json = $this->decodeJson(file_get_contents($filename));
+            $group = basename($filename, '.json');
+            $databaseName = isset($json->database_name) ? $json->database_name : null;
+            $database2Name = isset($json->database2_name) ? $json->database2_name : null;
+            $collectionName = isset($json->collection_name) ? $json->collection_name : null;
+            $collection2Name = isset($json->collection2_name) ? $json->collection2_name : null;
+
+            foreach ($json->tests as $test) {
+                $name = $group . ': ' . $test->description;
+                $testArgs[] = [$name, $test, $databaseName, $collectionName, $database2Name, $collection2Name];
+            }
+        }
+
+        return $testArgs;
+    }
+
+    /**
+     * Create a change stream.
+     *
+     * @param stdClass $test
+     * @return ChangeStream
+     * @throws LogicException if the target is unsupported
+     */
+    private function createChangeStream(stdClass $test)
+    {
+        $context = $this->getContext();
+        $pipeline = isset($test->changeStreamPipeline) ? $test->changeStreamPipeline : [];
+        $options = isset($test->changeStreamOptions) ? (array) $test->changeStreamOptions : [];
+
+        switch ($test->target) {
+            case 'client':
+                return $context->client->watch($pipeline, $options);
+
+            case 'database':
+                return $context->getDatabase()->watch($pipeline, $options);
+
+            case 'collection':
+                return $context->getCollection()->watch($pipeline, $options);
+
+            default:
+                throw new LogicException('Unsupported target: ' . $test->target);
+        }
+    }
+
+    /**
+     * Convert the server requirements to a standard "runOn" array used by other
+     * specifications.
+     *
+     * @param stdClass $test
+     * @return array
+     */
+    private function createRunOn(stdClass $test)
+    {
+        $req = new stdClass;
+
+        /* Append ".99" as patch version, since command monitoring tests expect
+         * the minor version to be an inclusive upper bound. */
+        if (isset($test->maxServerVersion)) {
+            $req->maxServerVersion = $test->maxServerVersion;
+        }
+
+        if (isset($test->minServerVersion)) {
+            $req->minServerVersion = $test->minServerVersion;
+        }
+
+        if (isset($test->topology)) {
+            $req->topology = $test->topology;
+        }
+
+        return [$req];
+    }
+
+    /**
+     * Drop the database and create the collection.
+     *
+     * @param string $databaseName
+     * @param string $collectionName
+     */
+    private function dropDatabasesAndCreateCollection($databaseName, $collectionName)
+    {
+        $context = $this->getContext();
+
+        $database = $context->client->selectDatabase($databaseName);
+        $database->drop($context->defaultWriteOptions);
+        $database->createCollection($collectionName, $context->defaultWriteOptions);
+    }
+
+    /**
+     * Iterate a change stream.
+     *
+     * @param ChangeStream $changeStream
+     * @return BSONDocument[]
+     */
+    private function iterateChangeStream(ChangeStream $changeStream, $limit = 0)
+    {
+        $events = [];
+
+        for ($changeStream->rewind(); count($events) < $limit; $changeStream->next()) {
+            if ( ! $changeStream->valid()) {
+                continue;
+            }
+
+            $event = $changeStream->current();
+
+            $this->assertInstanceOf(BSONDocument::class, $event);
+
+            $events[] = $event;
+        }
+
+        return $events;
+    }
+}

tests/SpecTests/CommandExpectations.php

@@ -21,6 +21,7 @@ class CommandExpectations implements CommandSubscriber
     private $ignoreCommandFailed = false;
     private $ignoreCommandStarted = false;
     private $ignoreCommandSucceeded = false;
+    private $ignoreExtraEvents = false;
 
     private function __construct(array $events)
     {
@@ -44,6 +45,20 @@ class CommandExpectations implements CommandSubscriber
         }
     }
 
+    public static function fromChangeStreams(array $expectedEvents)
+    {
+        $o = new self($expectedEvents);
+
+        $o->ignoreCommandFailed = true;
+        $o->ignoreCommandSucceeded = true;
+        /* Change Streams spec tests do not include getMore commands in the
+         * list of expected events, so ignore any observed events beyond the
+         * number that are expected. */
+        $o->ignoreExtraEvents = true;;
+
+        return $o;
+    }
+
     public static function fromCommandMonitoring(array $expectedEvents)
     {
         return new self($expectedEvents);
@@ -125,11 +140,15 @@ class CommandExpectations implements CommandSubscriber
      */
     public function assert(FunctionalTestCase $test, Context $context)
     {
-        $test->assertCount(count($this->expectedEvents), $this->actualEvents);
+        $actualEvents = $this->ignoreExtraEvents
+            ? array_slice($this->actualEvents, 0, count($this->expectedEvents))
+            : $this->actualEvents;
+
+        $test->assertCount(count($this->expectedEvents), $actualEvents);
 
         $mi = new MultipleIterator(MultipleIterator::MIT_NEED_ANY);
         $mi->attachIterator(new ArrayIterator($this->expectedEvents));
-        $mi->attachIterator(new ArrayIterator($this->actualEvents));
+        $mi->attachIterator(new ArrayIterator($actualEvents));
 
         foreach ($mi as $events) {
             list($expectedEventAndClass, $actualEvent) = $events;

tests/SpecTests/Context.php

@@ -35,6 +35,15 @@ final class Context
         $this->outcomeCollectionName = $collectionName;
     }
 
+    public static function fromChangeStreams(stdClass $test, $databaseName, $collectionName)
+    {
+        $o = new self($databaseName, $collectionName);
+
+        $o->client = new Client(FunctionalTestCase::getUri());
+
+        return $o;
+    }
+
     public static function fromCommandMonitoring(stdClass $test, $databaseName, $collectionName)
     {
         $o = new self($databaseName, $collectionName);
@@ -98,7 +107,7 @@ final class Context
 
     public function getCollection(array $collectionOptions = [])
     {
-        return $this->client->selectCollection(
+        return $this->selectCollection(
             $this->databaseName,
             $this->collectionName,
             $this->prepareOptions($collectionOptions)
@@ -107,10 +116,7 @@ final class Context
 
     public function getDatabase(array $databaseOptions = [])
     {
-        return $this->client->selectDatabase(
-            $this->databaseName,
-            $this->prepareOptions($databaseOptions)
-        );
+        return $this->selectDatabase($this->databaseName, $databaseOptions);
     }
 
     /**
@@ -221,6 +227,23 @@ final class Context
         }
     }
 
+    public function selectCollection($databaseName, $collectionName, array $collectionOptions = [])
+    {
+        return $this->client->selectCollection(
+            $databaseName,
+            $collectionName,
+            $this->prepareOptions($collectionOptions)
+        );
+    }
+
+    public function selectDatabase($databaseName, array $databaseOptions = [])
+    {
+        return $this->client->selectDatabase(
+            $databaseName,
+            $this->prepareOptions($databaseOptions)
+        );
+    }
+
     private function prepareSessionOptions(array $options)
     {
         if (isset($options['defaultTransactionOptions'])) {

tests/SpecTests/DocumentsMatchConstraint.php

@@ -19,6 +19,7 @@ class DocumentsMatchConstraint extends Constraint
 {
     private $ignoreExtraKeysInRoot = false;
     private $ignoreExtraKeysInEmbedded = false;
+    private $placeholders = [];
     /* TODO: This is not currently used, but was preserved from the design of
      * TestCase::assertMatchesDocument(), which would sort keys and then compare
      * documents as JSON strings. If the TODO item in matches() is implemented
@@ -33,14 +34,15 @@ class DocumentsMatchConstraint extends Constraint
      * @param array|object $value
      * @param boolean      $ignoreExtraKeysInRoot     If true, ignore extra keys within the root document
      * @param boolean      $ignoreExtraKeysInEmbedded If true, ignore extra keys within embedded documents
-     * 
+     * @param array        $placeholders              Placeholders for any value
      */
-    public function __construct($value, $ignoreExtraKeysInRoot = false, $ignoreExtraKeysInEmbedded = false)
+    public function __construct($value, $ignoreExtraKeysInRoot = false, $ignoreExtraKeysInEmbedded = false, array $placeholders = [])
     {
         parent::__construct();
         $this->value = $this->prepareBSON($value, true, $this->sortKeys);
         $this->ignoreExtraKeysInRoot = $ignoreExtraKeysInRoot;
         $this->ignoreExtraKeysInEmbedded = $ignoreExtraKeysInEmbedded;
+        $this->placeholders = $placeholders;
     }
 
     /**
@@ -101,6 +103,10 @@ class DocumentsMatchConstraint extends Constraint
                 throw new RuntimeException('$actual is missing key: ' . $key);
             }
 
+            if (in_array($expectedValue, $this->placeholders, true)) {
+                continue;
+            }
+
             $actualValue = $actual[$key];
 
             if (($expectedValue instanceof BSONArray && $actualValue instanceof BSONArray) ||

tests/SpecTests/DocumentsMatchConstraintTest.php

@@ -46,6 +46,15 @@ class DocumentsMatchConstraintTest extends TestCase
         $this->assertResult(false, $c, [1, ['a' => 2]], 'Keys must have the correct value');
     }
 
+    public function testPlaceholders()
+    {
+        $c = new DocumentsMatchConstraint(['x' => '42', 'y' => 42, 'z' => ['a' => 24]], false, false, [24, 42]);
+
+        $this->assertResult(true, $c, ['x' => '42', 'y' => 'foo', 'z' => ['a' => 1]], 'Placeholders accept any value');
+        $this->assertResult(false, $c, ['x' => 42, 'y' => 'foo', 'z' => ['a' => 1]], 'Placeholder type must match');
+        $this->assertResult(true, $c, ['x' => '42', 'y' => 42, 'z' => ['a' => 24]], 'Exact match');
+    }
+
     private function assertResult($expectedResult, DocumentsMatchConstraint $constraint, $value, $message)
     {
         $this->assertSame($expectedResult, $constraint->evaluate($value, '', true), $message);

tests/SpecTests/ErrorExpectation.php

@@ -25,6 +25,7 @@ final class ErrorExpectation
         'OperationNotSupportedInTransaction' => 263,
     ];
 
+    private $code;
     private $codeName;
     private $isExpected = false;
     private $excludedLabels = [];
@@ -35,6 +36,26 @@ final class ErrorExpectation
     {
     }
 
+    public static function fromChangeStreams(stdClass $result)
+    {
+        $o = new self;
+
+        if (isset($result->error->code)) {
+            $o->code = $result->error->code;
+            $o->isExpected = true;
+        }
+
+        if (isset($result->error->errorLabels)) {
+            if (!self::isArrayOfStrings($result->error->errorLabels)) {
+                throw InvalidArgumentException::invalidType('errorLabels', $result->error->errorLabels, 'string[]');
+            }
+            $o->includedLabels = $result->error->errorLabels;
+            $o->isExpected = true;
+        }
+
+        return $o;
+    }
+
     public static function fromRetryableWrites(stdClass $outcome)
     {
         $o = new self;
@@ -88,6 +109,11 @@ final class ErrorExpectation
         return $o;
     }
 
+    public static function noError()
+    {
+        return new self();
+    }
+
     /**
      * Assert that the error expectation matches the actual outcome.
      *

tests/SpecTests/FunctionalTestCase.php

@@ -26,7 +26,7 @@ use UnexpectedValueException;
  *
  * @see https://github.com/mongodb/specifications
  */
-abstract class FunctionalTestCase extends BaseFunctionalTestCase
+class FunctionalTestCase extends BaseFunctionalTestCase
 {
     const TOPOLOGY_SINGLE = 'single';
     const TOPOLOGY_REPLICASET = 'replicaset';
@@ -60,7 +60,10 @@ abstract class FunctionalTestCase extends BaseFunctionalTestCase
      * @param stdClass $expectedCommand Expected command document
      * @param stdClass $actualCommand   Actual command document
      */
-    abstract public static function assertCommandMatches(stdClass $expected, stdClass $actual);
+    public static function assertCommandMatches(stdClass $expected, stdClass $actual)
+    {
+        throw new LogicException(sprintf('%s does not assert CommandStartedEvents', get_called_class()));
+    }
 
     /**
      * Assert that the expected and actual command reply documents match.
@@ -71,7 +74,10 @@ abstract class FunctionalTestCase extends BaseFunctionalTestCase
      * @param stdClass $expected Expected command reply document
      * @param stdClass $actual   Actual command reply document
      */
-    abstract public static function assertCommandReplyMatches(stdClass $expected, stdClass $actual);
+    public static function assertCommandReplyMatches(stdClass $expected, stdClass $actual)
+    {
+        throw new LogicException(sprintf('%s does not assert CommandSucceededEvents', get_called_class()));
+    }
 
     /**
      * Asserts that two given documents match.

tests/SpecTests/Operation.php

@@ -3,6 +3,7 @@
 namespace MongoDB\Tests\SpecTests;
 
 use MongoDB\Collection;
+use MongoDB\Database;
 use MongoDB\Driver\Cursor;
 use MongoDB\Driver\Session;
 use MongoDB\Driver\Exception\BulkWriteException;
@@ -19,6 +20,8 @@ final class Operation
 {
     const OBJECT_COLLECTION = 'collection';
     const OBJECT_DATABASE = 'database';
+    const OBJECT_SELECT_COLLECTION = 'selectCollection';
+    const OBJECT_SELECT_DATABASE = 'selectDatabase';
     const OBJECT_SESSION0 = 'session0';
     const OBJECT_SESSION1 = 'session1';
 
@@ -26,7 +29,9 @@ final class Operation
     public $resultExpectation;
 
     private $arguments = [];
+    private $collectionName;
     private $collectionOptions = [];
+    private $databaseName;
     private $databaseOptions = [];
     private $name;
     private $object = self::OBJECT_COLLECTION;
@@ -44,6 +49,35 @@ final class Operation
         }
     }
 
+    public static function fromChangeStreams(stdClass $operation)
+    {
+        $o = new self($operation);
+
+        // Expect all operations to succeed
+        $o->errorExpectation = ErrorExpectation::noError();
+
+        /* The Change Streams spec tests include a unique "rename" operation,
+         * which we should convert to a renameCollection command to be run
+         * against the admin database. */
+        if ($operation->name === 'rename') {
+            $o->object = self::OBJECT_SELECT_DATABASE;
+            $o->databaseName = 'admin';
+            $o->name = 'runCommand';
+            $o->arguments = ['command' => [
+                'renameCollection' => $operation->database . '.' . $operation->collection,
+                'to' => $operation->database . '.' . $operation->arguments->to,
+            ]];
+
+            return $o;
+        }
+
+        $o->databaseName = $operation->database;
+        $o->collectionName = $operation->collection;
+        $o->object = self::OBJECT_SELECT_COLLECTION;
+
+        return $o;
+    }
+
     public static function fromCommandMonitoring(stdClass $operation)
     {
         $o = new self($operation);
@@ -52,6 +86,9 @@ final class Operation
             $o->collectionOptions = (array) $operation->collectionOptions;
         }
 
+        /* We purposefully avoid setting a default error expectation, because
+         * some tests may trigger a write or command error. */
+
         return $o;
     }
 
@@ -134,16 +171,26 @@ final class Operation
     {
         switch ($this->object) {
             case self::OBJECT_COLLECTION:
-                return $this->executeForCollection($context);
+                $collection = $context->getCollection($this->collectionOptions);
+                return $this->executeForCollection($collection, $context);
 
             case self::OBJECT_DATABASE:
-                return $this->executeForDatabase($context);
+                $database = $context->getDatabase($this->databaseOptions);
+                return $this->executeForDatabase($database, $context);
+
+            case self::OBJECT_SELECT_COLLECTION:
+                $collection = $context->selectCollection($this->databaseName, $this->collectionName, $this->collectionOptions);
+                return $this->executeForCollection($collection, $context);
+
+            case self::OBJECT_SELECT_DATABASE:
+                $database = $context->selectDatabase($this->databaseName, $this->databaseOptions);
+                return $this->executeForDatabase($database, $context);
 
             case self::OBJECT_SESSION0:
-                return $this->executeForSession($context);
+                return $this->executeForSession($context->session0, $context);
 
             case self::OBJECT_SESSION1:
-                return $this->executeForSession($context);
+                return $this->executeForSession($context->session1, $context);
 
             default:
                 throw new LogicException('Unsupported object: ' . $this->object);
@@ -153,13 +200,13 @@ final class Operation
     /**
      * Executes the collection operation and return its result.
      *
-     * @param Context $context Execution context
+     * @param Collection $collection
+     * @param Context    $context    Execution context
      * @return mixed
      * @throws LogicException if the collection operation is unsupported
      */
-    private function executeForCollection(Context $context)
+    private function executeForCollection(Collection $collection, Context $context)
     {
-        $collection = $context->getCollection($this->collectionOptions);
         $args = $context->prepareOptions($this->arguments);
         $context->replaceArgumentSessionPlaceholder($args);
 
@@ -207,6 +254,9 @@ final class Operation
                     array_diff_key($args, ['fieldName' => 1, 'filter' => 1])
                 );
 
+            case 'drop':
+                return $collection->drop($args);
+
             case 'findOneAndReplace':
                 if (isset($args['returnDocument'])) {
                     $args['returnDocument'] = ('after' === strtolower($args['returnDocument']))
@@ -262,13 +312,13 @@ final class Operation
     /**
      * Executes the database operation and return its result.
      *
-     * @param Context $context Execution context
+     * @param Database $database
+     * @param Context  $context  Execution context
      * @return mixed
      * @throws LogicException if the database operation is unsupported
      */
-    private function executeForDatabase(Context $context)
+    private function executeForDatabase(Database $database, Context $context)
     {
-        $database = $context->getDatabase($this->databaseOptions);
         $args = $context->prepareOptions($this->arguments);
         $context->replaceArgumentSessionPlaceholder($args);
 
@@ -287,14 +337,13 @@ final class Operation
     /**
      * Executes the session operation and return its result.
      *
+     * @param Session $session
      * @param Context $context Execution context
      * @return mixed
      * @throws LogicException if the session operation is unsupported
      */
-    private function executeForSession(Context $context)
+    private function executeForSession(Session $session, Context $context)
     {
-        $session = $context->{$this->object};
-
         switch ($this->name) {
             case 'abortTransaction':
                 return $session->abortTransaction();
@@ -362,6 +411,9 @@ final class Operation
             case 'deleteOne':
                 return ResultExpectation::ASSERT_DELETE;
 
+            case 'drop':
+                return ResultExpectation::ASSERT_NOTHING;
+
             case 'findOneAndDelete':
             case 'findOneAndReplace':
             case 'findOneAndUpdate':

tests/SpecTests/ResultExpectation.php

@@ -29,9 +29,11 @@ final class ResultExpectation
     const ASSERT_SAME_DOCUMENTS = 8;
     const ASSERT_MATCHES_DOCUMENT = 9;
     const ASSERT_NULL = 10;
+    const ASSERT_CALLABLE = 11;
 
     private $assertionType = self::ASSERT_NOTHING;
     private $expectedValue;
+    private $assertionCallable;
 
     private function __construct($assertionType, $expectedValue)
     {
@@ -57,6 +59,19 @@ final class ResultExpectation
         $this->expectedValue = $expectedValue;
     }
 
+    public static function fromChangeStreams(stdClass $result, callable $assertionCallable)
+    {
+        if (!property_exists($result, 'success')) {
+            return new self(self::ASSERT_NOTHING, null);
+        }
+
+        $o = new self(self::ASSERT_CALLABLE, $result->success);
+
+        $o->assertionCallable = $assertionCallable;
+
+        return $o;
+    }
+
     public static function fromRetryableWrites(stdClass $outcome, $defaultAssertionType)
     {
         if (property_exists($outcome, 'result')) {
@@ -138,6 +153,10 @@ final class ResultExpectation
                 }
                 break;
 
+            case self::ASSERT_CALLABLE:
+                call_user_func($this->assertionCallable, $expected, $actual);
+                break;
+
             case self::ASSERT_DELETE:
                 $test->assertInstanceOf(DeleteResult::class, $actual);
 

tests/SpecTests/RetryableWritesSpecTest.php

@@ -12,16 +12,6 @@ use stdClass;
  */
 class RetryableWritesSpecTest extends FunctionalTestCase
 {
-    public static function assertCommandMatches(stdClass $expected, stdClass $actual)
-    {
-        throw new LogicException('Retryable writes spec tests do not assert CommandStartedEvents');
-    }
-
-    public static function assertCommandReplyMatches(stdClass $expected, stdClass $actual)
-    {
-        throw new LogicException('Retryable writes spec tests do not assert CommandSucceededEvents');
-    }
-
     /**
      * Execute an individual test case from the specification.
      *

tests/SpecTests/TransactionsSpecTest.php

@@ -109,11 +109,6 @@ class TransactionsSpecTest extends FunctionalTestCase
         static::assertDocumentsMatch($expected, $actual);
     }
 
-    public static function assertCommandReplyMatches(stdClass $expected, stdClass $actual)
-    {
-        throw new LogicException('Transactions spec tests do not assert CommandSucceededEvents');
-    }
-
     /**
      * Execute an individual test case from the specification.
      *

tests/SpecTests/change-streams/change-streams-errors.json

+{
+  "collection_name": "test",
+  "database_name": "change-stream-tests",
+  "collection2_name": "test2",
+  "database2_name": "change-stream-tests-2",
+  "tests": [
+    {
+      "description": "The watch helper must not throw a custom exception when executed against a single server topology, but instead depend on a server error",
+      "minServerVersion": "3.6.0",
+      "target": "collection",
+      "topology": [
+        "single"
+      ],
+      "changeStreamPipeline": [],
+      "changeStreamOptions": {},
+      "operations": [],
+      "expectations": [],
+      "result": {
+        "error": {
+          "code": 40573
+        }
+      }
+    },
+    {
+      "description": "Change Stream should error when an invalid aggregation stage is passed in",
+      "minServerVersion": "3.6.0",
+      "target": "collection",
+      "topology": [
+        "replicaset"
+      ],
+      "changeStreamPipeline": [
+        {
+          "$unsupported": "foo"
+        }
+      ],
+      "changeStreamOptions": {},
+      "operations": [
+        {
+          "database": "change-stream-tests",
+          "collection": "test",
+          "name": "insertOne",
+          "arguments": {
+            "document": {
+              "z": 3
+            }
+          }
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "aggregate": "test",
+              "cursor": {},
+              "pipeline": [
+                {
+                  "$changeStream": {
+                    "fullDocument": "default"
+                  }
+                },
+                {
+                  "$unsupported": "foo"
+                }
+              ]
+            },
+            "command_name": "aggregate",
+            "database_name": "change-stream-tests"
+          }
+        }
+      ],
+      "result": {
+        "error": {
+          "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"
+          ]
+        }
+      }
+    }
+  ]
+}

tests/SpecTests/change-streams/change-streams-errors.yml

+collection_name: &collection_name "test"
+database_name: &database_name "change-stream-tests"
+collection2_name: &collection2_name "test2"
+database2_name: &database2_name "change-stream-tests-2"
+tests:
+  -
+    description: The watch helper must not throw a custom exception when executed against a single server topology, but instead depend on a server error
+    minServerVersion: "3.6.0"
+    target: collection
+    topology:
+      - single
+    changeStreamPipeline: []
+    changeStreamOptions: {}
+    operations: []
+    expectations: []
+    result:
+      error:
+        code: 40573
+  -
+    description: Change Stream should error when an invalid aggregation stage is passed in
+    minServerVersion: "3.6.0"
+    target: collection
+    topology:
+      - replicaset
+    changeStreamPipeline:
+      -
+        $unsupported: foo
+    changeStreamOptions: {}
+    operations:
+      -
+        database: *database_name
+        collection: *collection_name
+        name: insertOne
+        arguments:
+          document:
+            z: 3
+    expectations:
+      -
+        command_started_event:
+          command:
+            aggregate: *collection_name
+            cursor: {}
+            pipeline:
+              - 
+                $changeStream:
+                  fullDocument: default
+              -
+                $unsupported: foo
+          command_name: aggregate
+          database_name: *database_name
+    result:
+      error:
+        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" ]