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

Author: blink1073

.evergreen/combine-coverage.sh

@@ -15,7 +15,7 @@ fi
 createvirtualenv "$PYTHON_BINARY" covenv
 # Keep in sync with run-tests.sh
 # coverage >=5 is needed for relative_files=true.
-pip install -q "coverage>=5,<=7.5"
+pip install -q "coverage[toml]>=5,<=7.5"
 
 pip list
 ls -la coverage/

.evergreen/config.yml

@@ -2854,7 +2854,7 @@ buildvariants:
   matrix_spec:
     platform: rhel7
     # Python 3.10+ requires OpenSSL 1.1.1+
-    python-version: ["3.8", "3.9", "pypy3.9", "pypy3.10"]
+    python-version: ["3.8", "3.9"]
     auth-ssl: "*"
   display_name: "OpenSSL 1.0.2 ${python-version} ${platform} ${auth-ssl}"
   tasks:
@@ -3153,7 +3153,7 @@ buildvariants:
 
 - name: testazurekms-variant
   display_name: "Azure KMS"
-  run_on: rhel87-small
+  run_on: debian11-small
   tasks:
     - name: testazurekms_task_group
       batchtime: 20160 # Use a batchtime of 14 days as suggested by the CSFLE test README

.evergreen/utils.sh

@@ -67,7 +67,6 @@ createvirtualenv () {
 
     export PIP_QUIET=1
     python -m pip install --upgrade pip
-    python -m pip install --upgrade hatch
 }
 
 # Usage:

pymongo/network_layer.py

@@ -18,32 +18,111 @@
 import asyncio
 import socket
 import struct
+import sys
+from asyncio import AbstractEventLoop, Future
 from typing import (
-    TYPE_CHECKING,
     Union,
 )
 
 from pymongo import ssl_support
 
-if TYPE_CHECKING:
-    from pymongo.pyopenssl_context import _sslConn
+try:
+    from ssl import SSLError, SSLSocket
+
+    _HAVE_SSL = True
+except ImportError:
+    _HAVE_SSL = False
+
+try:
+    from pymongo.pyopenssl_context import (
+        BLOCKING_IO_LOOKUP_ERROR,
+        BLOCKING_IO_READ_ERROR,
+        BLOCKING_IO_WRITE_ERROR,
+        _sslConn,
+    )
+
+    _HAVE_PYOPENSSL = True
+except ImportError:
+    _HAVE_PYOPENSSL = False
+    _sslConn = SSLSocket  # type: ignore
+    from pymongo.ssl_support import (  # type: ignore[assignment]
+        BLOCKING_IO_LOOKUP_ERROR,
+        BLOCKING_IO_READ_ERROR,
+        BLOCKING_IO_WRITE_ERROR,
+    )
 
 _UNPACK_HEADER = struct.Struct("<iiii").unpack
 _UNPACK_COMPRESSION_HEADER = struct.Struct("<iiB").unpack
 _POLL_TIMEOUT = 0.5
 # Errors raised by sockets (and TLS sockets) when in non-blocking mode.
-BLOCKING_IO_ERRORS = (BlockingIOError, *ssl_support.BLOCKING_IO_ERRORS)
+BLOCKING_IO_ERRORS = (BlockingIOError, BLOCKING_IO_LOOKUP_ERROR, *ssl_support.BLOCKING_IO_ERRORS)
 
 
-async def async_sendall(socket: Union[socket.socket, _sslConn], buf: bytes) -> None:
-    timeout = socket.gettimeout()
-    socket.settimeout(0.0)
+async def async_sendall(sock: Union[socket.socket, _sslConn], buf: bytes) -> None:
+    timeout = sock.gettimeout()
+    sock.settimeout(0.0)
     loop = asyncio.get_event_loop()
     try:
-        await asyncio.wait_for(loop.sock_sendall(socket, buf), timeout=timeout)  # type: ignore[arg-type]
+        if _HAVE_SSL and isinstance(sock, (SSLSocket, _sslConn)):
+            if sys.platform == "win32":
+                await asyncio.wait_for(_async_sendall_ssl_windows(sock, buf), timeout=timeout)
+            else:
+                await asyncio.wait_for(_async_sendall_ssl(sock, buf, loop), timeout=timeout)
+        else:
+            await asyncio.wait_for(loop.sock_sendall(sock, buf), timeout=timeout)  # type: ignore[arg-type]
     finally:
