unittest support: Coroutine was never awaited

[Akshita6]

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()

[Akshita6]

Any update here anyone?

[Tinche]

Hi, I can take a quick look over the weekend.

…

On Fri, Feb 9, 2018 at 12:16 AM Akshita6 ***@***.***> wrote: Any update here anyone? — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#77 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AB0h8e7h8KcgWB2-mWx5eZbGIub6_oFEks5tS4BqgaJpZM4R8JCs> .

[dgilge]

@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.

[Akshita6]

@dgilge It isn't a compiler error. It's a warning. I have pytest-asyncio installed so no issues with it. The tests run fine but I get a warning so I wanted to clarify if my implementation is right.

[Akshita6]

Thanks @Tinche. I will look forward to your reply.

[pquentin]

@Akshita6 Do you think you can continue to reduce the amount of code in order to produce a minimal, complete and verifiable example?

[Tinche]

Hi,

I tried simplifying the example to:

import asyncio
import pytest

from unittest.mock import MagicMock


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']
        ]
    }


async def get_data(data_id=None):
    await asyncio.sleep(0.1)
    try:
        res = TEST_DATA[data_id]
    except KeyError:
        raise ValueError()
    return res


class TestAsyncStatSvc(object):
    @pytest.mark.asyncio
    async def test_async_get_data(self, data_svc_fixture):

        for query_value in TEST_DATA.keys():

            execute_stub = MagicMock(return_value=TEST_DATA[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 = await data_svc_fixture.svc_fixture(data_id=query_value)
            assert result == TEST_DATA[query_value]

        with pytest.raises(ValueError):
            await data_svc_fixture.svc_fixture('Unobtained')


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(event_loop):
    ctx = Context(event_loop, get_data)
    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()

but this doesn't print any warnings. Could you please take the example and add to it until the problem manifests?

[d21d3q]

Same for me. It just stopped to work at some moment, but it seems that it doesn't work with TestCase, but works with stand alone function.
Here is my setup:
test_file.py:

import asyncio
import pytest
from unittest import TestCase

class Test(TestCase):
    @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

setup:

virtualenv venv -p python3.6
source venv...
pip install pytest pytest-asyncio
pytest
# and warning here

[asvetlov]

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.

[nicoddemus]

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.

[mordajon]

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.

[shaoran]

I also have the same warning (using the example on the README.md file), but this only happens if I install pytest-asyncio with conda. Installing it with pip works.

$ conda install pytest-asyncio  # also tested with -c conda-forge
$ pytest --markers | grep asyncio
$ conda uninstall pytest-asyncio -y
$ pip install pytest-asyncio
$ pytest --markers | grep asyncio
@pytest.mark.asyncio: mark the test as a coroutine, it will be run using an asyncio event loop

The strange thing is that when I installed it with conda:

$ pip freeze | grep pytest-asyncio
$ conda list | grep pytest-asyncio
pytest-asyncio            0.10.0          py37hc8dfbb8_1001    conda-forge

and when I install it with pip

$ pip freeze | grep pytest-asyncio
pytest-asyncio==0.10.0

so it seems to be a conda problem.

[Tinche]

The way pytest plugins register themselves with pytest is by using entrypoints: https://github.com/pytest-dev/pytest-asyncio/blob/master/setup.py#L53

So maybe this is the problem?

[mlaradji]

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.