# 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.

### 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.Server` 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.

To run these examples, you need to use –asyncio-mode=auto or add to your pytest config file:

    asyncio_mode = auto

A simple test would be:

    from aiohttp import web

    async def hello(request):
        return web.Response(text='Hello, world')

    async def test_hello(aiohttp_client):
        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

    value = web.AppKey("value", str)

    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
    async def cli(aiohttp_client):
        app = web.Application()
        app.router.add_get('/', previous)
        app.router.add_post('/', previous)
        return await 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:

pytest_aiohttp.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.

  Added in version 3.0.

  kwargs are parameters passed to

    `aiohttp.web.Application.make_handler()`

  Changed in version 3.0.

  Deprecated since version 3.2: The fixture was renamed from `test_server` to `aiohttp_server`.

pytest_aiohttp.aiohttp_client(app, server_kwargs=None, **kwargs)
pytest_aiohttp.aiohttp_client(server, **kwargs)
pytest_aiohttp.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` or `aiohttp.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` to `aiohttp_client`.

pytest_aiohttp.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.

  Added in version 3.0.

pytest_aiohttp.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` to `aiohttp_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
    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

        async def test_example(self):
            async with self.client.request("GET", "/") as resp:
                self.assertEqual(resp.status, 200)
                text = await resp.text()
            self.assertIn("Hello, world", text)

class aiohttp.test_utils.AioHTTPTestCase

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

  Derived from `unittest.IsolatedAsyncioTestCase`

  See `unittest.TestCase` and `unittest.IsolatedAsyncioTestCase` for inherited methods and behavior.

  This class additionally provides the following:

  client

    an aiohttp test client, `TestClient` instance.

  server

    an aiohttp test server, `TestServer` instance.

    Added in version 2.3.

  app

    The application returned by `get_application()` (`aiohttp.web.Application` instance).

  async get_client()

    > This async method can be overridden to return the `TestClient` object used in the test.

    > return:

    >   `TestClient` instance.

    > Added in version 2.3.

  async get_server()

    > This async method can be overridden to return the `TestServer` object used in the test.

    > return:

    >   `TestServer` instance.

    > Added in version 2.3.

  async get_application()

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

    > return:

    >   `aiohttp.web.Application` instance.

  async asyncSetUp()

    > This async method can be overridden to execute asynchronous code during the `setUp` stage of the `TestCase`:

    >     async def asyncSetUp(self):
    >         await super().asyncSetUp()
    >         await foo()

    > Added in version 2.3.

    > Changed in version 3.8: `await super().asyncSetUp()` call is required.

  async asyncTearDown()

    > This async method can be overridden to execute asynchronous code during the `tearDown` stage of the `TestCase`:

    >     async def asyncTearDown(self):
    >         await super().asyncTearDown()
    >         await foo()

    > Added in version 2.3.

    > Changed in version 3.8: `await super().asyncTearDown()` call is required.

#### 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=...)

  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 tuple(str, str)) – 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 (aiohttp.StreamWriter) – object for managing outcoming data

    - transport (asyncio.Transport) – asyncio transport instance

    - payload (aiohttp.StreamReader) – 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.

  Added in version 2.3: match_info parameter.

## Framework Agnostic Utilities

High level test creation:

    from aiohttp.test_utils import TestClient, TestServer
    from aiohttp import request

    async def test():
        app = _create_example_app()
        async with TestClient(TestServer(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

            await 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, TestServer

    async def test():
        app = _create_example_app()
        client = TestClient(TestServer(app))
        await 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

        await test_get_route()
        await client.close()

A full list of the utilities provided can be found at the `api reference`

## 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, socket_factory=get_port_socket)

  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.

      Added in version 3.0.

    - socket_factory (collections.abc.Callable[[str,int,socket.AddressFamily],socket.socket]) –

      optional

        Factory to create a socket for the server. By default creates a TCP socket and binds it to `host` and `port`.

      Added in version 3.8.

  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.Server` used for HTTP requests serving.

  server

    `asyncio.AbstractServer` used for managing accepted connections.

  socket_factory

    socket_factory used to create and bind a server socket.

    Added in version 3.8.

  async start_server(loop=None, **kwargs)
    Parameters:

      loop (asyncio.AbstractEventLoop) – the event_loop to use

    Start a test server.

  async close()

    Stop and finish executed test server.

  make_url(path)

    Return an absolute `URL` for given path.

class aiohttp.test_utils.RawTestServer(handler, *, scheme='http', host='127.0.0.1')

  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` 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 (`'127.0.0.1'`) by default.

    - port (int) –

      optional port for TCP socket, if not provided a random unused port is used.

      Added in version 3.0.

class aiohttp.test_utils.TestServer(app, *, scheme='http', host='127.0.0.1')

  Test server (derived from `BaseTestServer`) for starting `Application`.

  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.

      Added in version 3.0.

  app

    `aiohttp.web.Application` instance to run.

### Test Client

class aiohttp.test_utils.TestClient(app_or_server, *, loop=None, scheme='http', host='127.0.0.1', cookie_jar=None, **kwargs)

  A test client used for making calls to tested server.

  Parameters:

    - app_or_server –

      `BaseTestServer` instance for making client requests to it.

      In order to pass a `aiohttp.web.Application` you need to convert it first to `TestServer` first with `TestServer(app)`.

    - cookie_jar – an optional `aiohttp.CookieJar` instance, may be useful with `CookieJar(unsafe=True, treat_as_secure_origin="http://127.0.0.1")` 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.

  app

    An alias for `self.server.app`. return `None` if `self.server` is not `TestServer` instance(e.g. `RawTestServer` instance for test low-level server).

  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.

  async start_server(**kwargs)

    Start a test server.

  async close()

    Stop and finish executed test server.

  make_url(path)

    Return an absolute `URL` for given path.

  async request(method, path, *args, **kwargs)

    Routes a request to tested http server.

    The interface is identical to `aiohttp.ClientSession.request()`, except the loop kwarg is overridden by the instance used by the test server.

  async get(path, *args, **kwargs)

    Perform an HTTP GET request.

  async post(path, *args, **kwargs)

    Perform an HTTP POST request.

  async options(path, *args, **kwargs)

    Perform an HTTP OPTIONS request.

  async head(path, *args, **kwargs)

    Perform an HTTP HEAD request.

  async put(path, *args, **kwargs)

    Perform an HTTP PUT request.

  async patch(path, *args, **kwargs)

    Perform an HTTP PATCH request.

  async delete(path, *args, **kwargs)

    Perform an HTTP DELETE request.

  async ws_connect(path, *args, **kwargs)

    Initiate websocket connection.

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

### Utilities

aiohttp.test_utils.make_mocked_coro(return_value)

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

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

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

  Create and return an `asyncio.AbstractEventLoop` instance.

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

  Note

  As side effect the function changes asyncio default loop by `asyncio.set_event_loop()` call.

  Previous default loop is not restored.

  It should not be a problem for test suite: every test expects a new test loop instance anyway.

  Changed in version 3.1: The function installs a created event loop as default.

aiohttp.test_utils.teardown_test_loop(loop)

  Teardown and cleanup an event_loop created by setup_test_loop.

  Parameters:

    loop (asyncio.AbstractEventLoop) – the loop to teardown