add pytest_plugins

aiohttp/pytest_plugins.py

@@ -0,0 +1,67 @@
+import asyncio
+import contextlib
+
+import pytest
+
+from .test_utils import (TestClient, loop_context, setup_test_loop,
+                         teardown_test_loop)
+
+
+@contextlib.contextmanager
+def _passthrough_loop_context(loop):
+    if loop:
+        # loop already exists, pass it straight through
+        yield loop
+    else:
+        # this shadows loop_context's standard behavior
+        loop = setup_test_loop()
+        yield loop
+        teardown_test_loop(loop)
+
+
+def pytest_pycollect_makeitem(collector, name, obj):
+    """
+    Fix pytest collecting for coroutines.
+    """
+    if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
+        return list(collector._genfunctions(name, obj))
+
+
+def pytest_pyfunc_call(pyfuncitem):
+    """
+    Run coroutines in an event loop instead of a normal function call.
+    """
+    if asyncio.iscoroutinefunction(pyfuncitem.function):
+        existing_loop = pyfuncitem.funcargs.get('loop', None)
+        with _passthrough_loop_context(existing_loop) as _loop:
+            testargs = {arg: pyfuncitem.funcargs[arg]
+                        for arg in pyfuncitem._fixtureinfo.argnames}
+
+            task = _loop.create_task(pyfuncitem.obj(**testargs))
+            _loop.run_until_complete(task)
+
+        return True
+
+
+@pytest.yield_fixture
+def loop():
+    with loop_context() as _loop:
+        yield _loop
+
+
+@pytest.yield_fixture
+def test_client(loop):
+    client = None
+
+    @asyncio.coroutine
+    def _create_from_app_factory(app_factory, *args, **kwargs):
+        nonlocal client
+        app = app_factory(loop, *args, **kwargs)
+        client = TestClient(app)
+        yield from client.start_server()
+        return client
+
+    yield _create_from_app_factory
+
+    if client:
+        client.close()

aiohttp/test_utils.py

@@ -333,21 +333,26 @@ class TestClient:
     TestClient can also be used as a contextmanager, returning
     the instance of itself instantiated.
     """
+    _address = '127.0.0.1'
 
     def __init__(self, app, protocol="http"):
         self.app = app
         self._loop = loop = app.loop
         self.port = unused_port()
-        self._handler = handler = app.make_handler()
-        self._server = loop.run_until_complete(loop.create_server(
-            handler, '127.0.0.1', self.port
-        ))
+        self._handler = app.make_handler()
+        self._server = None
+        if not loop.is_running():
+            loop.run_until_complete(self.start_server())
         self._session = ClientSession(loop=self._loop)
-        self._root = "{}://127.0.0.1:{}".format(
-            protocol, self.port
-        )
+        self._root = '{}://{}:{}'.format(protocol, self._address, self.port)
         self._closed = False
 
+    @asyncio.coroutine
+    def start_server(self):
+        self._server = yield from self._loop.create_server(
+            self._handler, self._address, self.port
+        )
+
     @property
     def session(self):
         """a raw handler to the aiohttp.ClientSession.  unlike the methods on

docs/testing.rst

@@ -8,8 +8,81 @@ Testing
 Testing aiohttp web servers
 ---------------------------
 
-aiohttp provides test framework agnostic utilities for web
-servers. An example would be::
+aiohttp provides plugins for pytest_ making writing web
+server tests extremely easy, it also provides
+:ref:`test framework agnostic utilities <framework-agnostic-utilities>` for
+testing with other frameworks such as :ref:`unittest <unittest-example>`.
+
+Pytest example
+~~~~~~~~~~~~~~
+
+The :data:`test_client` fixture available from :data:`aiohttp.pytest_plugins`
+allows you to create a client to make requests to test your app.
+
+A simple would be::
+
+    from aiohttp import web
+    pytest_plugins = 'aiohttp.pytest_plugins'
+
+    async def hello(request):
+        return web.Response(body=b'Hello, world')
+
+    def create_app(loop):
+        app = web.Application(loop=loop)
+        app.router.add_route('GET', '/', hello)
+        return app
+
+    async def test_hello(test_client):
+        client = await test_client(create_app)
+        resp = await client.get('/')
+        assert resp.status == 200
+        text = await resp.text()
+        assert 'Hello, world' in text
+
+
+It also provides access to the app instance allowing tests to check the state
+of the app. Tests can be made even more succinct with a fixture to create an
+app test client::
+
+    import pytest
+    from aiohttp import web
+    pytest_plugins = 'aiohttp.pytest_plugins'
+
+
+    async def previous(request):
+        if request.method == 'POST':
+            request.app['value'] = (await request.post())['value']
+            return web.Response(body=b'thanks for the data')
+        return web.Response(body='value: {}'.format(request.app['value']).encode())
+
+    def create_app(loop):
+        app = web.Application(loop=loop)
+        app.router.add_route('*', '/', previous)
+        return app
+
+    @pytest.fixture
+    def cli(loop, test_client):
+        return loop.run_until_complete(test_client(create_app))
+
+    async def test_set_value(cli):
+        resp = await cli.post('/', data={'value': 'foo'})
+        assert resp.status == 200
+        assert await resp.text() == 'thanks for the data'
+        assert cli.app['value'] == 'foo'
+
+    async def test_get_value(cli):
+        cli.app['value'] = 'bar'
+        resp = await cli.get('/')
+        assert resp.status == 200
+        assert await resp.text() == 'value: bar'
+
+
+.. _framework-agnostic-utilities:
+
+Framework agnostic utilities
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+High level test creation::
 
     from aiohttp.test_utils import TestClient, loop_context
     from aiohttp import request
