Fix coroutine wrapper to await after exception (#7785)

[Dreamsorcerer]

This PR fixes #7117 and similar issues. Short explanation - our coroutine wrapper does not properly handle the exception, which breaks coroutine handling. As you can see, any task expects results from throw -
https://github.com/python/cpython/blob/main/Lib/asyncio/tasks.py#L303 but it seems like in aiohttp it was acidently removed by this commit stalkerg@f04ecc2#diff-f334e752b4894ef951105572ab8b195aeb8db90eb6e48b1dfbd9a01da4c854f5L842

This is repro a case without aiohttp:

import ssl
import collections

class TestCoroWrapper(collections.abc.Coroutine):
    __slots__ = ("_coro", "_resp")
    def __init__(self, coro):
        self._coro = coro

    def __getattr__(self, attr):
        return getattr(self._coro, attr)

    def send(self, arg):
        return self._coro.send(arg)

    def throw(self, arg):
        self._coro.throw(arg)

    def close(self):
        return self._coro.close()

    def __await__(self):
        ret = self._coro.__await__()
        return ret

async def ssl_call(context):
    loop = asyncio.get_event_loop()
    return await loop.create_connection(
        lambda: asyncio.Protocol(),
        '2404:6800:4004:824::2004',
        443,
        ssl=context,
        family=socket.AddressFamily.AF_INET6,
        proto=6,
        flags=socket.AddressInfo.AI_NUMERICHOST | socket.AddressInfo.AI_NUMERICSERV,
        server_hostname='www.google.com',
        local_addr=None
    )

async def prepare_call():
    context = ssl.create_default_context()
    try:
        connection = await ssl_call(context)
    except ssl.SSLError as e:
        print(f"Got exception1: {e}")
        raise e

    return connection

async def my_task():
    try:
        await prepare_call()
    except Exception as e:
        print(f"Got exception2: {e}")

    await asyncio.sleep(1)
    raise Exception("test")

async def main():
    my_coro = TestCoroWrapper(my_task())
    print(f"is coro? {asyncio.iscoroutine(my_coro)}")
    task = asyncio.create_task(my_coro)
    await task

asyncio.run(main())

The TestCoroWrapper here is equivalent of
_BaseRequestContextManager. If you run such code like: SSL_CERT_FILE=/dev/null SSL_CERT_DIR=/dev/null python test.py you will get an error: await wasn't used with future.
The main idea here is that you are trying to await the sleep function after getting and catching an exception from the native (SSL) module.

Now I should explain why repro with aiohttp for some users return the same error:

import asyncio
import aiohttp

async def main():
    async with aiohttp.ClientSession() as session:
        try:
            response = await asyncio.ensure_future(session.get('https://www.google.com/'))
            print(await response.text())
        finally:
            response.release()

asyncio.run(main())

here it's happened because in TCPConnector._create_direct_connection we are getting all IPs for the given host and trying to connect one by one. If the first connection gets an error we will catch this error and try again for the next IP. If you have IPv6 you will have at least 2 IPs here (ipv6 and ipv4), and after the first error, you will try to connect to a second IP and get the same error.

Why it's problem only for asyncio.ensure_future? Because asyncio.ensure_future creates a task such a task starts processing coroutines and directly communicates with our coroutine wrapper witch not return a result for throw.

---

Co-authored-by: Sam Bull [email protected]
Co-authored-by: Sam Bull [email protected]
(cherry picked from commit a57dc31)