[TIP] Testing asyncio coroutines

Stefan Scherfke stefan at sofa-rockers.org
Tue Feb 17 08:09:13 PST 2015


Hi all,

I currently try to find a nice way to test asyncio coroutines.

TL;DR: I found a possible solution, but I’m not sure if it’s correct or
too hacky.

---

A naïve approach would be:


    # tests/test_coros.py
    import asyncio

    def test_coro():
        loop = asyncio.get_event_loop()

        @asyncio.coroutine
        def do_test():
            yield from asyncio.sleep(0.1)
            assert 0  # onoes!

        loop.run_until_complete(do_test())


I can also easily pull out getting a loop instance:


    # tests/conftest.py
    import asyncio


    @pytest.yield_fixture
    def loop():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        yield loop
        loop.close()


    # tests/test_coros.py
    def test_coro(loop):
        @asyncio.coroutine
        def do_test():
            yield from asyncio.sleep(0.1)
            assert 0  # onoes!

        loop.run_until_complete(do_test())


However, I actually want my tests to look like:

    # tests/test_coros.py
    @asyncio.coroutine
    def test_coro(loop):
        yield from asyncio.sleep(0.1)
        assert 0


I’ve come up with a (somewhat hacky?) solution:

    # tests/conftest.py
    import asyncio


    @pytest.yield_fixture
    def loop():
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        yield loop
        loop.close()


    def pytest_pycollect_makeitem(collector, name, obj):
        """Collect asyncio coroutines as normal functions, not as generators."""
        if asyncio.iscoroutinefunction(obj):
            return list(collector._genfunctions(name, obj))


    def pytest_pyfunc_call(pyfuncitem):
        """If ``pyfuncitem.obj`` is an asyncio coroutinefunction, execute it via
        the event loop instead of calling it directly."""
        testfunction = pyfuncitem.obj

        if not asyncio.iscoroutinefunction(testfunction):
            return

        # Copied from _pytest/python.py:pytest_pyfunc_call()
        funcargs = pyfuncitem.funcargs
        testargs = {}
        for arg in pyfuncitem._fixtureinfo.argnames:
            testargs[arg] = funcargs[arg]
        coro = testfunction(**testargs)  # Will no execute the test yet!

        # Run the coro in the event loop
        loop = testargs.get('loop', asyncio.get_event_loop())
        loop.run_until_complete(coro)

        return True  # TODO: What to return here?


It looks like it works normaly, that means I can use fixtutes in my
coroutine tests and if asserts fail, I get a correct report.

I’m not sure though what to return from pytest_pyfunc_call().  Also,
I’m not sure if this is the intended way to solve this.  What do you
think?  What can I do better?

Cheers,
Stefan




More information about the testing-in-python mailing list