@@ -65,43 +138,7 @@ common operations such as ws_connect, get, post, etc.
 
 Please see the full api at the :class:`TestClass api reference <aiohttp.test_utils.TestClient>`
 
-
-
-Pytest example
-~~~~~~~~~~~~~~
-
-A pytest example could look like::
-
-    # assuming you are using pytest-asyncio
-    from aiohttp.test_utils import TestClient, loop_context
-
-    @pytest.yield_fixture
-    def loop():
-        with loop_context() as loop:
-            yield loop
-
-    @pytest.fixture
-    def app(loop):
-        return create_app(loop)
-
-
-    @pytest.yield_fixture
-    def test_client(app):
-        client = TestClient(app)
-        yield client
-        client.close()
-
-    def test_get_route(loop, test_client):
-        @asyncio.coroutine
-        def test_get_route():
-            nonlocal test_client
-            resp = yield from test_client.request("GET", "/")
-            assert resp.status == 200
-            text = yield from resp.text()
-            assert "Hello, world" in text
-
-        loop.run_until_complete(test_get_route())
-
+.. _unittest-example:
 
 Unittest example
 ~~~~~~~~~~~~~~~~
@@ -146,6 +183,7 @@ functionality, the AioHTTPTestCase is provided::
 aiohttp.test_utils
 ------------------
 
+.. _pytest: http://pytest.org/latest/
 .. automodule:: aiohttp.test_utils
    :members: TestClient, AioHTTPTestCase, run_loop, loop_context, setup_test_loop, teardown_test_loop
    :undoc-members:

requirements-ci.txt

@@ -8,7 +8,8 @@ cython
 chardet
 tox
 sphinxcontrib-newsfeed
-pytest
+pytest>=2.9.2
+py>=1.4.31
 pytest-cov
 gunicorn
 pygments>=2.1

tests/test_pytest_plugins.py

@@ -0,0 +1,99 @@
+pytest_plugins = 'pytester'
+
+
+def test_myplugin(testdir):
+    testdir.makepyfile("""\
+import asyncio
+import pytest
+from aiohttp import web
+
+pytest_plugins = 'aiohttp.pytest_plugins'
+
+
+@asyncio.coroutine
+def hello(request):
+    return web.Response(body=b'Hello, world')
+
+
+def create_app(loop):
+    app = web.Application(loop=loop)
+    app.router.add_route('GET', '/', hello)
+    return app
+
+
+@asyncio.coroutine
+def test_hello(test_client):
+    client = yield from test_client(create_app)
+    resp = yield from client.get('/')
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert 'Hello, world' in text
+
+
+@asyncio.coroutine
+def test_hello_with_loop(test_client, loop):
+    client = yield from test_client(create_app)
+    resp = yield from client.get('/')
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert 'Hello, world' in text
+
+
+@asyncio.coroutine
+def test_hello_fails(test_client):
+    client = yield from test_client(create_app)
+    resp = yield from client.get('/')
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert 'Hello, wield' in text
+
+
+@asyncio.coroutine
+def test_noop():
+    pass
+
+
+@asyncio.coroutine
+def previous(request):
+    if request.method == 'POST':
+        request.app['value'] = (yield from request.post())['value']
+        return web.Response(body=b'thanks for the data')
+    else:
+        v = request.app.get('value', 'unknown')
+        return web.Response(body='value: {}'.format(v).encode())
+
+
+def create_stateful_app(loop):
+    app = web.Application(loop=loop)
+    app.router.add_route('*', '/', previous)
+    return app
+
+
+@pytest.fixture
+def cli(loop, test_client):
+    return loop.run_until_complete(test_client(create_stateful_app))
+
+
+@asyncio.coroutine
+def test_set_value(cli):
+    resp = yield from cli.post('/', data={'value': 'foo'})
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert text == 'thanks for the data'
+    assert cli.app['value'] == 'foo'
+
+
+@asyncio.coroutine
+def test_get_value(cli):
+    resp = yield from cli.get('/')
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert text == 'value: unknown'
+    cli.app['value'] = 'bar'
+    resp = yield from cli.get('/')
+    assert resp.status == 200
+    text = yield from resp.text()
+    assert text == 'value: bar'
+""")
+    result = testdir.runpytest()
+    result.assert_outcomes(passed=5, failed=1)