No SSL no auth single tests passing

Author: NoahStapp

pymongo/network_layer.py

@@ -101,7 +101,7 @@ async def close(self):
         await self.conn[1].wait_closed()
 
     def is_closing(self):
-        self.conn[0].is_closing()
+        return self.conn[0].is_closing()
 
     @property
     def get_conn(self) -> PyMongoProtocol:
@@ -340,10 +340,11 @@ def connection_lost(self, exc):
         self._connection_lost = True
         pending = list(self._pending_messages)
         for msg in pending:
-            if exc is None:
-                msg.set_result(None)
-            else:
-                msg.set_exception(exc)
+            if not msg.done():
+                if exc is None:
+                    msg.set_result(None)
+                else:
+                    msg.set_exception(exc)
             self._done_messages.append(msg)
 
         if not self._closed.done():
@@ -374,7 +375,7 @@ def data(self):
         return self._buffer
 
     async def wait_closed(self):
-        await self._closed
+        await asyncio.wait([self._closed])
 
 
 async def async_sendall(conn: PyMongoProtocol, buf: bytes) -> None:
@@ -500,10 +501,10 @@ async def async_receive_message(
 ) -> Union[_OpReply, _OpMsg]:
     """Receive a raw BSON message or raise socket.error."""
     timeout: Optional[Union[float, int]]
+    timeout = conn.conn.gettimeout
     if _csot.get_timeout():
         deadline = _csot.get_deadline()
     else:
-        timeout = conn.conn.get_conn.gettimeout
         if timeout:
             deadline = time.monotonic() + timeout
         else:
@@ -517,24 +518,31 @@ async def async_receive_message(
     cancellation_task = create_task(_poll_cancellation(conn))
     read_task = create_task(conn.conn.get_conn.read(request_id, max_message_size, debug))
     tasks = [read_task, cancellation_task]
-    done, pending = await asyncio.wait(tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
-    for task in pending:
-        task.cancel()
-    if pending:
-        await asyncio.wait(pending)
-    if len(done) == 0:
-        raise socket.timeout("timed out")
-    if read_task in done:
-        data, op_code = read_task.result()
-
-        try:
-            unpack_reply = _UNPACK_REPLY[op_code]
-        except KeyError:
-            raise ProtocolError(
-                f"Got opcode {op_code!r} but expected {_UNPACK_REPLY.keys()!r}"
-            ) from None
-        return unpack_reply(data)
-    raise _OperationCancelled("operation cancelled")
+    try:
+        done, pending = await asyncio.wait(
+            tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED
+        )
+        for task in pending:
+            task.cancel()
+        if pending:
+            await asyncio.wait(pending)
+        if len(done) == 0:
+            raise socket.timeout("timed out")
+        if read_task in done:
+            data, op_code = read_task.result()
+            try:
+                unpack_reply = _UNPACK_REPLY[op_code]
+            except KeyError:
+                raise ProtocolError(
+                    f"Got opcode {op_code!r} but expected {_UNPACK_REPLY.keys()!r}"
+                ) from None
+            return unpack_reply(data)
+        raise _OperationCancelled("operation cancelled")
+    except asyncio.CancelledError:
+        for task in tasks:
+            task.cancel()
+        await asyncio.wait(tasks)
+        raise
 
 
 def receive_message(

test/asynchronous/test_connection_monitoring.py

@@ -280,7 +280,7 @@ async def cleanup():
             for t in self.targets.values():
                 await t.join(5)
             for conn in self.labels.values():
-                conn.close_conn(None)
+                await conn.close_conn(None)
 
         self.addAsyncCleanup(cleanup)
 

test/asynchronous/test_pooling.py

@@ -124,7 +124,7 @@ async def run_mongo_thread(self):
         self.state = "connection"
 
     def __del__(self):
-        if self.sock:
+        if _IS_SYNC and self.sock:
             self.sock.close_conn(None)
 
 
@@ -215,7 +215,7 @@ async def test_pool_removes_closed_socket(self):
 
         async with cx_pool.checkout() as conn:
             # Use Connection's API to close the socket.
-            conn.close_conn(None)
+            await conn.close_conn(None)
 
         self.assertEqual(0, len(cx_pool.conns))
 
@@ -228,7 +228,7 @@ async def test_pool_removes_dead_socket(self):
         async with cx_pool.checkout() as conn:
             # Simulate a closed socket without telling the Connection it's
             # closed.
-            conn.conn.close()
+            await conn.conn.close()
             self.assertTrue(conn.conn_closed())
 
         async with cx_pool.checkout() as new_connection:
@@ -302,7 +302,7 @@ async def test_pool_check(self):
         async with cx_pool.checkout() as conn:
             # Simulate a closed socket without telling the Connection it's
             # closed.
-            conn.conn.close()
+            await conn.conn.close()
 
         # Swap pool's address with a bad one.
         address, cx_pool.address = cx_pool.address, ("foo.com", 1234)
@@ -373,7 +373,7 @@ async def test_checkout_more_than_max_pool_size(self):
             self.assertEqual(t.state, "get_socket")
 
         for socket_info in socks:
-            socket_info.close_conn(None)
+            await socket_info.close_conn(None)
 
     async def test_maxConnecting(self):
         client = await self.async_rs_or_single_client()

test/test_pooling.py

@@ -124,7 +124,7 @@ def run_mongo_thread(self):
         self.state = "connection"
 
     def __del__(self):
-        if self.sock:
+        if _IS_SYNC and self.sock:
             self.sock.close_conn(None)