PYTHON-1041 - Update monitoring spec tests to latest

Also fix a number of bugs in the test runner.

Author: behackett

test/command_monitoring/find.json

@@ -270,6 +270,7 @@
     },
     {
       "description": "A successful find event with a getmore and killcursors",
+      "ignore_if_server_version_greater_than": "3.0",
       "operation": {
         "name": "find",
         "arguments": {
@@ -395,6 +396,108 @@
         }
       ]
     },
+    {
+      "description": "A successful find event with a getmore and the server kills the cursor",
+      "ignore_if_server_version_less_than": "3.1",
+      "ignore_if_topology_type" : ["sharded"],
+      "operation": {
+        "name": "find",
+        "arguments": {
+          "filter": {
+            "_id": {
+              "$gte": 1
+            }
+          },
+          "sort": {
+            "_id": 1
+          },
+          "batchSize": 3,
+          "limit": 4
+        }
+      },
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "find": "test",
+              "filter": {
+                "_id": {
+                  "$gte": 1
+                }
+              },
+              "sort": {
+                "_id": 1
+              },
+              "batchSize": 3,
+              "limit": 4
+            },
+            "command_name": "find",
+            "database_name": "command-monitoring-tests"
+          }
+        },
+        {
+          "command_succeeded_event": {
+            "reply": {
+              "ok": 1,
+              "cursor": {
+                "id": {
+                  "$numberLong": "42"
+                },
+                "ns": "command-monitoring-tests.test",
+                "firstBatch": [
+                  {
+                    "_id": 1,
+                    "x": 11
+                  },
+                  {
+                    "_id": 2,
+                    "x": 22
+                  },
+                  {
+                    "_id": 3,
+                    "x": 33
+                  }
+                ]
+              }
+            },
+            "command_name": "find"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "getMore": {
+                "$numberLong": "42"
+              },
+              "collection": "test",
+              "batchSize": 1
+            },
+            "command_name": "getMore",
+            "database_name": "command-monitoring-tests"
+          }
+        },
+        {
+          "command_succeeded_event": {
+            "reply": {
+              "ok": 1,
+              "cursor": {
+                "id": {
+                  "$numberLong": "0"
+                },
+                "ns": "command-monitoring-tests.test",
+                "nextBatch": [
+                  {
+                    "_id": 4,
+                    "x": 44
+                  }
+                ]
+              }
+            },
+            "command_name": "getMore"
+          }
+        }
+      ]
+    },
     {
       "description": "A failed find event",
       "operation": {

test/test_command_monitoring_spec.py

@@ -26,7 +26,8 @@
 from bson.json_util import object_hook
 from pymongo import monitoring
 from pymongo.errors import OperationFailure
-from pymongo.read_preferences import make_read_preference
+from pymongo.read_preferences import (make_read_preference,
+                                      read_pref_mode_from_name)
 from pymongo.write_concern import WriteConcern
 from test import unittest, client_context
 from test.utils import single_client, wait_until, EventListener
@@ -49,7 +50,6 @@ class TestAllScenarios(unittest.TestCase):
     @client_context.require_connection
     def setUpClass(cls):
         cls.listener = EventListener()
-        cls.listener.add_command_filter('killCursors')
         cls.saved_listeners = monitoring._LISTENERS
         monitoring._LISTENERS = monitoring._Listeners([])
         cls.client = single_client(event_listeners=[cls.listener])
@@ -68,27 +68,42 @@ def run_scenario(self):
         dbname = scenario_def['database_name']
         collname = scenario_def['collection_name']
 
-        # Clear the kill cursors queue.
-        self.client._kill_cursors_executor.wake()
-
         for test in scenario_def['tests']:
+            ver = client_context.version[:2]
+            if "ignore_if_server_version_greater_than" in test:
+                version = test["ignore_if_server_version_greater_than"]
+                if ver > tuple(map(int, version.split("."))):
+                    continue
+            if "ignore_if_server_version_less_than" in test:
+                version = test["ignore_if_server_version_less_than"]
+                if ver < tuple(map(int, version.split("."))):
+                    continue
+            if "ignore_if_topology_type" in test:
+                types = set(test["ignore_if_topology_type"])
+                if "sharded" in types and client_context.is_mongos:
+                    continue
+
             coll = self.client[dbname][collname]
             coll.drop()
             coll.insert_many(scenario_def['data'])
             self.listener.results.clear()
             name = camel_to_snake(test['operation']['name'])
-            args = test['operation']['arguments']
             # Don't send $readPreference to mongos before 2.4.
             if (client_context.version.at_least(2, 4, 0)
-                    and 'readPreference' in args):
-                pref = make_read_preference(
-                    args['readPreference']['mode'], None)
-                coll = coll.with_options(read_preference=pref)
-            if 'writeConcern' in args:
+                    and 'read_preference' in test['operation']):
+                mode = read_pref_mode_from_name(
+                    test['operation']['read_preference']['mode'])
+                coll = coll.with_options(
+                    read_preference=make_read_preference(mode, None))
+
+            test_args = test['operation']['arguments']
+            if 'writeConcern' in test_args:
+                concern = test_args.pop('writeConcern')
                 coll = coll.with_options(
-                    write_concern=WriteConcern(**args['writeConcern']))
-            for arg in args:
-                args[camel_to_snake(arg)] = args.pop(arg)
+                    write_concern=WriteConcern(**concern))
+            args = {}
+            for arg in test_args:
+                args[camel_to_snake(arg)] = test_args[arg]
 
             if name == 'bulk_write':
                 bulk_args = []
@@ -102,25 +117,20 @@ def run_scenario(self):
                 except OperationFailure:
                     pass
             elif name == 'find':
-                if 'limit' in args:
-                    # XXX: Skip killCursors test when using the find command.
-                    if client_context.version.at_least(3, 1, 1):
-                        continue
-                    self.listener.remove_command_filter('killCursors')
                 if 'sort' in args:
                     args['sort'] = list(args['sort'].items())
                 try:
                     # Iterate the cursor.
                     tuple(coll.find(**args))
                 except OperationFailure:
                     pass
-                # Wait for the killCursors thread to run.
-                if 'limit' in args:
+                # Wait for the killCursors thread to run if necessary.
+                if 'limit' in args and client_context.version[:2] < (3, 1):
+                    self.client._kill_cursors_executor.wake()
                     started = self.listener.results['started']
                     wait_until(
                         lambda: started[-1].command_name == 'killCursors',
                         "publish a start event for killCursors.")
-                    self.listener.add_command_filter('killCursors')
             else:
                 try:
                     getattr(coll, name)(**args)
@@ -132,7 +142,8 @@ def run_scenario(self):
                 if event_type == "command_started_event":
                     event = self.listener.results['started'].pop(0)
                     # The tests substitute 42 for any number other than 0.
-                    if event.command_name == 'getMore':
+                    if (event.command_name == 'getMore'
+                            and event.command['getMore']):
                         event.command['getMore'] = 42
                     elif event.command_name == 'killCursors':
                         event.command['cursors'] = [42]