Skip to content

Commit d9858d1

Browse files
jmikolajmikola
committed
Refactor change stream tests use Operation instead of Collection
For consistency with other operation functional tests, this now constructs and executes operation classes directly.
1 parent 15b4702 commit d9858d1

File tree

1 file changed

+54
-73
lines changed

1 file changed

+54
-73
lines changed

‎tests/Operation/WatchFunctionalTest.php

Lines changed: 54 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -3,86 +3,78 @@
33
namespace MongoDB\Tests\Operation;
44

55
use MongoDB\Client;
6-
use MongoDB\Collection;
76
use MongoDB\Operation\DatabaseCommand;
7+
use MongoDB\Operation\InsertOne;
8+
use MongoDB\Operation\Watch;
89

910
class WatchFunctionalTest extends FunctionalTestCase
1011
{
1112
public function setUp()
1213
{
1314
parent::setUp();
15+
1416
if (version_compare($this->getFeatureCompatibilityVersion(), '3.6', '<')) {
1517
$this->markTestSkipped('$changeStream is only supported on FCV 3.6 or higher');
1618
}
1719
}
1820

1921
public function testResume()
2022
{
21-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
23+
$this->insertDocument(['_id' => 1, 'x' => 'foo']);
2224

23-
$result = $this->collection->insertOne(['x' => 1]);
24-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
25-
$this->assertSame(1, $result->getInsertedCount());
25+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
26+
$changeStream = $operation->execute($this->getPrimaryServer());
2627

27-
$changeStream = $this->collection->watch();
2828
$changeStream->rewind();
2929
$this->assertNull($changeStream->current());
3030

31-
$result = $this->collection->insertOne(['x' => 2]);
32-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
33-
$this->assertSame(1, $result->getInsertedCount());
31+
$this->insertDocument(['_id' => 2, 'x' => 'bar']);
3432

3533
$changeStream->next();
3634
$expectedResult = (object) ([
3735
'_id' => $changeStream->current()->_id,
3836
'operationType' => 'insert',
39-
'fullDocument' => (object) ['_id' => $result->getInsertedId(), 'x' => 2],
37+
'fullDocument' => (object) ['_id' => 2, 'x' => 'bar'],
4038
'ns' => (object) ['db' => 'phplib_test', 'coll' => 'WatchFunctionalTest.e68b9f01'],
41-
'documentKey' => (object) ['_id' => $result->getInsertedId()]
39+
'documentKey' => (object) ['_id' => 2]
4240
]);
4341
$this->assertEquals($changeStream->current(), $expectedResult);
4442

4543
$operation = new DatabaseCommand($this->getDatabaseName(), ["killCursors" => $this->getCollectionName(), "cursors" => [$changeStream->getCursorId()]]);
4644
$operation->execute($this->getPrimaryServer());
4745

48-
$result = $this->collection->insertOne(['x' => 3]);
49-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
50-
$this->assertSame(1, $result->getInsertedCount());
46+
$this->insertDocument(['_id' => 3, 'x' => 'baz']);
5147

5248
$changeStream->next();
5349
$expectedResult = (object) ([
5450
'_id' => $changeStream->current()->_id,
5551
'operationType' => 'insert',
56-
'fullDocument' => (object) ['_id' => $result->getInsertedId(), 'x' => 3],
52+
'fullDocument' => (object) ['_id' => 3, 'x' => 'baz'],
5753
'ns' => (object) ['db' => 'phplib_test', 'coll' => 'WatchFunctionalTest.e68b9f01'],
58-
'documentKey' => (object) ['_id' => $result->getInsertedId()]
54+
'documentKey' => (object) ['_id' => 3]
5955
]);
6056
$this->assertEquals($changeStream->current(), $expectedResult);
6157
}
6258

6359
public function testNoChangeAfterResumeBeforeInsert()
6460
{
65-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
61+
$this->insertDocument(['_id' => 1, 'x' => 'foo']);
6662

67-
$result = $this->collection->insertOne(['x' => 1]);
68-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
69-
$this->assertSame(1, $result->getInsertedCount());
63+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
64+
$changeStream = $operation->execute($this->getPrimaryServer());
7065

71-
$changeStream = $this->collection->watch();
7266
$changeStream->rewind();
7367
$this->assertNull($changeStream->current());
7468

75-
$result = $this->collection->insertOne(['x' => 2]);
76-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
77-
$this->assertSame(1, $result->getInsertedCount());
69+
$this->insertDocument(['_id' => 2, 'x' => 'bar']);
7870

7971
$changeStream->next();
8072
$expectedResult = (object) ([
8173
'_id' => $changeStream->current()->_id,
8274
'operationType' => 'insert',
83-
'fullDocument' => (object) ['_id' => $result->getInsertedId(), 'x' => 2],
75+
'fullDocument' => (object) ['_id' => 2, 'x' => 'bar'],
8476
'ns' => (object) ['db' => 'phplib_test', 'coll' => 'WatchFunctionalTest.4a554985'],
85-
'documentKey' => (object) ['_id' => $result->getInsertedId()]
77+
'documentKey' => (object) ['_id' => 2]
8678
]);
8779
$this->assertEquals($changeStream->current(), $expectedResult);
8880

@@ -92,26 +84,23 @@ public function testNoChangeAfterResumeBeforeInsert()
9284
$changeStream->next();
9385
$this->assertNull($changeStream->current());
9486

95-
$result = $this->collection->insertOne(['x' => 3]);
96-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
97-
$this->assertSame(1, $result->getInsertedCount());
87+
$this->insertDocument(['_id' => 3, 'x' => 'baz']);
9888

9989
$changeStream->next();
10090
$expectedResult = (object) ([
10191
'_id' => $changeStream->current()->_id,
10292
'operationType' => 'insert',
103-
'fullDocument' => (object) ['_id' => $result->getInsertedId(), 'x' => 3],
93+
'fullDocument' => (object) ['_id' => 3, 'x' => 'baz'],
10494
'ns' => (object) ['db' => 'phplib_test', 'coll' => 'WatchFunctionalTest.4a554985'],
105-
'documentKey' => (object) ['_id' => $result->getInsertedId()]
95+
'documentKey' => (object) ['_id' => 3]
10696
]);
10797
$this->assertEquals($changeStream->current(), $expectedResult);
10898
}
10999

110100
public function testResumeAfterKillThenNoOperations()
111101
{
112-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
113-
114-
$changeStream = $this->collection->watch();
102+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
103+
$changeStream = $operation->execute($this->getPrimaryServer());
115104

116105
$operation = new DatabaseCommand($this->getDatabaseName(), ["killCursors" => $this->getCollectionName(), "cursors" => [$changeStream->getCursorId()]]);
117106
$operation->execute($this->getPrimaryServer());
@@ -122,32 +111,26 @@ public function testResumeAfterKillThenNoOperations()
122111

123112
public function testResumeAfterKillThenOperation()
124113
{
125-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
126-
127-
$changeStream = $this->collection->watch();
114+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
115+
$changeStream = $operation->execute($this->getPrimaryServer());
128116

129117
$operation = new DatabaseCommand($this->getDatabaseName(), ["killCursors" => $this->getCollectionName(), "cursors" => [$changeStream->getCursorId()]]);
130118
$operation->execute($this->getPrimaryServer());
131119

132-
$result = $this->collection->insertOne(['x' => 3]);
133-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
134-
$this->assertSame(1, $result->getInsertedCount());
120+
$this->insertDocument(['_id' => 1, 'x' => 'foo']);
135121

136122
$changeStream->next();
137123
$this->assertNull($changeStream->current());
138124
}
139125

140126
public function testKey()
141127
{
142-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
143-
144-
$changeStream = $this->collection->watch();
128+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
129+
$changeStream = $operation->execute($this->getPrimaryServer());
145130

146131
$this->assertNull($changeStream->key());
147132

148-
$result = $this->collection->insertOne(['x' => 1]);
149-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
150-
$this->assertSame(1, $result->getInsertedCount());
133+
$this->insertDocument(['_id' => 1, 'x' => 'foo']);
151134

152135
$changeStream->next();
153136
$this->assertSame(1, $changeStream->key());
@@ -163,24 +146,20 @@ public function testKey()
163146
$changeStream->next();
164147
$this->assertNull($changeStream->key());
165148

166-
$result = $this->collection->insertOne(['x' => 2]);
167-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
168-
$this->assertSame(1, $result->getInsertedCount());
149+
$this->insertDocument(['_id' => 2, 'x' => 'bar']);
169150

170151
$changeStream->next();
171152
$this->assertSame(2, $changeStream->key());
172153
}
173154

174155
public function testNonEmptyPipeline()
175156
{
176-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
177-
178157
$pipeline = [['$project' => ['foo' => [0]]]];
179-
$changeStream = $this->collection->watch($pipeline, []);
180158

181-
$result = $this->collection->insertOne(['x' => 1]);
182-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
183-
$this->assertSame(1, $result->getInsertedCount());
159+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), $pipeline);
160+
$changeStream = $operation->execute($this->getPrimaryServer());
161+
162+
$this->insertDocument(['_id' => 1]);
184163

185164
$changeStream->next();
186165
$expectedResult = (object) ([
@@ -192,9 +171,9 @@ public function testNonEmptyPipeline()
192171

193172
public function testCursorWithEmptyBatchNotClosed()
194173
{
195-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
174+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), []);
175+
$changeStream = $operation->execute($this->getPrimaryServer());
196176

197-
$changeStream = $this->collection->watch();
198177
$this->assertNotNull($changeStream);
199178
}
200179

@@ -203,14 +182,12 @@ public function testCursorWithEmptyBatchNotClosed()
203182
*/
204183
public function testFailureAfterResumeTokenRemoved()
205184
{
206-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
207-
208185
$pipeline = [['$project' => ['_id' => 0 ]]];
209-
$changeStream = $this->collection->watch($pipeline, []);
210186

211-
$result = $this->collection->insertOne(['x' => 1]);
212-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
213-
$this->assertSame(1, $result->getInsertedCount());
187+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), $pipeline);
188+
$changeStream = $operation->execute($this->getPrimaryServer());
189+
190+
$this->insertDocument(['x' => 1]);
214191