-        socket.settimeout(timeout)
+        sock.settimeout(timeout)
+
+
+async def _async_sendall_ssl(
+    sock: Union[socket.socket, _sslConn], buf: bytes, loop: AbstractEventLoop
+) -> None:
+    fd = sock.fileno()
+    sent = 0
+
+    def _is_ready(fut: Future) -> None:
+        loop.remove_writer(fd)
+        loop.remove_reader(fd)
+        if fut.done():
+            return
+        fut.set_result(None)
+
+    while sent < len(buf):
+        try:
+            sent += sock.send(buf)
+        except BLOCKING_IO_ERRORS as exc:
+            fd = sock.fileno()
+            # Check for closed socket.
+            if fd == -1:
+                raise SSLError("Underlying socket has been closed") from None
+            if isinstance(exc, BLOCKING_IO_READ_ERROR):
+                fut = loop.create_future()
+                loop.add_reader(fd, _is_ready, fut)
+                await fut
+            if isinstance(exc, BLOCKING_IO_WRITE_ERROR):
+                fut = loop.create_future()
+                loop.add_writer(fd, _is_ready, fut)
+                await fut
+            if _HAVE_PYOPENSSL and isinstance(exc, BLOCKING_IO_LOOKUP_ERROR):
+                fut = loop.create_future()
+                loop.add_reader(fd, _is_ready, fut)
+                loop.add_writer(fd, _is_ready, fut)
+                await fut
+
+
+# The default Windows asyncio event loop does not support loop.add_reader/add_writer: https://docs.python.org/3/library/asyncio-platforms.html#asyncio-platform-support
+async def _async_sendall_ssl_windows(sock: Union[socket.socket, _sslConn], buf: bytes) -> None:
+    view = memoryview(buf)
+    total_length = len(buf)
+    total_sent = 0
+    while total_sent < total_length:
+        try:
+            sent = sock.send(view[total_sent:])
+        except BLOCKING_IO_ERRORS:
+            await asyncio.sleep(0.5)
+            sent = 0
+        total_sent += sent
 
 
-def sendall(socket: Union[socket.socket, _sslConn], buf: bytes) -> None:
-    socket.sendall(buf)
+def sendall(sock: Union[socket.socket, _sslConn], buf: bytes) -> None:
+    sock.sendall(buf)

pymongo/pyopenssl_context.py

@@ -90,6 +90,9 @@ def _is_ip_address(address: Any) -> bool:
 # According to the docs for socket.send it can raise
 # WantX509LookupError and should be retried.
 BLOCKING_IO_ERRORS = (_SSL.WantReadError, _SSL.WantWriteError, _SSL.WantX509LookupError)
+BLOCKING_IO_READ_ERROR = _SSL.WantReadError
+BLOCKING_IO_WRITE_ERROR = _SSL.WantWriteError
+BLOCKING_IO_LOOKUP_ERROR = _SSL.WantX509LookupError
 
 
 def _ragged_eof(exc: BaseException) -> bool:

pymongo/ssl_context.py

@@ -30,6 +30,8 @@
 
 # Errors raised by SSL sockets when in non-blocking mode.
 BLOCKING_IO_ERRORS = (_ssl.SSLWantReadError, _ssl.SSLWantWriteError)
+BLOCKING_IO_READ_ERROR = _ssl.SSLWantReadError
+BLOCKING_IO_WRITE_ERROR = _ssl.SSLWantWriteError
 
 # Base Exception class
 SSLError = _ssl.SSLError

pymongo/ssl_support.py

@@ -53,6 +53,9 @@
     IPADDR_SAFE = True
     SSLError = _ssl.SSLError
     BLOCKING_IO_ERRORS = _ssl.BLOCKING_IO_ERRORS
+    BLOCKING_IO_READ_ERROR = _ssl.BLOCKING_IO_READ_ERROR
+    BLOCKING_IO_WRITE_ERROR = _ssl.BLOCKING_IO_WRITE_ERROR
+    BLOCKING_IO_LOOKUP_ERROR = BLOCKING_IO_READ_ERROR
 
     def get_ssl_context(
         certfile: Optional[str],

test/__init__.py

@@ -728,9 +728,9 @@ def require_failCommand_fail_point(self, func):
 
     def require_failCommand_appName(self, func):
         """Run a test only if the server supports the failCommand appName."""
-        # SERVER-47195
+        # SERVER-47195 and SERVER-49336.
         return self._require(
-            lambda: (self.test_commands_enabled and self.version >= (4, 4, -1)),
+            lambda: (self.test_commands_enabled and self.version >= (4, 4, 7)),
             "failCommand appName must be supported",
             func=func,
         )

test/asynchronous/__init__.py

@@ -730,9 +730,9 @@ def require_failCommand_fail_point(self, func):
 
     def require_failCommand_appName(self, func):
         """Run a test only if the server supports the failCommand appName."""
-        # SERVER-47195
+        # SERVER-47195 and SERVER-49336.
         return self._require(
-            lambda: (self.test_commands_enabled and self.version >= (4, 4, -1)),
+            lambda: (self.test_commands_enabled and self.version >= (4, 4, 7)),
             "failCommand appName must be supported",
             func=func,
         )

test/connection_monitoring/pool-checkin-make-available.json

@@ -22,7 +22,8 @@
     {
       "type": "ConnectionCheckedOut",
       "connectionId": 1,
-      "address": 42
+      "address": 42,
+      "duration": 42
     },
     {
       "type": "ConnectionCheckedIn",
@@ -32,7 +33,8 @@
     {
       "type": "ConnectionCheckedOut",
       "connectionId": 1,
-      "address": 42
+      "address": 42,
+      "duration": 42
     }
   ],
   "ignore": [