PYTHON-2395 Resync tests and add prose test

Author: ShaneHarvey

test/server_selection/in_window/two-choices.json

@@ -12,11 +12,6 @@
         "address": "b:27017",
         "avg_rtt_ms": 35,
         "type": "Mongos"
-      },
-      {
-        "address": "c:27017",
-        "avg_rtt_ms": 7000,
-        "type": "Mongos"
       }
     ]
   },

test/test_server_selection_in_window.py

@@ -15,17 +15,18 @@
 """Test the topology module's Server Selection Spec implementation."""
 
 import os
+import threading
+
 from pymongo.common import clean_node, HEARTBEAT_FREQUENCY
-from pymongo.pool import PoolOptions
 from pymongo.read_preferences import ReadPreference
 from pymongo.settings import TopologySettings
 from pymongo.topology import Topology
-from test import unittest
+from test import client_context, IntegrationTest, unittest
 from test.utils_selection_tests import (
     get_addresses,
     get_topology_settings_dict,
     make_server_description)
-from test.utils import TestCreator
+from test.utils import TestCreator, rs_client, OvertCommandListener
 
 
 # Location of JSON test specifications.
@@ -106,5 +107,74 @@ def tests(self, scenario_def):
 
 CustomTestCreator(create_test, TestAllScenarios, TEST_PATH).create_tests()
 
+
+class FinderThread(threading.Thread):
+    def __init__(self, collection, iterations):
+        super(FinderThread, self).__init__()
+        self.daemon = True
+        self.collection = collection
+        self.iterations = iterations
+        self.passed = False
+
+    def run(self):
+        for _ in range(self.iterations):
+            self.collection.find_one({})
+        self.passed = True
+
+
+class TestProse(IntegrationTest):
+    def frequencies(self, client, listener):
+        coll = client.test.test
+        N_FINDS = 10
+        N_THREADS = 10
+        threads = [FinderThread(coll, N_FINDS) for _ in range(N_THREADS)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        for thread in threads:
+            self.assertTrue(thread.passed)
+
+        events = listener.results['started']
+        self.assertEqual(len(events), N_FINDS * N_THREADS)
+        nodes = client.nodes
+        self.assertEqual(len(nodes), 2)
+        freqs = {address: 0 for address in nodes}
+        for event in events:
+            freqs[event.connection_id] += 1
+        for address in freqs:
+            freqs[address] = freqs[address]/float(len(events))
+        return freqs
+
+    @client_context.require_failCommand_appName
+    @client_context.require_multiple_mongoses
+    def test_load_balancing(self):
+        listener = OvertCommandListener()
+        client = rs_client(client_context.mongos_seeds(),
+                           appName='loadBalancingTest',
+                           event_listeners=[listener])
+        self.addCleanup(client.close)
+        # Delay find commands on
+        delay_finds = {
+            'configureFailPoint': 'failCommand',
+            'mode': {'times': 10000},
+            'data': {
+                'failCommands': ['find'],
+                'blockConnection': True,
+                'blockTimeMS': 500,
+                'appName': 'loadBalancingTest',
+            },
+        }
+        with self.fail_point(delay_finds):
+            nodes = client_context.client.nodes
+            self.assertEqual(len(nodes), 1)
+            delayed_server = next(iter(nodes))
+            freqs = self.frequencies(client, listener)
+            self.assertLessEqual(freqs[delayed_server], 0.20)
+        listener.reset()
+        freqs = self.frequencies(client, listener)
+        self.assertAlmostEqual(freqs[delayed_server], 0.50, delta=0.05)
+
+
 if __name__ == "__main__":
     unittest.main()