Use pytest-asyncio

Author: Marco Paolini

tests/conftest.py

@@ -1,79 +1,19 @@
-import asyncio
-import gc
 from io import StringIO
 
 import pytest
 
 
-# COPIED FROM aiohttp #
-
-
-@pytest.yield_fixture
-def loop(request):
-    #loop = asyncio.get_event_loop()
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(None)
-    #asyncio.set_event_loop(loop)
-
-    yield loop
-
-    is_closed = getattr(loop, 'is_closed')
-    if is_closed is not None:
-        closed = is_closed()
-    else:
-        closed = loop._closed
-    if not closed:
-        loop.call_soon(loop.stop)
-        loop.run_forever()
-        loop.close()
-    gc.collect()
-    asyncio.set_event_loop(None)
-
-
-@pytest.mark.tryfirst
-def pytest_pycollect_makeitem(collector, name, obj):
-    if collector.funcnamefilter(name):
-        if not callable(obj):
-            return
-        item = pytest.Function(name, parent=collector)
-        if 'run_loop' in item.keywords:
-            return list(collector._genfunctions(name, obj))
-
-
-@pytest.mark.tryfirst
-def pytest_pyfunc_call(pyfuncitem):
-    """
-    Run asyncio marked test functions in an event loop instead of a normal
-    function call.
-    """
-    if 'run_loop' in pyfuncitem.keywords:
-        funcargs = pyfuncitem.funcargs
-        loop = funcargs['loop']
-        testargs = {arg: funcargs[arg]
-                    for arg in pyfuncitem._fixtureinfo.argnames}
-        loop.run_until_complete(pyfuncitem.obj(**testargs))
-        return True
-
-
-def pytest_runtest_setup(item):
-    if 'run_loop' in item.keywords and 'loop' not in item.fixturenames:
-        # inject an event loop fixture for all async tests
-        item.fixturenames.append('loop')
-
-# end copy from aiohttp #
-
-
 @pytest.yield_fixture
-def websocket_cli(loop):
+def websocket_cli(event_loop):
 
     from pushpull.websocket.client import challenge
 
     inp, out = StringIO(), StringIO()
-    yield challenge('http://localhost:8080', 'test', inp, out, loop=loop), inp, out
+    yield challenge('http://localhost:8080', 'test', inp, out, loop=event_loop), inp, out
 
 
 @pytest.yield_fixture
-def websocket_server(loop):
+def websocket_server(event_loop):
 
     # TODO
     yield None

tests/test_cli.py

@@ -5,19 +5,19 @@
 import pytest
 
 
-@pytest.mark.run_loop
-async def test_client_server_down(loop, websocket_cli):
+@pytest.mark.asyncio
+async def test_client_server_down(event_loop, websocket_cli):
     challenge, inp, out = websocket_cli
     with pytest.raises(aiohttp.errors.ClientOSError):
-        await asyncio.wait_for(challenge, timeout=1, loop=loop)
+        await asyncio.wait_for(challenge, timeout=1, loop=event_loop)
 
 
+@pytest.mark.asyncio
 @pytest.mark.skip(reason='TODO')
-@pytest.mark.run_loop
-async def test_client_server_send_one(loop, websocket_cli, websocket_server):
+async def test_client_server_send_one(event_loop, websocket_cli, websocket_server):
     challenge, inp, out = websocket_cli
     with mock.patch('pushpull.websocket.client.logger') as logger:
         inp.write('hey')
         inp.write('\n')
-        await asyncio.wait_for(challenge, timeout=1, loop=loop)
+        await asyncio.wait_for(challenge, timeout=1, loop=event_loop)
         assert logger.debug.n_calls == 1

tox.ini

@@ -2,6 +2,8 @@
 envlist = py35
 
 [testenv]
-deps=pytest
+deps =
+  pytest
+  pytest-asyncio==0.3.0
 commands=py.test tests