aiohttp Writing tests

There are several tools for testing Python code.

We show how to work with the most powerful and popular tool called pytest .

We need to install pytest itself and pytest-aiohttp plugin first:

pip install pytest
pip install pytest-aiohttp

Note

pytest-aiohttp is not compatible with another pupular plugin pytest-asyncio.

Making test asynchronous

Just use async def test_*() instead of plain def test_*(). pytest-aiohttp does all dirty work for you:

async def test_a() -> None:
    await asyncio.sleep(0.01)
    assert 1 == 2  # False

Testing server code

Let’s sligtly modify application code to accept a path to database explicitly:

async def init_app(db_path: Path) -> web.Application:
    app = web.Application()
    app["DB_PATH"] = db_path
    ...


async def init_db(app: web.Application) -> AsyncIterator[None]:
    sqlite_db = app["DB_PATH"]
    db = await aiosqlite.connect(sqlite_db)
    ...

After that we can use a separate isolated test database for running unit tests.

We need handy fixtures to initialize test DB:

@pytest.fixture
def db_path(tmp_path: Path) -> Path:
    path = tmp_path / "test_sqlite.db"
    try_make_db(path)
    return path


@pytest.fixture
async def db(db_path: Path) -> aiosqlite.Connection:
    """DB connection to access test DB directly from tests"""
    conn = await aiosqlite.connect(db_path)
    conn.row_factory = aiosqlite.Row
    yield conn
    await conn.close()

Note

Fixtures have function scope (default), that mean that every test gets a new fresh empty database.

Another fixtire instantiates our server and starts it on random local TCP port using aiohttp provided aiohttp_client fixture:

@pytest.fixture
async def client(aiohttp_client: Any, db_path: Path) -> _TestClient:
    app = await init_app(db_path)
    return await aiohttp_client(app)

Test posts list:

async def test_list_empty(client: _TestClient) -> None:
    resp = await client.get("/api")
    assert resp.status == 200, await resp.text()
    data = await resp.json()
    assert data == {"data": [], "status": "ok"}

Use db fixture to add a post before testing /api/{post_id} web-handler:

async def test_get_post(client: _TestClient, db: aiosqlite.Connection) -> None:
    async with db.execute(
        "INSERT INTO posts (title, text, owner, editor) VALUES (?, ?, ?, ?)",
        ["title", "text", "user", "user"],
    ) as cursor:
        post_id = cursor.lastrowid
    await db.commit()

    resp = await client.get(f"/api/{post_id}")
    assert resp.status == 200
    data = await resp.json()
    assert data == {
        "data": {
            "editor": "user",
            "id": "1",
            "owner": "user",
            "text": "text",
            "title": "title",
        },
        "status": "ok",
    }

Testing client with real server

Create a fixture to start a test server and Client instance:

@pytest.fixture
async def server(aiohttp_server: Any, db_path: Path) -> _TestServer:
    app = await init_app(db_path)
    return await aiohttp_server(app)


@pytest.fixture
async def client(server: _TestServer) -> Client:
    async with Client(server.make_url("/"), "test_user") as client:
        yield client

Use Client instance to test against running server:

async def test_get_post(client: Client, db: aiosqlite.Connection) -> None:
    post = await client.create("test title", "test text")

    async with db.execute(
        "SELECT title, text, owner, editor FROM posts WHERE id = ?", [post.id]
    ) as cursor:
        record = await cursor.fetchone()
        assert record["title"] == "test title"
        assert record["text"] == "test text"
        assert record["owner"] == "test_user"
        assert record["editor"] == "test_user"

Testing client with fake server

Assume we have no server code. We need to use fake server:

async def test_get_post(aiohttp_server: Any) -> None:
    async def handler(request: web.Request) -> web.Response:
        data = await request.json()
        assert data["title"] == "test title"
        assert data["text"] == "test text"
        return web.json_response(
            {
                "status": "ok",
                "data": {
                    "id": 1,
                    "title": "test title",
                    "text": "test text",
                    "owner": "test_user",
                    "editor": "test_user",
                },
            }
        )

    app = web.Application()
    app.add_routes([web.post("/api", handler)])
    server = await aiohttp_server(app)
    async with Client(server.make_url("/"), "test_user") as client:
        post = await client.create("test title", "test text")

        assert post.id == 1
        assert post.title == "test title"
        assert post.text == "test text"
        assert post.owner == "test_user"
        assert post.editor == "test_user"

Working with HTTPS

To test HTTPS proper SSL certificates are required.

Certificate is coupled with DNS, you don’t want to pay for testing certs.

There are two options: use pre-generated self-signed certificate pair of use awesome trustme library:

@pytest.fixture
def tls_certificate_authority() -> Any:
    return trustme.CA()


@pytest.fixture
def tls_certificate(tls_certificate_authority: Any) -> Any:
    return tls_certificate_authority.issue_server_cert("localhost",
                                                       "127.0.0.1",
                                                       "::1")


@pytest.fixture
def server_ssl_ctx(tls_certificate: Any) -> ssl.SSLContext:
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
    tls_certificate.configure_cert(ssl_ctx)
    return ssl_ctx


@pytest.fixture
def client_ssl_ctx(tls_certificate_authority: Any) -> ssl.SSLContext:
    ssl_ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
    tls_certificate_authority.configure_trust(ssl_ctx)
    return ssl_ctx

Pass server_ssl_ctx and client_ssl_ctx as ssl parameter to test secured TCP connection instead of plain TCP.

Client mocking

There is aioresponses third-party library:

pip install aioresponses

Usage:

from aioresponses import aioresponses

async def test_request() -> None:
    with aioresponses() as mocked:
        mocked.get('http://example.com', status=200, body='test')
        session = aiohttp.ClientSession()
        resp = await session.get('http://example.com')

        assert resp.status == 200
        assert "test" == await resp.text()

TBD (by Andrew)

• Writing tests for aiohttp client

• Writing tests for aiohttp server

• Tests using pytest and pytest-asyncio.