Fix async ssl handshake

Author: NoahStapp

pymongo/network_layer.py

@@ -238,7 +238,9 @@ async def async_receive_data(
             raise socket.timeout("timed out")
         if read_task in done:
             return read_task.result()
-        raise _OperationCancelled("operation cancelled")
+        elif conn.cancel_context.cancelled:
+            raise _OperationCancelled("operation cancelled")
+        return read_task.result()
     finally:
         sock.settimeout(sock_timeout)
 

pymongo/pyopenssl_context.py

@@ -117,14 +117,15 @@ def __init__(
         self._is_async = is_async
 
     def _call(self, call: Callable[..., _T], *args: Any, **kwargs: Any) -> _T:
+        is_async = kwargs.pop("allow_async", True) and self._is_async
         timeout = self.gettimeout()
         if timeout:
             start = _time.monotonic()
         while True:
             try:
                 return call(*args, **kwargs)
             except BLOCKING_IO_ERRORS as exc:
-                if self._is_async:
+                if is_async:
                     raise exc
                 # Check for closed socket.
                 if self.fileno() == -1:
@@ -146,6 +147,7 @@ def _call(self, call: Callable[..., _T], *args: Any, **kwargs: Any) -> _T:
                 continue
 
     def do_handshake(self, *args: Any, **kwargs: Any) -> None:
+        kwargs["allow_async"] = False
         return self._call(super().do_handshake, *args, **kwargs)
 
     def recv(self, *args: Any, **kwargs: Any) -> bytes:

test/asynchronous/test_client.py

@@ -1703,6 +1703,7 @@ def compression_settings(client):
                 # No error
                 await client.pymongo_test.test.find_one()
 
+    @async_client_context.require_sync
     async def test_reset_during_update_pool(self):
         client = await self.async_rs_or_single_client(minPoolSize=10)
         await client.admin.command("ping")
@@ -1727,10 +1728,7 @@ async def _run(self):
                     await asyncio.sleep(0.001)
 
             def run(self):
-                if _IS_SYNC:
-                    self._run()
-                else:
-                    asyncio.run(self._run())
+                self._run()
 
         t = ResetPoolThread(pool)
         t.start()

test/test_client.py

@@ -1661,6 +1661,7 @@ def compression_settings(client):
                 # No error
                 client.pymongo_test.test.find_one()
 
+    @client_context.require_sync
     def test_reset_during_update_pool(self):
         client = self.rs_or_single_client(minPoolSize=10)
         client.admin.command("ping")
@@ -1685,10 +1686,7 @@ def _run(self):
                     time.sleep(0.001)
 
             def run(self):
-                if _IS_SYNC:
-                    self._run()
-                else:
-                    asyncio.run(self._run())
+                self._run()
 
         t = ResetPoolThread(pool)
         t.start()