Testing¶
Testing aiohttp web servers¶
aiohttp provides plugin for pytest making writing web server tests extremely easy, it also provides test framework agnostic utilities for testing with other frameworks such as unittest.
Before starting to write your tests, you may also be interested on reading how to write testable services that interact with the loop.
For using pytest plugin please install pytest-aiohttp library:
$ pip install pytest-aiohttp
If you don’t want to install pytest-aiohttp for some reason you may
insert pytest_plugins = 'aiohttp.pytest_plugin'
line into
conftest.py
instead for the same functionality.
Provisional Status¶
The module is a provisional.
aiohttp has a year and half period for removing deprecated API (Policy for Backward Incompatible Changes).
But for aiohttp.test_tools
the deprecation period could be reduced.
Moreover we may break backward compatibility without deprecation peroid for some very strong reason.
The Test Client and Servers¶
aiohttp test utils provides a scaffolding for testing aiohttp-based web servers.
They are consist of two parts: running test server and making HTTP requests to this server.
TestServer
runs aiohttp.web.Application
based server, RawTestServer
starts
aiohttp.web.WebServer
low level server.
For performing HTTP requests to these servers you have to create a
test client: TestClient
instance.
The client incapsulates aiohttp.ClientSession
by providing
proxy methods to the client for common operations such as
ws_connect, get, post, etc.
Pytest¶
The aiohttp_client
fixture available from pytest-aiohttp plugin
allows you to create a client to make requests to test your app.
A simple would be:
from aiohttp import web
async def hello(request):
return web.Response(text='Hello, world')
async def test_hello(aiohttp_client, loop):
app = web.Application()
app.router.add_get('/', hello)
client = await aiohttp_client(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
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('utf-8'))
@pytest.fixture
def cli(loop, aiohttp_client):
app = web.Application()
app.router.add_get('/', previous)
app.router.add_post('/', previous)
return loop.run_until_complete(aiohttp_client(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.server.app['value'] == 'foo'
async def test_get_value(cli):
cli.server.app['value'] = 'bar'
resp = await cli.get('/')
assert resp.status == 200
assert await resp.text() == 'value: bar'
Pytest tooling has the following fixtures:
-
aiohttp.test_utils.
aiohttp_server
(app, *, port=None, **kwargs)¶ A fixture factory that creates
TestServer
:async def test_f(aiohttp_server): app = web.Application() # fill route table server = await aiohttp_server(app)
The server will be destroyed on exit from test function.
- app is the
aiohttp.web.Application
used - to start server.
port optional, port the server is run at, if not provided a random unused port is used.
New in version 3.0.
- kwargs are parameters passed to
aiohttp.web.Application.make_handler()
Changed in version 3.0: The fixture was renamed from
test_server
toaiohttp_server
.- app is the
-
aiohttp.test_utils.
aiohttp_client
(app, server_kwargs=None, **kwargs)¶ -
aiohttp.test_utils.
aiohttp_client
(server, **kwargs) -
aiohttp.test_utils.
aiohttp_client
(raw_server, **kwargs) A fixture factory that creates
TestClient
for access to tested server:async def test_f(aiohttp_client): app = web.Application() # fill route table client = await aiohttp_client(app) resp = await client.get('/')
client and responses are cleaned up after test function finishing.
The fixture accepts
aiohttp.web.Application
,aiohttp.test_utils.TestServer
oraiohttp.test_utils.RawTestServer
instance.server_kwargs are parameters passed to the test server if an app is passed, else ignored.
kwargs are parameters passed to
aiohttp.test_utils.TestClient
constructor.Changed in version 3.0: The fixture was renamed from
test_client
toaiohttp_client
.
-
aiohttp.test_utils.
aiohttp_raw_server
(handler, *, port=None, **kwargs)¶ A fixture factory that creates
RawTestServer
instance from given web handler.:async def test_f(aiohttp_raw_server, aiohttp_client): async def handler(request): return web.Response(text="OK") raw_server = await aiohttp_raw_server(handler) client = await aiohttp_client(raw_server) resp = await client.get('/')
handler should be a coroutine which accepts a request and returns response, e.g.
port optional, port the server is run at, if not provided a random unused port is used.
New in version 3.0.
-
aiohttp.test_utils.
aiohttp_unused_port
¶ Function to return an unused port number for IPv4 TCP protocol:
async def test_f(aiohttp_client, aiohttp_unused_port): port = aiohttp_unused_port() app = web.Application() # fill route table client = await aiohttp_client(app, server_kwargs={'port': port}) ...
Changed in version 3.0: The fixture was renamed from
unused_port
toaiohttp_unused_port
.
Unittest¶
To test applications with the standard library’s unittest or unittest-based functionality, the AioHTTPTestCase is provided:
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from aiohttp import web
class MyAppTestCase(AioHTTPTestCase):
async def get_application(self):
"""
Override the get_app method to return your application.
"""
async def hello(request):
return web.Response(text='Hello, world')
app = web.Application()
app.router.add_get('/', hello)
return app
# the unittest_run_loop decorator can be used in tandem with
# the AioHTTPTestCase to simplify running
# tests that are asynchronous
@unittest_run_loop
async def test_example(self):
request = await self.client.request("GET", "/")
assert request.status == 200
text = await request.text()
assert "Hello, world" in text
# a vanilla example
def test_example_vanilla(self):
async def test_get_route():
url = "/"
resp = await self.client.request("GET", url)
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text
self.loop.run_until_complete(test_get_route())
-
class
aiohttp.test_utils.
AioHTTPTestCase
[source]¶ A base class to allow for unittest web applications using aiohttp.
Derived from
unittest.TestCase
Provides the following:
-
client
¶ an aiohttp test client,
TestClient
instance.
-
server
¶ an aiohttp test server,
TestServer
instance.New in version 2.3.
-
loop
¶ The event loop in which the application and server are running.
-
app
¶ The application returned by
get_app()
(aiohttp.web.Application
instance).
-
coroutine
get_client
()[source]¶ This async method can be overridden to return the
TestClient
object used in the test.Returns: TestClient
instance.New in version 2.3.
-
coroutine
get_server
()[source]¶ This async method can be overridden to return the
TestServer
object used in the test.Returns: TestServer
instance.New in version 2.3.
-
coroutine
get_application
()[source]¶ This async method should be overridden to return the
aiohttp.web.Application
object to test.Returns: aiohttp.web.Application
instance.
-
coroutine
setUpAsync
()[source]¶ This async method do nothing by default and can be overridden to execute asynchronous code during the
setUp
stage of theTestCase
.New in version 2.3.
Note
The
TestClient
’s methods are asynchronous: you have to execute function on the test client using asynchronous methods.A basic test class wraps every test method by
unittest_run_loop()
decorator:class TestA(AioHTTPTestCase): @unittest_run_loop async def test_f(self): resp = await self.client.get('/')
-
-
unittest_run_loop:
A decorator dedicated to use with asynchronous methods of an
AioHTTPTestCase
.Handles executing an asynchronous function, using the
AioHTTPTestCase.loop
of theAioHTTPTestCase
.
Faking request object¶
aiohttp provides test utility for creating fake
aiohttp.web.Request
objects:
aiohttp.test_utils.make_mocked_request()
, it could be useful in
case of simple unit tests, like handler tests, or simulate error
conditions that hard to reproduce on real server:
from aiohttp import web
from aiohttp.test_utils import make_mocked_request
def handler(request):
assert request.headers.get('token') == 'x'
return web.Response(body=b'data')
def test_handler():
req = make_mocked_request('GET', '/', headers={'token': 'x'})
resp = handler(req)
assert resp.body == b'data'
Warning
We don’t recommend to apply
make_mocked_request()
everywhere for
testing web-handler’s business object – please use test client and
real networking via ‘localhost’ as shown in examples before.
make_mocked_request()
exists only for
testing complex cases (e.g. emulating network errors) which
are extremely hard or even impossible to test by conventional
way.
-
aiohttp.test_utils.
make_mocked_request
(method, path, headers=None, *, version=HttpVersion(1, 1), closing=False, app=None, match_info=sentinel, reader=sentinel, writer=sentinel, transport=sentinel, payload=sentinel, sslcontext=None, loop=...)[source]¶ Creates mocked web.Request testing purposes.
Useful in unit tests, when spinning full web server is overkill or specific conditions and errors are hard to trigger.
Parameters: - method (str) – str, that represents HTTP method, like; GET, POST.
- path (str) – str, The URL including PATH INFO without the host or scheme
- headers (dict, multidict.CIMultiDict, list of pairs) – mapping containing the headers. Can be anything accepted by the multidict.CIMultiDict constructor.
- match_info (dict) – mapping containing the info to match with url parameters.
- version (aiohttp.protocol.HttpVersion) – namedtuple with encoded HTTP version
- closing (bool) – flag indicates that connection should be closed after response.
- app (aiohttp.web.Application) – the aiohttp.web application attached for fake request
- writer – object for managing outcoming data
- transport (asyncio.transports.Transport) – asyncio transport instance
- payload (aiohttp.streams.FlowControlStreamReader) – raw payload reader object
- sslcontext (ssl.SSLContext) – ssl.SSLContext object, for HTTPS connection
- loop (
asyncio.AbstractEventLoop
) – An event loop instance, mocked loop by default.
Returns: aiohttp.web.Request
object.New in version 2.3: match_info parameter.
Framework Agnostic Utilities¶
High level test creation:
from aiohttp.test_utils import TestClient, loop_context
from aiohttp import request
# loop_context is provided as a utility. You can use any
# asyncio.BaseEventLoop class in it's place.
with loop_context() as loop:
app = _create_example_app()
with TestClient(app, loop=loop) as client:
async def test_get_route():
nonlocal client
resp = await client.get("/")
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text
loop.run_until_complete(test_get_route())
If it’s preferred to handle the creation / teardown on a more granular basis, the TestClient object can be used directly:
from aiohttp.test_utils import TestClient
with loop_context() as loop:
app = _create_example_app()
client = TestClient(app, loop=loop)
loop.run_until_complete(client.start_server())
root = "http://127.0.0.1:{}".format(port)
async def test_get_route():
resp = await client.get("/")
assert resp.status == 200
text = await resp.text()
assert "Hello, world" in text
loop.run_until_complete(test_get_route())
loop.run_until_complete(client.close())
A full list of the utilities provided can be found at the
api reference
Writing testable services¶
Some libraries like motor, aioes and others depend on the asyncio loop for
executing the code. When running your normal program, these libraries pick
the main event loop by doing asyncio.get_event_loop
. The problem during
testing is that there is no main loop assigned because an independent
loop for each test is created without assigning it as the main one.
This raises a problem when those libraries try to find it. Luckily, the ones that are well written, allow passing the loop explicitly. Let’s have a look at the aioes client signature:
def __init__(self, endpoints, *, loop=None, **kwargs)
As you can see, there is an optional loop
kwarg. Of course, we are not
going to test directly the aioes client but our service that depends on it
will. So, if we want our AioESService
to be easily testable, we should
define it as follows:
import asyncio
from aioes import Elasticsearch
class AioESService:
def __init__(self, loop=None):
self.es = Elasticsearch(["127.0.0.1:9200"], loop=loop)
async def get_info(self):
cluster_info = await self.es.info()
print(cluster_info)
if __name__ == "__main__":
client = AioESService()
loop = asyncio.get_event_loop()
loop.run_until_complete(client.get_info())
Note that it is accepting an optional loop
kwarg. For the normal flow of
execution it won’t affect because we can still call the service without passing
the loop explicitly having a main loop available. The problem comes when you
try to do a test like:
import pytest
from main import AioESService
class TestAioESService:
async def test_get_info(self):
cluster_info = await AioESService().get_info()
assert isinstance(cluster_info, dict)
If you try to run the test, it will fail with a similar error:
...
RuntimeError: There is no current event loop in thread 'MainThread'.
If you check the stack trace, you will see aioes is complaining that there is no current event loop in the main thread. Pass explicit loop to solve it.
If you rely on code which works with implicit loops only you may try to use hackish approach from FAQ.
Testing API Reference¶
Test server¶
Runs given aiohttp.web.Application
instance on random TCP port.
After creation the server is not started yet, use
start_server()
for actual server
starting and close()
for
stopping/cleanup.
Test server usually works in conjunction with
aiohttp.test_utils.TestClient
which provides handy client methods
for accessing to the server.
-
class
aiohttp.test_utils.
BaseTestServer
(*, scheme='http', host='127.0.0.1', port=None)[source]¶ Base class for test servers.
Parameters: - scheme (str) – HTTP scheme, non-protected
"http"
by default. - host (str) – a host for TCP socket, IPv4 local host
(
'127.0.0.1'
) by default. - port (int) –
optional port for TCP socket, if not provided a random unused port is used.
New in version 3.0.
-
scheme
¶ A scheme for tested application,
'http'
for non-protected run and'https'
for TLS encrypted server.
-
host
¶ host used to start a test server.
-
port
¶ port used to start the test server.
-
handler
¶ aiohttp.web.WebServer
used for HTTP requests serving.
-
server
¶ asyncio.AbstractServer
used for managing accepted connections.
- scheme (str) – HTTP scheme, non-protected
-
class
aiohttp.test_utils.
RawTestServer
(handler, *, scheme="http", host='127.0.0.1')[source]¶ Low-level test server (derived from
BaseTestServer
).Parameters: - handler –
a coroutine for handling web requests. The handler should accept
aiohttp.web.BaseRequest
and return a response instance, e.g.StreamResponse
orResponse
.The handler could raise
HTTPException
as a signal for non-200 HTTP response. - scheme (str) – HTTP scheme, non-protected
"http"
by default. - host (str) – a host for TCP socket, IPv4 local host
(
'127.0.0.1'
) by default. - port (int) –
optional port for TCP socket, if not provided a random unused port is used.
New in version 3.0.
- handler –
-
class
aiohttp.test_utils.
TestServer
(app, *, scheme="http", host='127.0.0.1')[source]¶ Test server (derived from
BaseTestServer
) for startingApplication
.Parameters: - app –
aiohttp.web.Application
instance to run. - scheme (str) – HTTP scheme, non-protected
"http"
by default. - host (str) – a host for TCP socket, IPv4 local host
(
'127.0.0.1'
) by default. - port (int) –
optional port for TCP socket, if not provided a random unused port is used.
New in version 3.0.
-
app
¶ aiohttp.web.Application
instance to run.
- app –
Test Client¶
-
class
aiohttp.test_utils.
TestClient
(app_or_server, *, loop=None, scheme='http', host='127.0.0.1', cookie_jar=None, **kwargs)[source]¶ A test client used for making calls to tested server.
Parameters: - app_or_server –
BaseTestServer
instance for making client requests to it.If the parameter is
aiohttp.web.Application
the tool createsTestServer
implicitly for serving the application. - cookie_jar – an optional
aiohttp.CookieJar
instance, may be useful withCookieJar(unsafe=True)
option. - scheme (str) – HTTP scheme, non-protected
"http"
by default. - loop (asyncio.AbstractEventLoop) – the event_loop to use
- host (str) – a host for TCP socket, IPv4 local host
(
'127.0.0.1'
) by default.
-
scheme
¶ A scheme for tested application,
'http'
for non-protected run and'https'
for TLS encrypted server.
-
host
¶ host used to start a test server.
-
port
¶ port used to start the server
-
server
¶ BaseTestServer
test server instance used in conjunction with client.
-
session
¶ An internal
aiohttp.ClientSession
.Unlike the methods on the
TestClient
, client session requests do not automatically include the host in the url queried, and will require an absolute path to the resource.
-
coroutine
request
(method, path, *args, **kwargs)[source]¶ Routes a request to tested http server.
The interface is identical to
asyncio.ClientSession.request()
, except the loop kwarg is overridden by the instance used by the test server.
-
coroutine
ws_connect
(path, *args, **kwargs)[source]¶ Initiate websocket connection.
The api corresponds to
aiohttp.ClientSession.ws_connect()
.
- app_or_server –
Utilities¶
-
aiohttp.test_utils.
make_mocked_coro
(return_value)[source]¶ Creates a coroutine mock.
Behaves like a coroutine which returns return_value. But it is also a mock object, you might test it as usual
Mock
:mocked = make_mocked_coro(1) assert 1 == await mocked(1, 2) mocked.assert_called_with(1, 2)
Parameters: return_value – A value that the the mock object will return when called. Returns: A mock object that behaves as a coroutine which returns return_value when called.
-
aiohttp.test_utils.
unused_port
()[source]¶ Return an unused port number for IPv4 TCP protocol.
Return int: ephemeral port number which could be reused by test server.
-
aiohttp.test_utils.
loop_context
(loop_factory=<function asyncio.new_event_loop>)[source]¶ A contextmanager that creates an event_loop, for test purposes.
Handles the creation and cleanup of a test loop.