pytest-asyncio: unittest support: Coroutine was never awaited

I am trying to create a test case for the below async function get_data. I get Runtime warning as RuntimeWarning: coroutine 'TestAsyncStatSvc.test_async_get_data' was never awaited testfunction(**testargs)

Below is my code. Please guide on this one. AFAIK, we get this exception when there is no event loop so I created one in the fixture. What is there that I am missing?

async def get_data(self, data_id=None):
          sql = """
              SELECT id, description
              FROM data_table
          """
          sql_params = []
          if data_id:
              sql += """
              WHERE id = $1
              """
              sql_params += [data_id]
          result = await self.dbproxy_async.execute_query_async(sql, sql_params)

          if result.empty and data_id:
              raise NotFound('data %s not found.' % data_id)
          return result.to_json()

Below Is the test case:

  class TestAsyncStatSvc(object):
      
          TEST_DATA = {
                  'Key1': ['Key1', 'Data desc for key1'],
                  'Key2': ['Key2', 'Data desc for key2'],
                  None: [
                      ['Key1', 'Data desc for key1'],
                      ['Key2', 'Data desc for key2']
                  ]
              }
          
          @pytest.mark.asyncio
          async def test_async_get_data(self, data_svc_fixture):
                  for query_value in TESTDATA.keys:

                      execute_stub = MagicMock(return_value=self.TESTDATA[query_value])

                      # Wrap the stub in a coroutine (so it can be awaited)
                      execute_coro = asyncio.coroutine(execute_stub)

                      # Stub the database db_proxy
                      db_proxy = MagicMock()
                      db_proxy.execute_query_async = execute_coro
          
                      result = data_svc_fixture.get_data(data_id=query_value)            
                      assert result == self.TESTDATA[query_value]
                      
                      if query_value is not None:
                          assert len(comm) == 1
                      else:
                          assert len(comm) == 2
          
                  with pytest.raises(NotFound):
                      data_svc_fixture.get_data('Unobtained')

And here are the fixtures:

        class Context:
            def __init__(self, loop, svc_fixture):
                self.loop = loop
                self.svc_fixture = svc_fixture
        
        
        @pytest.yield_fixture(scope="function")
        def data_svc_fixture(db_proxy_fixture, get_service_fixture, my_svc_configurator_fixture, event_loop):
            ctx = Context(event_loop, get_service_fixture('MySvc'))
            yield ctx
            event_loop.run_until_complete(asyncio.gather(*asyncio.Task.all_tasks(event_loop), return_exceptions=True))
    
        @pytest.yield_fixture()
        def event_loop():
              loop = asyncio.new_event_loop()
              asyncio.set_event_loop(loop)
              yield loop
              loop.close()

About this issue

  • Original URL
  • State: open
  • Created 6 years ago
  • Reactions: 1
  • Comments: 19 (6 by maintainers)

Commits related to this issue

Most upvoted comments

@mlaradji Got recently the same trouble and solved it like you with aiounittest. But when using aiounittest you do not need the decorator @pytest.mark.asyncio and the module pytest-asyncio anymore. I think pytest-asyncio does not work with classes derived from TestCase at all, just like the previous discussion indicate. Looking forward to when you guys solve this issue 😃

Correct me if I’m wrong, but pytest markers make no sense if applied to unittest.TestCase derived tests.

Actually marks are applied to unittest.TestCase subclasses because their methods are translated into pytest items. This is easy to demonstrate:

from unittest import TestCase
import pytest

class T(TestCase):

    def test_ok(self):
        pass

    @pytest.mark.skip
    def test_fail(self):
        assert 0
λ pytest foo.py -q
s.                                                                                                               [100%]
1 passed, 1 skipped in 0.03 seconds

Now if @pytest.mark.asyncio can work with unittest.TestCase subclasses is another matter because pytest uses a unittest runner to run the test function; not sure how that plays with pytest-asyncio.

Correct me if I’m wrong, but pytest markers make no sense if applied to unittest.TestCase derived tests. pytest-asyncio just doesn’t apply own custom test runner to test_bar method because the method is written in unittest style.

@Akshita6 Just in case you didn’t check which pulgins are loaded when you run pytest… I had the same error. In my case pytest-asyncio wasn’t installed properly.

Hey @seifertm! Thank you for the reply. Yes, they are!

The first time I read the repo I didn’t notice that, when I hit this wall I re-read and noticed that. I was able to work it out using only unittest.IsolatedAsyncioTestCase.

I seem to have resolved the issue by using aiounittest (only 1 contributer unfortunately) instead of unittest:

import asyncio
import pytest
from aiounittest import AsyncTestCase

class Test(AsyncTestCase):
    @pytest.mark.asyncio
    async def test_bar(self):
        await asyncio.sleep(0)
        assert False  # will not reach here

@pytest.mark.asyncio
async def test_foo():
    await asyncio.sleep(0)
    assert True

Test output:

# pipenv run pytest test.py
=================================================================================== test session starts ====================================================================================
platform linux -- Python 3.8.3, pytest-5.4.2, py-1.8.1, pluggy-0.13.1
rootdir: /opt/polaris/api
plugins: asyncio-0.12.0
collected 2 items                                                                                                                                                                          

test.py F.                                                                                                                                                                           [100%]

========================================================================================= FAILURES =========================================================================================
______________________________________________________________________________________ Test.test_bar _______________________________________________________________________________________

self = <api.test.Test testMethod=test_bar>

    @pytest.mark.asyncio
    async def test_bar(self):
        await asyncio.sleep(0)
>       assert False  # will not reach here
E       AssertionError: assert False

test.py:9: AssertionError
================================================================================= short test summary info ==================================================================================
FAILED test.py::Test::test_bar - AssertionError: assert False
=============================================================================== 1 failed, 1 passed in 0.10s ================================================================================

I am not sure however if pytest-asyncio is working as intended.

as @asvetlov said, unittest style doesn’t make use of @pytest.fixture at all.

just try to add one and you see it doesn’t inject it.