<?php
 namespace MongoDB\Tests\SpecTests;
 use ArrayIterator;
 use LogicException;
 use MongoDB\ChangeStream;
 use MongoDB\Driver\Exception\Exception;
 use MongoDB\Model\BSONDocument;
 use MultipleIterator;
 use stdClass;
 use function basename;
 use function count;
 use function file_get_contents;
 use function glob;
 class ChangeStreamsSpecTest extends FunctionalTestCase {
 private static $incompleteTests = ['change-streams-errors: Change Stream should error when _id is projected out' => 'PHPC-1419'];
 public static function assertCommandMatches(stdClass $expected, stdClass $actual) {
 static::assertDocumentsMatch($expected, $actual);
 }
 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);
 }
 }
 public function testChangeStreams(stdClass $test, $databaseName = null, $collectionName = null, $database2Name = null, $collection2Name = null) {
 if (isset(self::$incompleteTests[$this->dataDescription()])) {
 $this->markTestIncomplete(self::$incompleteTests[$this->dataDescription()]);
 }
 $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();
 }
 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;
 }
 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->getClient()->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);
 }
 }
 private function createRunOn(stdClass $test) {
 $req = new stdClass();
 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];
 }
 private function dropDatabasesAndCreateCollection($databaseName, $collectionName) {
 $context = $this->getContext();
 $database = $context->getClient()->selectDatabase($databaseName);
 $database->drop($context->defaultWriteOptions);
 $database->createCollection($collectionName, $context->defaultWriteOptions);
 }
 private function iterateChangeStream(ChangeStream $changeStream, $limit = 0) {
 if ($limit < 0) {
 throw new LogicException('$limit is negative');
 }
 $maxIterations = $limit + 1;
 $events = [];
 for ($i = 0, $changeStream->rewind();
 $i < $maxIterations;
 $i++, $changeStream->next()) {
 if (! $changeStream->valid()) {
 continue;
 }
 $event = $changeStream->current();
 $this->assertInstanceOf(BSONDocument::class, $event);
 $events[] = $event;
 if (count($events) >= $limit) {
 break;
 }
 }
 return $events;
 }
 }
 