Writing unit tests when using aiohttp and asyncio

Question

I am updating one of my Python packages so it is asynchronous (using aiohttp instead of requests). I am also updating my unit tests so they work with the new asynchronous version, but I'm having some trouble with this.

Here is a snippet from my package:

async def fetch(session, url):
    while True:
        try:
            async with session.get(url) as response:
                assert response.status == 200
                return await response.json()
        except Exception as error:
            pass


class FPL():
    def __init__(self, session):
        self.session = session

    async def get_user(self, user_id, return_json=False):
        url = API_URLS["user"].format(user_id)
        user = await fetch(self.session, url)

        if return_json:
            return user
        return User(user, session=self.session)

which all seems to be working when used so:

async def main():
    async with aiohttp.ClientSession() as session:
         fpl = FPL(session)
         user = await fpl.get_user(3808385)
         print(user)

loop = asynio.get_event_loop()
loop.run_until_complete(main())

>>> User 3808385

Unfortunately I am having some trouble with my unit tests. I thought I could simply do something like

def _run(coroutine):
    return asyncio.get_event_loop().run_until_complete(coroutine)


class FPLTest(unittest.TestCase):
    def setUp(self):
        session = aiohttp.ClientSession()
        self.fpl = FPL(session)

    def test_user(self):
        user = _run(self.fpl.get_user("3523615"))
        self.assertIsInstance(user, User)

        user = _run(self.fpl.get_user("3523615", True))
        self.assertIsInstance(user, dict)

if __name__ == '__main__':
    unittest.main()

it gives errors such as

DeprecationWarning: The object should be created from async function loop=loop)

and

ResourceWarning: Unclosed client session <aiohttp.client.ClientSession object at 0x7fbe647fd208>

I've tried adding a _close() function to the FPL class that closes the session, and then calling this from the tests, but this also doesn't work and still says there is an unclosed client session.

Is it possible to do this and am I simply doing something wrong, or am I better off using something like asynctestor pytest-aiohttp instead?

EDIT: I've also checked aiohttp's documentation and found an example showing how to test applications with the standard library’s unittest. Unfortunately I can't get it to work, since loop provided in AioHTTPTestCase has been deprecated since 3.5 and is throwing an error:

class FPLTest(AioHTTPTestCase):
    def setUp(self):
        session = aiohttp.ClientSession()
        self.fpl = FPL(session)

    @unittest_run_loop
    async def test_user(self):
        user = await self.fpl.get_user("3523615")
        self.assertIsInstance(user, User)

        user = await self.fpl.get_user("3523615", True)
        self.assertIsInstance(user, dict)

gives

tests/test_fpl.py:20: DeprecationWarning: The object should be created from async function
  session = aiohttp.ClientSession()
  ...
======================================================================
ERROR: test_user (__main__.FPLTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/amos/Documents/fpl/venv/lib/python3.7/site-packages/aiohttp/test_utils.py", line 477, in new_func
    return self.loop.run_until_complete(
AttributeError: 'FPLTest' object has no attribute 'loop'

======================================================================
ERROR: test_user (__main__.FPLTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/amos/Documents/fpl/venv/lib/python3.7/site-packages/aiohttp/test_utils.py", line 451, in tearDown
    self.loop.run_until_complete(self.tearDownAsync())
AttributeError: 'FPLTest' object has no attribute 'loop'

Leventhal

Answer 1

Use pytest with aiohttp-pytest:

async def test_test_user(loop):
    async with aiohttp.ClientSession() as session:
         fpl = FPL(session)
         user = await fpl.get_user(3808385)
    assert isinstance(user, User)

Proverb of the modern python developer: life is too short not to use pytest.

You'll likely also want to setup a mock server to receive your http requests during tests, I don't have a trivial example, but a full working example can be seen here.

Bracci

I quickly used this example to see if it would work, and it does - thanks! I was hoping I wouldn't have to convert all my tests and could continue using unittest, but I guess I will try out pytest (never used it before). – Leventhal

You don't have to convert all your tests, only the ones which require async behaviour. pytest runs standard unittest styles tests absolutely fine. – Bracci

Also, switching from unittest to py.test is much like switching from urllib2 to requests - such an improvement that one never looks back and only regrets not doing the switch sooner! – Amasa

I agree, it's really great so far. In the end I created some fixtures which allow me to access e.g. the fpl object in my tests. It didn't even take that long to convert everything and it looks much cleaner than before. – Leventhal

Answer 2

You can use the pytest-asyncio plugin and the AsyncMock class from the standard unittest library. Make sure you installed the packages:

pip install pytest pytest-asyncio

To allow await code, decorate your test function with the @pytest.mark.asyncio marker which is coming from the plugin. Next, create a mock instance and then patch the aiohttp.ClientSession.get method that returns with an AsyncMock class:

from unittest.mock import AsyncMock, patch

import aiohttp
import pytest


# The fetch function to be tested
async def fetch_data(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as response:
        return await response.json()

@pytest.mark.asyncio
async def test_fetch_data_success():
    url = 'http://test.com'
    expected_response = {'key': 'value'}

    mock_response = AsyncMock()
    mock_response.__aenter__.return_value = mock_response
    mock_response.__aexit__.return_value = None
    mock_response.json = AsyncMock(return_value=expected_response)

    with patch('aiohttp.ClientSession.get', return_value=mock_response):
        async with aiohttp.ClientSession() as session:
            response = await fetch_data(session, url)
            assert response == expected_response

We need to mock the __aenter__ and the __aexit__ methods at the mock_response instance since we are using the async context manager. If it is not done, you may get an error like this:

session = <aiohttp.client.ClientSession object at 0x0000018C35EF0880>, url = 'http://test.com'

    async def fetch_data(session: aiohttp.ClientSession, url: str):
>       async with session.get(url) as response:
E       AttributeError: __aenter__

tests\async_fetch_test.py:8: AttributeError

Leaseback