PHPLIB-881: Consolidate validation of aggregation pipeline options (#1078)

- Use MongoDB\is_pipeline to detect or validate pipelines
- Use standard wording for "Aggregation pipeline"

Author: GromNaN

psalm-baseline.xml

@@ -418,7 +418,7 @@
   </file>
   <file src="src/Operation/CreateCollection.php">
     <MixedArgument occurrences="2">
-      <code>$i</code>
+      <code>$options['pipeline']</code>
       <code>$this-&gt;options['typeMap']</code>
     </MixedArgument>
     <MixedAssignment occurrences="4">

src/Client.php

@@ -352,7 +352,7 @@ public function startSession(array $options = [])
      * Create a change stream for watching changes to the cluster.
      *
      * @see Watch::__construct() for supported options
-     * @param array $pipeline List of pipeline operations
+     * @param array $pipeline Aggregation pipeline
      * @param array $options  Command options
      * @return ChangeStream
      * @throws InvalidArgumentException for parameter/option parsing errors

src/Collection.php

@@ -199,7 +199,7 @@ public function __toString()
      * "result" array from the command response document.
      *
      * @see Aggregate::__construct() for supported options
-     * @param array $pipeline List of pipeline operations
+     * @param array $pipeline Aggregation pipeline
      * @param array $options  Command options
      * @return Traversable
      * @throws UnexpectedValueException if the command response was malformed
@@ -1105,7 +1105,7 @@ public function updateOne($filter, $update, array $options = [])
      * Create a change stream for watching changes to the collection.
      *
      * @see Watch::__construct() for supported options
-     * @param array $pipeline List of pipeline operations
+     * @param array $pipeline Aggregation pipeline
      * @param array $options  Command options
      * @return ChangeStream
      * @throws InvalidArgumentException for parameter/option parsing errors

src/Database.php

@@ -187,7 +187,7 @@ public function __toString()
      * and $listLocalSessions. Requires MongoDB >= 3.6
      *
      * @see Aggregate::__construct() for supported options
-     * @param array $pipeline List of pipeline operations
+     * @param array $pipeline Aggregation pipeline
      * @param array $options  Command options
      * @return Traversable
      * @throws UnexpectedValueException if the command response was malformed
@@ -598,7 +598,7 @@ public function selectGridFSBucket(array $options = [])
      * Create a change stream for watching changes to the database.
      *
      * @see Watch::__construct() for supported options
-     * @param array $pipeline List of pipeline operations
+     * @param array $pipeline Aggregation pipeline
      * @param array $options  Command options
      * @return ChangeStream
      * @throws InvalidArgumentException for parameter/option parsing errors

src/Operation/Aggregate.php

@@ -31,7 +31,6 @@
 use MongoDB\Exception\UnsupportedException;
 use stdClass;
 
-use function array_is_list;
 use function current;
 use function is_array;
 use function is_bool;
@@ -40,7 +39,7 @@
 use function is_string;
 use function MongoDB\create_field_path_type_map;
 use function MongoDB\is_last_pipeline_operator_write;
-use function sprintf;
+use function MongoDB\is_pipeline;
 
 /**
  * Operation for the aggregate command.
@@ -132,20 +131,14 @@ class Aggregate implements Executable, Explainable
      *
      * @param string      $databaseName   Database name
      * @param string|null $collectionName Collection name
-     * @param array       $pipeline       List of pipeline operations
+     * @param array       $pipeline       Aggregation pipeline
      * @param array       $options        Command options
      * @throws InvalidArgumentException for parameter/option parsing errors
      */
     public function __construct(string $databaseName, ?string $collectionName, array $pipeline, array $options = [])
     {
-        if (! array_is_list($pipeline)) {
-            throw new InvalidArgumentException('$pipeline is not a list');
-        }
-
-        foreach ($pipeline as $i => $operation) {
-            if (! is_array($operation) && ! is_object($operation)) {
-                throw InvalidArgumentException::invalidType(sprintf('$pipeline[%d]', $i), $operation, 'array or object');
-            }
+        if (! is_pipeline($pipeline, true /* allowEmpty */)) {
+            throw new InvalidArgumentException('$pipeline is not a valid aggregation pipeline');
         }
 
         $options += ['useCursor' => true];

src/Operation/CreateCollection.php

@@ -24,15 +24,13 @@
 use MongoDB\Driver\WriteConcern;
 use MongoDB\Exception\InvalidArgumentException;
 
-use function array_is_list;
-use function assert;
 use function current;
 use function is_array;
 use function is_bool;
 use function is_integer;
 use function is_object;
 use function is_string;
-use function sprintf;
+use function MongoDB\is_pipeline;
 use function trigger_error;
 
 use const E_USER_DEPRECATED;
@@ -242,18 +240,8 @@ public function __construct(string $databaseName, string $collectionName, array
             trigger_error('The "autoIndexId" option is deprecated and will be removed in a future release', E_USER_DEPRECATED);
         }
 
-        if (isset($options['pipeline'])) {
-            $pipeline = $options['pipeline'];
-            assert(is_array($pipeline));
-            if (! array_is_list($pipeline)) {
-                throw new InvalidArgumentException('The "pipeline" option is not a list');
-            }
-
-            foreach ($options['pipeline'] as $i => $operation) {
-                if (! is_array($operation) && ! is_object($operation)) {
-                    throw InvalidArgumentException::invalidType(sprintf('$options["pipeline"][%d]', $i), $operation, 'array or object');
-                }
-            }
+        if (isset($options['pipeline']) && ! is_pipeline($options['pipeline'], true /* allowEmpty */)) {
+            throw new InvalidArgumentException('"pipeline" option is not a valid aggregation pipeline');
         }
 
         $this->databaseName = $databaseName;

src/Operation/Watch.php

@@ -196,7 +196,7 @@ class Watch implements Executable, /* @internal */ CommandSubscriber
      * @param Manager     $manager        Manager instance from the driver
      * @param string|null $databaseName   Database name
      * @param string|null $collectionName Collection name
-     * @param array       $pipeline       List of pipeline operations
+     * @param array       $pipeline       Aggregation pipeline
      * @param array       $options        Command options
      * @throws InvalidArgumentException for parameter/option parsing errors
      */

src/functions.php

@@ -300,7 +300,7 @@ function is_in_transaction(array $options): bool
  * executed against a primary server.
  *
  * @internal
- * @param array $pipeline List of pipeline operations
+ * @param array $pipeline Aggregation pipeline
  */
 function is_last_pipeline_operator_write(array $pipeline): bool
 {

tests/FunctionsTest.php

@@ -243,9 +243,9 @@ public function testIsLastPipelineOperatorWrite(callable $cast): void
     }
 
     /** @dataProvider providePipelines */
-    public function testIsPipeline($expected, $pipeline): void
+    public function testIsPipeline($expected, $pipeline, $allowEmpty = false): void
     {
-        $this->assertSame($expected, is_pipeline($pipeline));
+        $this->assertSame($expected, is_pipeline($pipeline, $allowEmpty));
     }
 
     public function providePipelines(): array
@@ -286,6 +286,9 @@ public function providePipelines(): array
             'invalid pipeline element type: Serializable' => [false, new BSONArray([new BSONArray([])])],
             'invalid pipeline element type: PackedArray' => [false, PackedArray::fromPHP([[]])],
             // Empty array has no pipeline stages
+            'valid empty: array' => [true, [], true],
+            'valid empty: Serializable' => [true, new BSONArray([]), true],
+            'valid empty: PackedArray' => [true, PackedArray::fromPHP([]), true],
             'invalid empty: array' => [false, []],
             'invalid empty: Serializable' => [false, new BSONArray([])],
             'invalid empty: PackedArray' => [false, PackedArray::fromPHP([])],

tests/Operation/AggregateTest.php

@@ -10,7 +10,7 @@ class AggregateTest extends TestCase
     public function testConstructorPipelineArgumentMustBeAList(): void
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('$pipeline is not a list');
+        $this->expectExceptionMessage('$pipeline is not a valid aggregation pipeline');
         new Aggregate($this->getDatabaseName(), $this->getCollectionName(), [1 => ['$match' => ['x' => 1]]]);
     }
 

tests/Operation/CreateCollectionTest.php

@@ -10,7 +10,7 @@ class CreateCollectionTest extends TestCase
     public function testConstructorPipelineOptionMustBeAList(): void
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('The "pipeline" option is not a list');
+        $this->expectExceptionMessage('"pipeline" option is not a valid aggregation pipeline');
         new CreateCollection($this->getDatabaseName(), $this->getCollectionName(), ['pipeline' => [1 => ['$match' => ['x' => 1]]]]);
     }
 

tests/Operation/WatchTest.php

@@ -23,7 +23,7 @@ public function testConstructorCollectionNameShouldBeNullIfDatabaseNameIsNull():
     public function testConstructorPipelineArgumentMustBeAList(): void
     {
         $this->expectException(InvalidArgumentException::class);
-        $this->expectExceptionMessage('$pipeline is not a list');
+        $this->expectExceptionMessage('$pipeline is not a valid aggregation pipeline');
 
         /* Note: Watch uses array_unshift() to prepend the $changeStream stage
          * to the pipeline. Since array_unshift() reindexes numeric keys, we'll