OiO.lk Community platform!

Oio.lk is an excellent forum for developers, providing a wide range of resources, discussions, and support for those in the developer community. Join oio.lk today to connect with like-minded professionals, share insights, and stay updated on the latest trends and technologies in the development field.
  You need to log in or register to access the solved answers to this problem.
  • You have reached the maximum number of guest views allowed
  • Please register below to remove this limitation

Writing unit tests when using aiohttp and asyncio

  • Thread starter Thread starter Amos
  • Start date Start date
A

Amos

Guest
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:

Code:
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:

Code:
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

Code:
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

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

and

Code:
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:

Code:
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

Code:
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'
<p>I am updating one of my Python packages so it is asynchronous (using <code>aiohttp</code> instead of <code>requests</code>). I am also updating my unit tests so they work with the new asynchronous version, but I'm having some trouble with this.</p>

<p>Here is a snippet from my package:</p>

<pre><code>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)
</code></pre>

<p>which all seems to be working when used so:</p>

<pre><code>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
</code></pre>

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

<pre><code>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()
</code></pre>

<p>it gives errors such as</p>

<pre><code>DeprecationWarning: The object should be created from async function loop=loop)
</code></pre>

<p>and </p>

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

<p>I've tried adding a <code>_close()</code> function to the <code>FPL</code> 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.</p>

<p>Is it possible to do this and am I simply doing something wrong, or am I better off using something like <code>asynctest</code>or <code>pytest-aiohttp</code> instead?</p>

<p>EDIT: I've also checked <code>aiohttp</code>'s documentation and found an <a href="https://docs.aiohttp.org/en/stable/testing.html#unittest" rel="noreferrer">example</a> showing how to test applications with the standard library’s unittest. Unfortunately I can't get it to work, since <code>loop</code> provided in <code>AioHTTPTestCase</code> has been deprecated since 3.5 and is throwing an error:</p>

<pre><code>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)
</code></pre>

<p>gives</p>

<pre><code>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'
</code></pre>
 

Latest posts

B
Replies
0
Views
1
Blundering Ecologist
B
Top