diff --git a/aiohttp/pytest_plugins.py b/aiohttp/pytest_plugins.py new file mode 100644 index 00000000000..5d9a8f5c378 --- /dev/null +++ b/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() diff --git a/aiohttp/test_utils.py b/aiohttp/test_utils.py index a3a5aa2d610..eff99f0d721 100644 --- a/aiohttp/test_utils.py +++ b/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 diff --git a/docs/testing.rst b/docs/testing.rst index d62dd309882..0b0bf20230a 100644 --- a/docs/testing.rst +++ b/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 ` for +testing with other frameworks such as :ref:`unittest `. + +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 ` - - -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: diff --git a/requirements-ci.txt b/requirements-ci.txt index d3d294b3f27..9a581de9c58 100644 --- a/requirements-ci.txt +++ b/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 diff --git a/tests/test_pytest_plugins.py b/tests/test_pytest_plugins.py new file mode 100644 index 00000000000..3cb821369b0 --- /dev/null +++ b/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)