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.


The test_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(test_client, loop):
    app = web.Application(loop=loop)
    app.router.add_get('/', hello)
    client = await test_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'))

def cli(loop, test_client):
    app = web.Application(loop=loop)
    app.router.add_get('/', hello)
    return loop.run_until_complete(test_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.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'

Pytest tooling has the following fixtures:

aiohttp.test_utils.test_server(app, **kwargs)

A fixture factory that creates TestServer:

async def test_f(loop, test_server):
    app = web.Application(loop=loop)
    # fill route table

    server = await test_server(app)

The server will be destroyed on exit from test function.

app is the aiohttp.web.Application used
to start server.
kwargs are parameters passed to
aiohttp.test_utils.test_client(app, **kwargs)
aiohttp.test_utils.test_client(server, **kwargs)
aiohttp.test_utils.test_client(raw_server, **kwargs)

A fixture factory that creates TestClient for access to tested server:

async def test_f(loop, test_client):
    app = web.Application(loop=loop)
    # fill route table

    client = await test_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 or aiohttp.test_utils.RawTestServer instance.

kwargs are parameters passed to aiohttp.test_utils.TestClient constructor.

aiohttp.test_utils.raw_test_server(handler, **kwargs)

A fixture factory that creates RawTestServer instance from given web handler.

handler should be a coroutine which accepts a request and returns response, e.g.:

async def test_f(raw_test_server, test_client):

    async def handler(request):
        return web.Response(text="OK")

    raw_server = await raw_test_server(handler)
    client = await test_client(raw_server)
    resp = await client.get('/')


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, loop):
        """Override the get_app method to return your application.
        # it's important to use the loop passed here.
        return web.Application(loop=loop)

    # the unittest_run_loop decorator can be used in tandem with
    # the AioHTTPTestCase to simplify running
    # tests that are asynchronous
    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(self):
        async def test_get_route():
            url = root + "/"
            resp = await self.client.request("GET", url, loop=loop)
            assert resp.status == 200
            text = await resp.text()
            assert "Hello, world" in text

class aiohttp.test_utils.AioHTTPTestCase[source]

A base class to allow for unittest web applications using aiohttp.

Derived from unittest.TestCase

Provides the following:


an aiohttp test client, TestClient instance.


The event loop in which the application and server are running.


The application returned by get_app() (aiohttp.web.Application instance).

coroutine get_application(loop)[source]

This async method should be overridden to return the aiohttp.web.Application object to test.

Parameters:loop (asyncio.AbstractEventLoop) – the event_loop to use
Returns:aiohttp.web.Application instance.

Standard test initialization method.


Standard test finalization method.


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):

    async def test_f(self):
        resp = await self.client.get('/')

A decorator dedicated to use with asynchronous methods of an AioHTTPTestCase.

Handles executing an asynchronous function, using the AioHTTPTestCase.loop of the AioHTTPTestCase.

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'


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, reader=sentinel, writer=sentinel, transport=sentinel, payload=sentinel, sslcontext=None, secure_proxy_ssl_header=None)[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.

  • 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.
  • 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
  • reader (aiohttp.parsers.StreamParser) – object for storing and managing incoming data
  • 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
  • secure_proxy_ssl_header (tuple) – A tuple representing a HTTP header/value combination that signifies a request is secure.

aiohttp.web.Request object.

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(loop)
    with TestClient(app) 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


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(loop)
    client = TestClient(app)
    root = "{}".format(port)

    async def test_get_route():
        resp = await client.get("/")
        assert resp.status == 200
        text = await resp.text()
        assert "Hello, world" in text


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([""], loop=loop)

    async def get_info(self):
        cluster_info = await self.es.info()

if __name__ == "__main__":
    client = AioESService()
    loop = asyncio.get_event_loop()

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='')[source]

Base class for test servers.

  • scheme (str) – HTTP scheme, non-protected "http" by default.
  • host (str) – a host for TCP socket, IPv4 local host ('') by default.

A scheme for tested application, 'http' for non-protected run and 'htttps' for TLS encrypted server.


host used to start a test server.


A random port used to start a server.


aiohttp.web.WebServer used for HTTP requests serving.


asyncio.AbstractServer used for managing accepted connections.

coroutine start_server(**kwargs)[source]

Start a test server.

coroutine close()[source]

Stop and finish executed test server.


Return an absolute URL for given path.

class aiohttp.test_utils.RawTestServer(handler, *, loop=None, scheme="http", host='')[source]

Low-level test server (derived from BaseTestServer).

  • handler

    a coroutine for handling web requests. The handler should accept aiohttp.web.BaseRequest and return a response instance, e.g. StreamResponse or Response.

    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 ('') by default.
class aiohttp.test_utils.TestServer(app, *, scheme="http", host='')[source]

Test server (derived from BaseTestServer) for starting Application.

  • appaiohttp.web.Application instance to run.
  • scheme (str) – HTTP scheme, non-protected "http" by default.
  • host (str) – a host for TCP socket, IPv4 local host ('') by default.

aiohttp.web.Application instance to run.

Test Client

class aiohttp.test_utils.TestClient(app_or_server, *, scheme='http', host='', cookie_jar=None, **kwargs)[source]

A test client used for making calls to tested server.

  • app_or_server

    BaseTestServer instance for making client requests to it.

    If the parameter is aiohttp.web.Application the tool creates TestServer implicitly for serving the application.

  • cookie_jar – an optional aiohttp.CookieJar instance, may be useful with CookieJar(unsafe=True) option.
  • scheme (str) – HTTP scheme, non-protected "http" by default.
  • host (str) – a host for TCP socket, IPv4 local host ('') by default.

A scheme for tested application, 'http' for non-protected run and 'htttps' for TLS encrypted server.


host used to start a test server.


A random port used to start a server.


BaseTestServer test server instance used in conjunction with client.


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 start_server(**kwargs)[source]

Start a test server.

coroutine close()[source]

Stop and finish executed test server.


Return an absolute URL for given path.

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 get(path, *args, **kwargs)[source]

Perform an HTTP GET request.

coroutine post(path, *args, **kwargs)[source]

Perform an HTTP POST request.

coroutine options(path, *args, **kwargs)[source]

Perform an HTTP OPTIONS request.

coroutine head(path, *args, **kwargs)[source]

Perform an HTTP HEAD request.

coroutine put(path, *args, **kwargs)[source]

Perform an HTTP PUT request.

coroutine patch(path, *args, **kwargs)[source]

Perform an HTTP PATCH request.

coroutine delete(path, *args, **kwargs)[source]

Perform an HTTP DELETE request.

coroutine ws_connect(path, *args, **kwargs)[source]

Initiate websocket connection.

The api corresponds to aiohttp.ClientSession.ws_connect().



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.

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.

aiohttp.test_utils.setup_test_loop(loop_factory=<function asyncio.new_event_loop>)[source]

Create and return an asyncio.AbstractEventLoop instance.

The caller should also call teardown_test_loop, once they are done with the loop.


Teardown and cleanup an event_loop created by setup_test_loop.

Parameters:loop (asyncio.AbstractEventLoop) – the loop to teardown
blog comments powered by Disqus