PYTHON-1597 Don't run Change Stream tests on MMAPv1 (mongodb#356)

Author: prashantmital

test/__init__.py

@@ -162,6 +162,7 @@ def __init__(self):
         self.nodes = set()
         self.replica_set_name = None
         self.cmd_line = None
+        self.server_status = None
         self.version = Version(-1)  # Needs to be comparable with Version
         self.auth_enabled = False
         self.test_commands_enabled = False
@@ -248,6 +249,7 @@ def _init_client(self):
                 # May not have this if OperationFailure was raised earlier.
                 self.cmd_line = self.client.admin.command('getCmdLineOpts')
 
+            self.server_status = self.client.admin.command('serverStatus')
             self.ismaster = ismaster = self.client.admin.command('isMaster')
             self.sessions_enabled = 'logicalSessionTimeoutMinutes' in ismaster
 
@@ -424,6 +426,25 @@ def require_connection(self, func):
             "Cannot connect to MongoDB on %s" % (self.pair,),
             func=func)
 
+    def require_no_mmap(self, func):
+        """Run a test only if the server is not using the MMAPv1 storage
+        engine. Only works for standalone and replica sets; tests are
+        run regardless of storage engine on sharded clusters. """
+        def is_not_mmap():
+            if self.is_mongos:
+                return True
+            try:
+                storage_engine = self.server_status.get(
+                    'storageEngine').get('name')
+            except AttributeError:
+                # Raised if the storageEngine key does not exist or if
+                # self.server_status is None.
+                return False
+            return storage_engine != 'mmapv1'
+
+        return self._require(
+            is_not_mmap, "Storage engine must not be MMAPv1", func=func)
+
     def require_version_min(self, *ver):
         """Run a test only if the server version is at least ``version``."""
         other_version = Version(*ver)

test/test_change_stream.py

@@ -54,6 +54,7 @@ class TestClusterChangeStream(IntegrationTest):
 
     @classmethod
     @client_context.require_version_min(4, 0, 0, -1)
+    @client_context.require_no_mmap
     @client_context.require_no_standalone
     def setUpClass(cls):
         super(TestClusterChangeStream, cls).setUpClass()
@@ -97,6 +98,7 @@ class TestDatabaseChangeStream(IntegrationTest):
 
     @classmethod
     @client_context.require_version_min(4, 0, 0, -1)
+    @client_context.require_no_mmap
     @client_context.require_no_standalone
     def setUpClass(cls):
         super(TestDatabaseChangeStream, cls).setUpClass()
@@ -147,6 +149,7 @@ class TestCollectionChangeStream(IntegrationTest):
 
     @classmethod
     @client_context.require_version_min(3, 5, 11)
+    @client_context.require_no_mmap
     @client_context.require_no_standalone
     def setUpClass(cls):
         super(TestCollectionChangeStream, cls).setUpClass()
@@ -645,6 +648,8 @@ def create_tests():
 
             for test in scenario_def['tests']:
                 new_test = create_test(scenario_def, test)
+                new_test = client_context.require_no_mmap(new_test)
+
                 if 'minServerVersion' in test:
                     min_ver = tuple(
                         int(elt) for