PYTHON-2475 Implement Atlas Data Lake prose specification tests (#665)

(cherry picked from commit 00ed232)

Author: prashantmital

.evergreen/config.yml

@@ -309,17 +309,15 @@ functions:
         script: |
           set -o xtrace
           ${PREPARE_SHELL}
-          cd ${DRIVERS_TOOLS}/.evergreen/atlas_data_lake
-          DRIVERS_TOOLS="${DRIVERS_TOOLS}" sh build-mongohouse-local.sh
+          sh ${DRIVERS_TOOLS}/.evergreen/atlas_data_lake/build-mongohouse-local.sh
     - command: shell.exec
       type: setup
       params:
         background: true
         script: |
           set -o xtrace
           ${PREPARE_SHELL}
-          cd ${DRIVERS_TOOLS}/.evergreen/atlas_data_lake
-          DRIVERS_TOOLS="${DRIVERS_TOOLS}" sh run-mongohouse-local.sh
+          sh ${DRIVERS_TOOLS}/.evergreen/atlas_data_lake/run-mongohouse-local.sh
 
   "stop mongo-orchestration":
     - command: shell.exec

test/__init__.py

@@ -524,6 +524,13 @@ def require_connection(self, func):
             "Cannot connect to MongoDB on %s" % (self.pair,),
             func=func)
 
+    def require_data_lake(self, func):
+        """Run a test only if we are connected to Atlas Data Lake."""
+        return self._require(
+            lambda: self.is_data_lake,
+            "Not connected to Atlas Data Lake 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

test/test_data_lake.py

@@ -19,24 +19,86 @@
 
 sys.path[0:0] = [""]
 
-from test import client_context, unittest
+from pymongo.auth import MECHANISMS
+from test import client_context, unittest, IntegrationTest
 from test.crud_v2_format import TestCrudV2
-from test.utils import TestCreator
+from test.utils import (
+    rs_client_noauth, rs_or_single_client, OvertCommandListener, TestCreator)
 
 
 # Location of JSON test specifications.
 _TEST_PATH = os.path.join(
     os.path.dirname(os.path.realpath(__file__)), "data_lake")
 
 
+class TestDataLakeMustConnect(IntegrationTest):
+    def test_connected_to_data_lake(self):
+        data_lake = os.environ.get('DATA_LAKE')
+        if not data_lake:
+            self.skipTest('DATA_LAKE is not set')
+
+        self.assertTrue(client_context.is_data_lake,
+                        'client context.is_data_lake must be True when '
+                        'DATA_LAKE is set')
+
+
+class TestDataLakeProse(IntegrationTest):
+    # Default test database and collection names.
+    TEST_DB = 'test'
+    TEST_COLLECTION = 'driverdata'
+
+    @classmethod
+    @client_context.require_data_lake
+    def setUpClass(cls):
+        super(TestDataLakeProse, cls).setUpClass()
+
+    # Test killCursors
+    def test_1(self):
+        listener = OvertCommandListener()
+        client = rs_or_single_client(event_listeners=[listener])
+        cursor = client[self.TEST_DB][self.TEST_COLLECTION].find(
+            {}, batch_size=2)
+        next(cursor)
+
+        # find command assertions
+        find_cmd = listener.results["succeeded"][-1]
+        self.assertEqual(find_cmd.command_name, "find")
+        cursor_id = find_cmd.reply["cursor"]["id"]
+        cursor_ns = find_cmd.reply["cursor"]["ns"]
+
+        # killCursors command assertions
+        cursor.close()
+        started = listener.results["started"][-1]
+        self.assertEqual(started.command_name, 'killCursors')
+        succeeded = listener.results["succeeded"][-1]
+        self.assertEqual(succeeded.command_name, 'killCursors')
+
+        self.assertIn(cursor_id, started.command["cursors"])
+        target_ns = ".".join([started.command['$db'],
+                              started.command['killCursors']])
+        self.assertEqual(cursor_ns, target_ns)
+
+        self.assertIn(cursor_id, succeeded.reply["cursorsKilled"])
+
+    # Test no auth
+    def test_2(self):
+        client = rs_client_noauth()
+        client.admin.command('ping')
+
+    # Test with auth
+    def test_3(self):
+        for mechanism in ['SCRAM-SHA-1', 'SCRAM-SHA-256']:
+            client = rs_or_single_client(authMechanism=mechanism)
+            client[self.TEST_DB][self.TEST_COLLECTION].find_one()
+
+
 class DataLakeTestSpec(TestCrudV2):
     # Default test database and collection names.
     TEST_DB = 'test'
     TEST_COLLECTION = 'driverdata'
 
     @classmethod
-    @unittest.skipUnless(client_context.is_data_lake,
-                         'Not connected to Atlas Data Lake')
+    @client_context.require_data_lake
     def setUpClass(cls):
         super(DataLakeTestSpec, cls).setUpClass()