aiormq 6.2.3 fails to clean up connection if first frame_receiver.get_frame() fails

[TBBle]

We have a test in our test suite using aio-pika 7.1.2 and hence aiormq 6.2.3, which feeds an expired SSL certificate into the connect request.

The trace for the (expected) exception is:

ERROR    handler.client_service:client_service.py:664 [SSL: SSLV3_ALERT_CERTIFICATE_EXPIRED] sslv3 alert certificate expired (_ssl.c:2633)
Traceback (most recent call last):
  File "C:\Projects\PS\services\projects\handler\src\handler\client_service.py", line 648, in serve_forever
    async with self.message_handler as message_source:
  File "C:\Projects\PS\services\projects\handler\src\handler\video_generation_message_source\rabbitmq.py", line 54, in __aenter__
    await self.__listener.connect()
  File "C:\Projects\PS\services\lib\shared\src\ps\shared\rabbitmq.py", line 95, in connect
    self.connection = await aio_pika.connect_robust(rmq_url)
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aio_pika\robust_connection.py", line 325, in connect_robust
    await connection.connect(
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aio_pika\robust_connection.py", line 133, in connect
    await super().connect(
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aio_pika\connection.py", line 135, in connect
    await self._make_connection(timeout=timeout, **kwargs)
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aio_pika\connection.py", line 114, in _make_connection
    connection: aiormq.abc.AbstractConnection = await asyncio.wait_for(
  File "C:\Program Files\Python39\lib\asyncio\tasks.py", line 442, in wait_for
    return await fut
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aiormq\connection.py", line 731, in connect
    await connection.connect(client_properties or {})
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aiormq\base.py", line 162, in wrap
    return await self.create_task(func(self, *args, **kwargs))
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aiormq\abc.py", line 40, in __inner
    return await self.task
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aiormq\connection.py", line 397, in connect
    _, _, frame = await frame_receiver.get_frame()
  File "C:\Users\paulh\AppData\Local\pypoetry\Cache\virtualenvs\handler-vLGyY0no-py3.9\lib\site-packages\aiormq\connection.py", line 143, in get_frame
    frame_header = await countdown(self.reader.readexactly(1))
  File "C:\Program Files\Python39\lib\asyncio\tasks.py", line 479, in wait_for
    return fut.result()
  File "C:\Program Files\Python39\lib\asyncio\streams.py", line 723, in readexactly
    await self._wait_for_data('readexactly')
  File "C:\Program Files\Python39\lib\asyncio\streams.py", line 517, in _wait_for_data
    await self._waiter
  File "C:\Program Files\Python39\lib\asyncio\sslproto.py", line 528, in data_received
    ssldata, appdata = self._sslpipe.feed_ssldata(data)
  File "C:\Program Files\Python39\lib\asyncio\sslproto.py", line 199, in feed_ssldata
    chunk = self._sslobj.read(self.max_size)
  File "C:\Program Files\Python39\lib\ssl.py", line 888, in read
    v = self._sslobj.read(len)
ssl.SSLError: [SSL: SSLV3_ALERT_CERTIFICATE_EXPIRED] sslv3 alert certificate expired (_ssl.c:2633)

However, observed on Python 3.9 on Windows, the exception raised at aiormq\connection.py line 397 occurs after the asyncio.open_connection call on line 380, but outside the big try-catch on lines 408-449 that ensures that any exception raised will close the sockets, and hence leaks the socket without ever putting the connection somewhere we can close it (because aio_pika has wrapped this call in a factory function, so even if the best solution was to call close even on a failed connection, aio-pika would have to do that manually (as would any other user of this library))

The relevant caller in aio-pika, for reference:

    async def _make_connection(
        self, *, timeout: TimeoutType = None, **kwargs: Any
    ) -> aiormq.abc.AbstractConnection:
        connection: aiormq.abc.AbstractConnection = await asyncio.wait_for(
            aiormq.connect(self.url, **kwargs), timeout=timeout,
        )
        connection.closing.add_done_callback(
            partial(self._on_connection_close, connection),
        )
        await connection.ready()
        return connection

The symptom for us is that a later test fails with

E               ResourceWarning: unclosed transport <asyncio.sslproto._SSLProtocolTransport object at 0x0000022DB33E7FA0>

because the leaked SSLProtocolTransport is not detected until tear-down of a later test (due to how pytest-aiohttp cleans up its loops, I suspect).

I found the trivial fix was to move all the code after the await asyncio.open_connection, i.e. lines 391 to 407 down inside the big try-catch, i.e. indented once and moved after line 408.

That way a failure in that block will still trigger the clean-up.