Merge branch 'master' of github.com:mongodb/mongo-python-driver

Author: blink1073

pymongo/asynchronous/mongo_client.py

@@ -1195,7 +1195,8 @@ def __del__(self) -> None:
                     ResourceWarning,
                     stacklevel=2,
                 )
-        except AttributeError:
+        except (AttributeError, TypeError):
+            # Ignore errors at interpreter exit.
             pass
 
     def _close_cursor_soon(

pymongo/asynchronous/pool.py

@@ -1249,6 +1249,9 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
         async with self.lock:
             conn_id = self.next_connection_id
             self.next_connection_id += 1
+            # Use a temporary context so that interrupt_connections can cancel creating the socket.
+            tmp_context = _CancellationContext()
+            self.active_contexts.add(tmp_context)
 
         listeners = self.opts._event_listeners
         if self.enabled_for_cmap:
@@ -1267,6 +1270,8 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
         try:
             sock = await _configured_socket(self.address, self.opts)
         except BaseException as error:
+            async with self.lock:
+                self.active_contexts.discard(tmp_context)
             if self.enabled_for_cmap:
                 assert listeners is not None
                 listeners.publish_connection_closed(
@@ -1292,6 +1297,9 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
         conn = AsyncConnection(sock, self, self.address, conn_id)  # type: ignore[arg-type]
         async with self.lock:
             self.active_contexts.add(conn.cancel_context)
+            self.active_contexts.discard(tmp_context)
+        if tmp_context.cancelled:
+            conn.cancel_context.cancel()
         try:
             if self.handshake:
                 await conn.hello()
@@ -1301,6 +1309,8 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
 
             await conn.authenticate()
         except BaseException:
+            async with self.lock:
+                self.active_contexts.discard(conn.cancel_context)
             conn.close_conn(ConnectionClosedReason.ERROR)
             raise
 

pymongo/message.py

@@ -252,6 +252,10 @@ def _gen_find_command(
         if limit < 0:
             cmd["singleBatch"] = True
     if batch_size:
+        # When limit and batchSize are equal we increase batchSize by 1 to
+        # avoid an unnecessary killCursors.
+        if limit == batch_size:
+            batch_size += 1
         cmd["batchSize"] = batch_size
     if read_concern.level and not (session and session.in_transaction):
         cmd["readConcern"] = read_concern.document

pymongo/pool_options.py

@@ -70,13 +70,14 @@
         "version": platform.mac_ver()[0],
     }
 elif sys.platform == "win32":
+    _ver = sys.getwindowsversion()
     _METADATA["os"] = {
-        "type": platform.system(),
-        # "Windows XP", "Windows 7", "Windows 10", etc.
-        "name": " ".join((platform.system(), platform.release())),
-        "architecture": platform.machine(),
-        # Windows patch level (e.g. 5.1.2600-SP3)
-        "version": "-".join(platform.win32_ver()[1:3]),
+        "type": "Windows",
+        "name": "Windows",
+        # Avoid using platform calls, see PYTHON-4455.
+        "architecture": os.environ.get("PROCESSOR_ARCHITECTURE") or platform.machine(),
+        # Windows patch level (e.g. 10.0.17763-SP0).
+        "version": ".".join(map(str, _ver[:3])) + f"-SP{_ver[-1] or '0'}",
     }
 elif sys.platform.startswith("java"):
     _name, _ver, _arch = platform.java_ver()[-1]

pymongo/synchronous/mongo_client.py

@@ -1193,7 +1193,8 @@ def __del__(self) -> None:
                     ResourceWarning,
                     stacklevel=2,
                 )
-        except AttributeError:
+        except (AttributeError, TypeError):
+            # Ignore errors at interpreter exit.
             pass
 
     def _close_cursor_soon(

pymongo/synchronous/pool.py

@@ -1243,6 +1243,9 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
         with self.lock:
             conn_id = self.next_connection_id
             self.next_connection_id += 1
+            # Use a temporary context so that interrupt_connections can cancel creating the socket.
+            tmp_context = _CancellationContext()
+            self.active_contexts.add(tmp_context)
 
         listeners = self.opts._event_listeners
         if self.enabled_for_cmap:
@@ -1261,6 +1264,8 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
         try:
             sock = _configured_socket(self.address, self.opts)
         except BaseException as error:
+            with self.lock:
+                self.active_contexts.discard(tmp_context)
             if self.enabled_for_cmap:
                 assert listeners is not None
                 listeners.publish_connection_closed(
@@ -1286,6 +1291,9 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
         conn = Connection(sock, self, self.address, conn_id)  # type: ignore[arg-type]
         with self.lock:
             self.active_contexts.add(conn.cancel_context)
+            self.active_contexts.discard(tmp_context)
+        if tmp_context.cancelled:
+            conn.cancel_context.cancel()
         try:
             if self.handshake:
                 conn.hello()
@@ -1295,6 +1303,8 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
 
             conn.authenticate()
         except BaseException:
+            with self.lock:
+                self.active_contexts.discard(conn.cancel_context)
             conn.close_conn(ConnectionClosedReason.ERROR)
             raise
 

requirements/typing.txt

@@ -1,5 +1,5 @@
 mypy==1.13.0
-pyright==1.1.388
+pyright==1.1.389
 typing_extensions
 -r ./encryption.txt
 -r ./ocsp.txt

test/asynchronous/test_retryable_reads.py

@@ -174,9 +174,8 @@ async def test_retryable_reads_in_sharded_cluster_multiple_available(self):
             retryReads=True,
         )
 
-        async with self.fail_point(fail_command):
-            with self.assertRaises(AutoReconnect):
-                await client.t.t.find_one({})
+        with self.assertRaises(AutoReconnect):
+            await client.t.t.find_one({})
 
         # Disable failpoints on each mongos
         for client in mongos_clients:

test/asynchronous/test_retryable_writes.py

@@ -1,4 +1,4 @@
-# Copyright 2017 MongoDB, Inc.
+# Copyright 2017-present MongoDB, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -43,7 +43,6 @@
 from bson.int64 import Int64
 from bson.raw_bson import RawBSONDocument
 from bson.son import SON
-from pymongo.asynchronous.mongo_client import AsyncMongoClient
 from pymongo.errors import (
     AutoReconnect,
     ConnectionFailure,
@@ -226,47 +225,6 @@ async def test_supported_single_statement_no_retry(self):
                     f"{msg} sent txnNumber with {event.command_name}",
                 )
 
-    @async_client_context.require_no_standalone
-    async def test_supported_single_statement_supported_cluster(self):
-        for method, args, kwargs in retryable_single_statement_ops(self.db.retryable_write_test):
-            msg = f"{method.__name__}(*{args!r}, **{kwargs!r})"
-            self.listener.reset()
-            await method(*args, **kwargs)
-            commands_started = self.listener.started_events
-            self.assertEqual(len(self.listener.succeeded_events), 1, msg)
-            first_attempt = commands_started[0]
-            self.assertIn(
-                "lsid",
-                first_attempt.command,
-                f"{msg} sent no lsid with {first_attempt.command_name}",
-            )
-            initial_session_id = first_attempt.command["lsid"]
-            self.assertIn(
-                "txnNumber",
-                first_attempt.command,
-                f"{msg} sent no txnNumber with {first_attempt.command_name}",
-            )
-
-            # There should be no retry when the failpoint is not active.
-            if async_client_context.is_mongos or not async_client_context.test_commands_enabled:
-                self.assertEqual(len(commands_started), 1)
-                continue
-
-            initial_transaction_id = first_attempt.command["txnNumber"]
-            retry_attempt = commands_started[1]
-            self.assertIn(
-                "lsid",
-                retry_attempt.command,
-                f"{msg} sent no lsid with {first_attempt.command_name}",
-            )
-            self.assertEqual(retry_attempt.command["lsid"], initial_session_id, msg)
-            self.assertIn(
-                "txnNumber",
-                retry_attempt.command,
-                f"{msg} sent no txnNumber with {first_attempt.command_name}",
-            )
-            self.assertEqual(retry_attempt.command["txnNumber"], initial_transaction_id, msg)
-
     async def test_supported_single_statement_unsupported_cluster(self):
         if async_client_context.is_rs or async_client_context.is_mongos:
             raise SkipTest("This cluster supports retryable writes")

test/asynchronous/unified_format.py

@@ -76,6 +76,7 @@
 from pymongo.asynchronous.helpers import anext
 from pymongo.encryption_options import _HAVE_PYMONGOCRYPT
 from pymongo.errors import (
+    AutoReconnect,
     BulkWriteError,
     ClientBulkWriteException,
     ConfigurationError,
@@ -545,15 +546,6 @@ def maybe_skip_test(self, spec):
                 or "Cancel server check" in spec["description"]
             ):
                 self.skipTest("MMAPv1 does not support retryWrites=True")
-        if (
-            "AsyncDatabase-level aggregate with $out includes read preference for 5.0+ server"
-            in spec["description"]
-        ):
-            if async_client_context.version[0] == 8:
-                self.skipTest("waiting on PYTHON-4356")
-        if "Aggregate with $out includes read preference for 5.0+ server" in spec["description"]:
-            if async_client_context.version[0] == 8:
-                self.skipTest("waiting on PYTHON-4356")
         if "Client side error in command starting transaction" in spec["description"]:
             self.skipTest("Implement PYTHON-1894")
         if "timeoutMS applied to entire download" in spec["description"]:
@@ -764,9 +756,10 @@ async def kill_all_sessions(self):
         for client in clients:
             try:
                 await client.admin.command("killAllSessions", [])
-            except OperationFailure:
+            except (OperationFailure, AutoReconnect):
                 # "operation was interrupted" by killing the command's
                 # own session.
+                # On 8.0+ killAllSessions sometimes returns a network error.
                 pass
 
     async def _databaseOperation_listCollections(self, target, *args, **kwargs):