PYTHON-2294 Resync SDAM spec tests to workaround slow elections Windows and macOS (#468)

PYTHON-2296 Test behavior of connectTimeoutMS=0 with streaming protocol
PYTHON-2311 Workaround inherent race in flaky streaming protocol test

Author: ShaneHarvey

pymongo/common.py

@@ -333,6 +333,16 @@ def validate_timeout_or_zero(option, value):
     return validate_positive_float(option, value) / 1000.0
 
 
+def validate_timeout_or_none_or_zero(option, value):
+    """Validates a timeout specified in milliseconds returning
+    a value in floating point seconds. value=0 and value="0" are treated the
+    same as value=None which means unlimited timeout.
+    """
+    if value is None or value == 0 or value == "0":
+        return None
+    return validate_positive_float(option, value) / 1000.0
+
+
 def validate_max_staleness(option, value):
     """Validates maxStalenessSeconds according to the Max Staleness Spec."""
     if value == -1 or value == "-1":
@@ -593,7 +603,7 @@ def validate_tzinfo(dummy, value):
     'authmechanismproperties': validate_auth_mechanism_properties,
     'authsource': validate_string,
     'compressors': validate_compressors,
-    'connecttimeoutms': validate_timeout_or_none,
+    'connecttimeoutms': validate_timeout_or_none_or_zero,
     'directconnection': validate_boolean_or_string,
     'heartbeatfrequencyms': validate_timeout_or_none,
     'journal': validate_boolean_or_string,
@@ -608,7 +618,7 @@ def validate_tzinfo(dummy, value):
     'retryreads': validate_boolean_or_string,
     'retrywrites': validate_boolean_or_string,
     'serverselectiontimeoutms': validate_timeout_or_zero,
-    'sockettimeoutms': validate_timeout_or_none,
+    'sockettimeoutms': validate_timeout_or_none_or_zero,
     'ssl_keyfile': validate_readable,
     'tls': validate_boolean_or_string,
     'tlsallowinvalidcertificates': validate_allow_invalid_certs,

pymongo/mongo_client.py

@@ -227,11 +227,13 @@ def __init__(
           - `socketTimeoutMS`: (integer or None) Controls how long (in
             milliseconds) the driver will wait for a response after sending an
             ordinary (non-monitoring) database operation before concluding that
-            a network error has occurred. Defaults to ``None`` (no timeout).
+            a network error has occurred. ``0`` or ``None`` means no timeout.
+            Defaults to ``None`` (no timeout).
           - `connectTimeoutMS`: (integer or None) Controls how long (in
             milliseconds) the driver will wait during server monitoring when
             connecting a new socket to a server before concluding the server
-            is unavailable. Defaults to ``20000`` (20 seconds).
+            is unavailable. ``0`` or ``None`` means no timeout.
+            Defaults to ``20000`` (20 seconds).
           - `server_selector`: (callable or None) Optional, user-provided
             function that augments server selection rules. The function should
             accept as an argument a list of
@@ -631,7 +633,7 @@ def __init__(
                 # Determine connection timeout from kwargs.
                 timeout = keyword_opts.get("connecttimeoutms")
                 if timeout is not None:
-                    timeout = common.validate_timeout_or_none(
+                    timeout = common.validate_timeout_or_none_or_zero(
                         keyword_opts.cased_key("connecttimeoutms"), timeout)
                 res = uri_parser.parse_uri(
                     entity, port, validate=True, warn=True, normalize=False,

test/discovery_and_monitoring_integration/connectTimeoutMS.json

@@ -0,0 +1,148 @@
+{
+  "runOn": [
+    {
+      "minServerVersion": "4.4"
+    }
+  ],
+  "database_name": "sdam-tests",
+  "collection_name": "connectTimeoutMS",
+  "data": [],
+  "tests": [
+    {
+      "description": "connectTimeoutMS=0",
+      "clientOptions": {
+        "retryWrites": false,
+        "connectTimeoutMS": 0,
+        "heartbeatFrequencyMS": 500,
+        "appname": "connectTimeoutMS=0"
+      },
+      "operations": [
+        {
+          "name": "insertMany",
+          "object": "collection",
+          "arguments": {
+            "documents": [
+              {
+                "_id": 1
+              },
+              {
+                "_id": 2
+              }
+            ]
+          }
+        },
+        {
+          "name": "configureFailPoint",
+          "object": "testRunner",
+          "arguments": {
+            "failPoint": {
+              "configureFailPoint": "failCommand",
+              "mode": {
+                "times": 2
+              },
+              "data": {
+                "failCommands": [
+                  "isMaster"
+                ],
+                "appName": "connectTimeoutMS=0",
+                "blockConnection": true,
+                "blockTimeMS": 550
+              }
+            }
+          }
+        },
+        {
+          "name": "wait",
+          "object": "testRunner",
+          "arguments": {
+            "ms": 750
+          }
+        },
+        {
+          "name": "insertMany",
+          "object": "collection",
+          "arguments": {
+            "documents": [
+              {
+                "_id": 3
+              },
+              {
+                "_id": 4
+              }
+            ]
+          }
+        },
+        {
+          "name": "assertEventCount",
+          "object": "testRunner",
+          "arguments": {
+            "event": "ServerMarkedUnknownEvent",
+            "count": 0
+          }
+        },
+        {
+          "name": "assertEventCount",
+          "object": "testRunner",
+          "arguments": {
+            "event": "PoolClearedEvent",
+            "count": 0
+          }
+        }
+      ],
+      "expectations": [
+        {
+          "command_started_event": {
+            "command": {
+              "insert": "connectTimeoutMS",
+              "documents": [
+                {
+                  "_id": 1
+                },
+                {
+                  "_id": 2
+                }
+              ]
+            },
+            "command_name": "insert",
+            "database_name": "sdam-tests"
+          }
+        },
+        {
+          "command_started_event": {
+            "command": {
+              "insert": "connectTimeoutMS",
+              "documents": [
+                {
+                  "_id": 3
+                },
+                {
+                  "_id": 4
+                }
+              ]
+            },
+            "command_name": "insert",
+            "database_name": "sdam-tests"
+          }
+        }
+      ],
+      "outcome": {
+        "collection": {
+          "data": [
+            {
+              "_id": 1
+            },
+            {
+              "_id": 2
+            },
+            {
+              "_id": 3
+            },
+            {
+              "_id": 4
+            }
+          ]
+        }
+      }
+    }
+  ]
+}

test/discovery_and_monitoring_integration/rediscover-quickly-after-step-down.json

@@ -64,8 +64,8 @@
           "command_name": "replSetStepDown",
           "arguments": {
             "command": {
-              "replSetStepDown": 20,
-              "secondaryCatchUpPeriodSecs": 20,
+              "replSetStepDown": 30,
+              "secondaryCatchUpPeriodSecs": 30,
               "force": false
             }
           }
@@ -74,7 +74,7 @@
           "name": "waitForPrimaryChange",
           "object": "testRunner",
           "arguments": {
-            "timeoutMS": 5000
+            "timeoutMS": 15000
           }
         },
         {

test/test_client.py

@@ -134,6 +134,24 @@ def test_keyword_arg_defaults(self):
         self.assertEqual(ReadPreference.PRIMARY, client.read_preference)
         self.assertAlmostEqual(12, client.server_selection_timeout)
 
+    def test_connect_timeout(self):
+        client = MongoClient(connect=False, connectTimeoutMS=None,
+                             socketTimeoutMS=None)
+        pool_opts = client._MongoClient__options.pool_options
+        self.assertEqual(None, pool_opts.socket_timeout)
+        self.assertEqual(None, pool_opts.connect_timeout)
+        client = MongoClient(connect=False, connectTimeoutMS=0,
+                             socketTimeoutMS=0)
+        pool_opts = client._MongoClient__options.pool_options
+        self.assertEqual(None, pool_opts.socket_timeout)
+        self.assertEqual(None, pool_opts.connect_timeout)
+        client = MongoClient(
+            'mongodb://localhost/?connectTimeoutMS=0&socketTimeoutMS=0',
+            connect=False)
+        pool_opts = client._MongoClient__options.pool_options
+        self.assertEqual(None, pool_opts.socket_timeout)
+        self.assertEqual(None, pool_opts.connect_timeout)
+
     def test_types(self):
         self.assertRaises(TypeError, MongoClient, 1)
         self.assertRaises(TypeError, MongoClient, 1.14)
@@ -996,8 +1014,8 @@ def test_socket_timeout_ms_validation(self):
         c = connected(rs_or_single_client(socketTimeoutMS=None))
         self.assertEqual(None, get_pool(c).opts.socket_timeout)
 
-        self.assertRaises(ValueError,
-                          rs_or_single_client, socketTimeoutMS=0)
+        c = connected(rs_or_single_client(socketTimeoutMS=0))
+        self.assertEqual(None, get_pool(c).opts.socket_timeout)
 
         self.assertRaises(ValueError,
                           rs_or_single_client, socketTimeoutMS=-1)

test/test_session.py

@@ -1249,6 +1249,7 @@ def test_cluster_time(self):
         # Prevent heartbeats from updating $clusterTime between operations.
         client = rs_or_single_client(event_listeners=[listener],
                                      heartbeatFrequencyMS=999999)
+        self.addCleanup(client.close)
         collection = client.pymongo_test.collection
         # Prepare for tests of find() and aggregate().
         collection.insert_many([{} for _ in range(10)])

test/test_streaming_protocol.py

@@ -26,6 +26,7 @@
                   unittest)
 from test.utils import (HeartbeatEventListener,
                         rs_or_single_client,
+                        single_client,
                         ServerEventListener,
                         wait_until)
 
@@ -160,14 +161,6 @@ def hb_started(event):
             return (isinstance(event, monitoring.ServerHeartbeatStartedEvent)
                     and event.connection_id == address)
 
-        def hb_succeeded(event):
-            return (isinstance(event, monitoring.ServerHeartbeatSucceededEvent)
-                    and event.connection_id == address)
-
-        def hb_failed(event):
-            return (isinstance(event, monitoring.ServerHeartbeatFailedEvent)
-                    and event.connection_id == address)
-
         hb_started_events = hb_listener.matching(hb_started)
         # Explanation of the expected heartbeat events:
         # Time: event
@@ -186,13 +179,50 @@ def hb_failed(event):
         # This can be reduced to ~15 after SERVER-49220 is fixed.
         self.assertLess(len(hb_started_events), 40)
 
-        # Check the awaited flag.
+    @client_context.require_failCommand_appName
+    def test_heartbeat_awaited_flag(self):
+        hb_listener = HeartbeatEventListener()
+        client = single_client(
+            event_listeners=[hb_listener], heartbeatFrequencyMS=500,
+            appName='heartbeatEventAwaitedFlag')
+        self.addCleanup(client.close)
+        # Force a connection.
+        client.admin.command('ping')
+
+        def hb_succeeded(event):
+            return isinstance(event, monitoring.ServerHeartbeatSucceededEvent)
+
+        def hb_failed(event):
+            return isinstance(event, monitoring.ServerHeartbeatFailedEvent)
+
+        fail_heartbeat = {
+            'mode': {'times': 2},
+            'data': {
+                'failCommands': ['isMaster'],
+                'closeConnection': True,
+                'appName': 'heartbeatEventAwaitedFlag',
+            },
+        }
+        with self.fail_point(fail_heartbeat):
+            wait_until(lambda: hb_listener.matching(hb_failed),
+                       "published failed event")
+        # Reconnect.
+        client.admin.command('ping')
+
         hb_succeeded_events = hb_listener.matching(hb_succeeded)
         hb_failed_events = hb_listener.matching(hb_failed)
         self.assertFalse(hb_succeeded_events[0].awaited)
-        self.assertTrue(hb_succeeded_events[1].awaited)
         self.assertTrue(hb_failed_events[0].awaited)
-        self.assertFalse(hb_failed_events[1].awaited)
+        # Depending on thread scheduling, the failed heartbeat could occur on
+        # the second or third check.
+        events = [type(e) for e in hb_listener.results[:4]]
+        if events == [monitoring.ServerHeartbeatStartedEvent,
+                      monitoring.ServerHeartbeatSucceededEvent,
+                      monitoring.ServerHeartbeatStartedEvent,
+                      monitoring.ServerHeartbeatFailedEvent]:
+            self.assertFalse(hb_succeeded_events[1].awaited)
+        else:
+            self.assertTrue(hb_succeeded_events[1].awaited)
 
 
 if __name__ == "__main__":