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'

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.