PYTHON-3260 Improve test_transaction_starts_with_batched_write and test_continuous_network_errors (mongodb#945)

(cherry picked from commit a624197)

Author: ShaneHarvey

test/test_client.py

@@ -1621,7 +1621,6 @@ def test_direct_connection(self):
         with self.assertRaises(ConfigurationError):
             MongoClient(["host1", "host2"], directConnection=True)
 
-    @unittest.skipIf(sys.platform.startswith("java"), "Jython does not support gc.get_objects")
     @unittest.skipIf("PyPy" in sys.version, "PYTHON-2927 fails often on PyPy")
     def test_continuous_network_errors(self):
         def server_description_count():
@@ -1637,7 +1636,7 @@ def server_description_count():
         gc.collect()
         with client_knobs(min_heartbeat_interval=0.003):
             client = MongoClient(
-                "invalid:27017", heartbeatFrequencyMS=3, serverSelectionTimeoutMS=100
+                "invalid:27017", heartbeatFrequencyMS=3, serverSelectionTimeoutMS=150
             )
             initial_count = server_description_count()
             self.addCleanup(client.close)
@@ -1651,6 +1650,70 @@ def server_description_count():
             # On Python 3.11 we seem to get more of a delta.
             self.assertAlmostEqual(initial_count, final_count, delta=20)
 
+    @client_context.require_failCommand_fail_point
+    def test_network_error_message(self):
+        client = single_client(retryReads=False)
+        self.addCleanup(client.close)
+        client.admin.command("ping")  # connect
+        with self.fail_point(
+            {"mode": {"times": 1}, "data": {"closeConnection": True, "failCommands": ["find"]}}
+        ):
+            expected = "%s:%s: " % client.address
+            with self.assertRaisesRegex(AutoReconnect, expected):
+                client.pymongo_test.test.find_one({})
+
+    @unittest.skipIf("PyPy" in sys.version, "PYTHON-2938 could fail on PyPy")
+    def test_process_periodic_tasks(self):
+        client = rs_or_single_client()
+        coll = client.db.collection
+        coll.insert_many([{} for _ in range(5)])
+        cursor = coll.find(batch_size=2)
+        cursor.next()
+        c_id = cursor.cursor_id
+        self.assertIsNotNone(c_id)
+        client.close()
+        # Add cursor to kill cursors queue
+        del cursor
+        wait_until(
+            lambda: client._MongoClient__kill_cursors_queue,
+            "waited for cursor to be added to queue",
+        )
+        client._process_periodic_tasks()  # This must not raise or print any exceptions
+        with self.assertRaises(InvalidOperation):
+            coll.insert_many([{} for _ in range(5)])
+
+    @unittest.skipUnless(_HAVE_DNSPYTHON, "DNS-related tests need dnspython to be installed")
+    def test_service_name_from_kwargs(self):
+        client = MongoClient(
+            "mongodb+srv://user:[email protected]",
+            srvServiceName="customname",
+            connect=False,
+        )
+        self.assertEqual(client._topology_settings.srv_service_name, "customname")
+        client = MongoClient(
+            "mongodb+srv://user:[email protected]"
+            "/?srvServiceName=shouldbeoverriden",
+            srvServiceName="customname",
+            connect=False,
+        )
+        self.assertEqual(client._topology_settings.srv_service_name, "customname")
+        client = MongoClient(
+            "mongodb+srv://user:[email protected]/?srvServiceName=customname",
+            connect=False,
+        )
+        self.assertEqual(client._topology_settings.srv_service_name, "customname")
+
+    @unittest.skipUnless(_HAVE_DNSPYTHON, "DNS-related tests need dnspython to be installed")
+    def test_srv_max_hosts_kwarg(self):
+        client = MongoClient("mongodb+srv://test1.test.build.10gen.cc/")
+        self.assertGreater(len(client.topology_description.server_descriptions()), 1)
+        client = MongoClient("mongodb+srv://test1.test.build.10gen.cc/", srvmaxhosts=1)
+        self.assertEqual(len(client.topology_description.server_descriptions()), 1)
+        client = MongoClient(
+            "mongodb+srv://test1.test.build.10gen.cc/?srvMaxHosts=1", srvmaxhosts=2
+        )
+        self.assertEqual(len(client.topology_description.server_descriptions()), 2)
+
     @unittest.skipIf(_HAVE_DNSPYTHON, "dnspython must not be installed")
     def test_srv_no_dnspython_error(self):
         with self.assertRaisesRegex(ConfigurationError, 'The "dnspython" module must be'):

test/test_transactions.py

@@ -30,6 +30,9 @@
 from test.utils_spec_runner import SpecRunner
 
 from bson.py3compat import StringIO
+from bson import encode
+from bson.raw_bson import RawBSONDocument
+
 from gridfs import GridFS, GridFSBucket
 from pymongo import WriteConcern, client_session
 from pymongo.client_session import TransactionOptions
@@ -332,14 +335,14 @@ def test_transaction_starts_with_batched_write(self):
         listener.reset()
         self.addCleanup(client.close)
         self.addCleanup(coll.drop)
-        large_str = "\0" * (10 * 1024 * 1024)
-        ops = [InsertOne({"a": large_str}) for _ in range(10)]
+        large_str = "\0" * (1 * 1024 * 1024)
+        ops = [InsertOne(RawBSONDocument(encode({"a": large_str}))) for _ in range(48)]
         with client.start_session() as session:
             with session.start_transaction():
                 coll.bulk_write(ops, session=session)
         # Assert commands were constructed properly.
         self.assertEqual(
-            ["insert", "insert", "insert", "commitTransaction"], listener.started_command_names()
+            ["insert", "insert", "commitTransaction"], listener.started_command_names()
         )
         first_cmd = listener.results["started"][0].command
         self.assertTrue(first_cmd["startTransaction"])
@@ -349,7 +352,7 @@ def test_transaction_starts_with_batched_write(self):
             self.assertNotIn("startTransaction", event.command)
             self.assertEqual(lsid, event.command["lsid"])
             self.assertEqual(txn_number, event.command["txnNumber"])
-        self.assertEqual(10, coll.count_documents({}))
+        self.assertEqual(48, coll.count_documents({}))
 
 
 class PatchSessionTimeout(object):