215192
$changeStream->next();
216193
}
@@ -223,29 +200,28 @@ public function testConnectionException()
223200
$changeStream = $collection->watch();
224201
$changeStream->next();
225202

226-
$result = $collection->insertOne(['x' => 1]);
227-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
228-
$this->assertSame(1, $result->getInsertedCount());
203+
$this->insertDocument(['_id' => 1, 'x' => 'foo']);
229204

230205
$changeStream->next();
231206
$expectedResult = (object) ([
232207
'_id' => $changeStream->current()->_id,
233208
'operationType' => 'insert',
234-
'fullDocument' => (object) ['_id' => $result->getInsertedId(), 'x' => 1],
209+
'fullDocument' => (object) ['_id' => 1, 'x' => 'foo'],
235210
'ns' => (object) ['db' => 'phplib_test', 'coll' => 'WatchFunctionalTest.226d95f1'],
236-
'documentKey' => (object) ['_id' => $result->getInsertedId()]
211+
'documentKey' => (object) ['_id' => 1]
237212
]);
238213
$this->assertEquals($changeStream->current(), $expectedResult);
239214
}
240215

241216
public function testMaxAwaitTimeMS()
242217
{
243-
$this->collection = new Collection($this->manager, $this->getDatabaseName(), $this->getCollectionName());
244218
/* On average, an acknowledged write takes about 20 ms to appear in a
245219
* change stream on the server so we'll use a higher maxAwaitTimeMS to
246220
* ensure we see the write. */
247221
$maxAwaitTimeMS = 100;
248-
$changeStream = $this->collection->watch([], ['maxAwaitTimeMS' => $maxAwaitTimeMS]);
222+
223+
$operation = new Watch($this->manager, $this->getDatabaseName(), $this->getCollectionName(), [], ['maxAwaitTimeMS' => $maxAwaitTimeMS]);
224+
$changeStream = $operation->execute($this->getPrimaryServer());
249225

250226
/* The initial change stream is empty so we should expect a delay when
251227
* we call rewind, since it issues a getMore. Expect to wait at least
@@ -272,14 +248,19 @@ public function testMaxAwaitTimeMS()
272248

273249
/* After inserting a document, the change stream will not issue a
274250
* getMore so we should not expect a delay. */
275-
$result = $this->collection->insertOne(['_id' => 1]);
276-
$this->assertInstanceOf('MongoDB\InsertOneResult', $result);
277-
$this->assertSame(1, $result->getInsertedCount());
251+
$this->insertDocument(['_id' => 1]);
278252

279253
$startTime = microtime(true);
280254
$changeStream->next();
281255
$duration = microtime(true) - $startTime;
282256
$this->assertLessThan($maxAwaitTimeMS * 0.001, $duration);
283257
$this->assertTrue($changeStream->valid());
284258
}
259+
260+
private function insertDocument($document)
261+
{
262+
$insertOne = new InsertOne($this->getDatabaseName(), $this->getCollectionName(), $document);
263+
$writeResult = $insertOne->execute($this->getPrimaryServer());
264+
$this->assertEquals(1, $writeResult->getInsertedCount());
265+
}
285266
}

0 commit comments

Comments
 (0)