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 notifications@github.com 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 https://github.com/pytest-dev/pytest-asyncio/issues/77#issuecomment-364281822, 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.

[tpiekarski]

@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 :)

[andrea-cassioli-maersk]

I got the same issue and I noticed that

• the same test can work nicely if I run it standalone, or get a warning if I run it as part of a larger test suite
• If more than one warning is reported, they are all from the same unit file
• the function marked as not awaited is always some from unittest

[jmg-duarte]

Hey all, what's the status on this?

[seifertm]

As of v0.18.3 unittest is unsupported:

Test classes subclassing the standard unittest library are not supported, users are recommended to use unitest.IsolatedAsyncioTestCase or an async framework such as asynctest.

At the moment I don't think it makes sense to replicate functionality in pytest-asyncio that already exists in the standard library. However, I'm always interested in concrete use cases that show otherwise.

@jmg-duarte I'm reaching out directly to you since you're the most recent commenter: Are your asyncio testing needs fulfilled by unittest.IsolatedAsyncioTestCase?

[jmg-duarte]

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.