uvloop-test.py

import os
import asyncio

if int(os.environ.get("USE_UVLOOP", 0)):
    print("with uvloop")
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

import aiohttp


async def fetch(session, allow_redirects, url):
    with aiohttp.Timeout(10):
        async with session.get(url, allow_redirects=allow_redirects) as response:  # noqa
            return await response.text()


async def main_with_proxy(loop, with_proxy, allow_redirects, url):
    conn = None
    if with_proxy:
        conn = aiohttp.ProxyConnector(proxy="http://127.0.0.1:1090", loop=loop)
    async with aiohttp.ClientSession(connector=conn, loop=loop) as session:
        body = await fetch(session, allow_redirects, url)


async def test(loop):

    TO_TEST = (
        # https
        {
            'with_proxy': True,
            'allow_redirects': True,
            'url': 'https://python.org',
        },
        {
            'with_proxy': True,
            'allow_redirects': False,
            'url': 'https://python.org',
        },
        {
            'with_proxy': False,
            'allow_redirects': True,
            'url': 'https://python.org',
        },
        {
            'with_proxy': False,
            'allow_redirects': False,
            'url': 'https://python.org',
        },

        # http -> https
        {
            'with_proxy': True,
            'allow_redirects': True,
            'url': 'http://python.org',
        },
        {
            'with_proxy': True,
            'allow_redirects': False,
            'url': 'http://python.org',
        },
        {
            'with_proxy': False,
            'allow_redirects': True,
            'url': 'http://python.org',
        },
        {
            'with_proxy': False,
            'allow_redirects': False,
            'url': 'http://python.org',
        },
    )

    for params in TO_TEST:
        try:
            await main_with_proxy(loop, **params)
            params.update({
                'result': 'OK',
            })
        except Exception as e:
            params.update({
                'result': type(e).__name__,
            })

    for test in TO_TEST:
        print()
        print('-' * 10)
        print('RESULT: ', test.pop('result'))
        for k, v in test.items():
            print('{}: {}'.format(k, v))


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test(loop))