Unskip tests that require replica set (#48)

Author: juliusgeo

.github/workflows/test.yml

@@ -25,6 +25,7 @@ jobs:
         uses: supercharge/[email protected]
         with:
           mongodb-version: ${{ matrix.mongodb-version }}
+          mongodb-replica-set: test-rs
       # add caching to reduce bandwidth and save time
       - uses: actions/cache@v2
         with:

pymongoexplain/explainable_collection.py

@@ -100,17 +100,26 @@ def watch(self, pipeline: Document = None, full_document: Document = None,
               max_await_time_ms: int = None, batch_size: int = None,
               collation=None, start_at_operation_time=None, session=None,
               start_after=None):
-        change_stream_options = {"start_after":start_after,
-                                 "resume_after":resume_after,
-                                "start_at_operation_time":start_at_operation_time,
-                                 "full_document":full_document}
+
+        change_stream_options = {}
+        if start_after:
+            change_stream_options["startAfter"] = start_after
+        if resume_after:
+            change_stream_options["resumeAfter"] = resume_after
+        if start_at_operation_time:
+            change_stream_options["startAtOperationTime"] = start_at_operation_time
+        if full_document:
+            change_stream_options["fullDocument"] = full_document
+
         if pipeline is not None:
             pipeline = [{"$changeStream": change_stream_options}]+pipeline
         else:
             pipeline = [{"$changeStream": change_stream_options}]
-
+        cursor_args = {}
+        if batch_size:
+            cursor_args["batchSize"] = batch_size
         command = AggregateCommand(self.collection, pipeline,
-                                    {"batch_size":batch_size},
+                                   cursor_args,
                                    {"collation":collation, "max_await_time_ms":
                                        max_await_time_ms})
         return self._explain_command(command)

test/test_collection.py

@@ -114,15 +114,14 @@ def test_delete_many(self):
         last_cmd_payload = self.explain.last_cmd_payload
         self._compare_command_dicts(last_cmd_payload, last_logger_payload)
 
-    @unittest.skip("Travis does not have replica sets set up yet")
     def test_watch(self):
         res = self.explain.watch()
         self.assertIn("queryPlanner", res["stages"][0]["$cursor"])
         self.collection.watch(pipeline=[{"$project": {"tags": 1}}],
-                               batch_size=10, full_document="updateLookup")
+                              batch_size=10, full_document="updateLookup")
         last_logger_payload = self.logger.cmd_payload
         res = self.explain.watch(pipeline=[{"$project": {"tags": 1}}],
-                            batch_size=10, full_document="updateLookup")
+                                 batch_size=10, full_document="updateLookup")
         self.assertIn("queryPlanner", res["stages"][0]["$cursor"])
         last_cmd_payload = self.explain.last_cmd_payload
         self._compare_command_dicts(last_cmd_payload, last_logger_payload)