How do I run an async background task during async pytest?

Question

I want to test interactions between a server MyServer and client MyClient using pytest. The server is async, and has an async start() method. Here is my setup:

import pytest
from .server import MyServer
from .client import MyClient

PORT = 8765
HOST = "127.0.0.1"

async def run_server():
    await MyServer().start(host=HOST, port=PORT)

@pytest.fixture
def client():
    return MyClient(host=HOST, port=PORT)

@pytest.mark.asyncio
async def test_can_get_response(client):
    response = await client.call_server()
    assert response is not None

I need the server to be booted up and run in the background when the tests run. There is a very simple solution here: before running pytest, just run a python file that calls run_server(), but it would be neater if I could boot up the server in this test file, so the server gets destroyed at the end.

My first instinct is to use threading:

from threading import Thread

Thread(target=run_server).start()

However, the tests get run before the thread has time to start:

Task was destroyed but it is pending!
task: <Task pending name='Task-2' coro=<run_server() running at /path/to/file> wait_for=<_GatheringFuture pending cb=[<TaskWakeupMethWrapper object at 0x105fc90a0>()]>>
Task was destroyed but it is pending!

Adding a time.sleep() so the server has time to boot doesn't make a difference, the tests run immediately.

I believe another alternative is to use a fixture, i.e. booting up the server for the course of each test. I attempted the following:

@pytest.fixture
async def run_server():
    await MyServer().start(host=HOST, port=PORT)

@pytest.fixture
def client():
    return MyClient(host=HOST, port=PORT)

@pytest.mark.asyncio
async def test_can_get_response(client, run_server):
    await run_server # Not run_server() because the fixture is already a coroutine
    response = await client.call_server()
    assert response is not None

This boots up the server as expected, but not in the background: MyServer().start() runs forever and blocks the tests from running.

How can I start my server in the background and run it over the course of the tests?

I don't know how your classes are made, but you could run your server as a Future object. `asyncio.create_task(MyServer().start(host=HOST, port=PORT))` — Cow, Feb 09 '22 at 11:15
You could use setup/teardown : https://stackoverflow.com/questions/51984719/pytest-setup-and-teardown-functions-same-as-self-written-functions — Devang Sanghani, Feb 09 '22 at 11:29
@DevangSanghani putting the above run_server in setup_function requires an async def, and the tests run before it executes. — Student, Feb 09 '22 at 11:37
@user56700 Pytest's event loop exists only within each function. If I add the task to that event loop, it gets added at the end, and doesn't start until after the test. If I asyncio.wait() that task, the tests run immediately. If I create a new event loop outside the tests, the tests are then blocked and do not run. — Student, Feb 09 '22 at 11:41
@Student cant you just run it in the event loop then? Like this answer: https://stackoverflow.com/questions/26270681/can-an-asyncio-event-loop-run-in-the-background-without-suspending-the-python-in - send loop to the function and wrap it. — Cow, Feb 09 '22 at 11:51

score 0 · Answer 1 · answered Feb 11 '22 at 08:31

Use Process instead of Thread. Here's an example (not full code, just to illustrate):

import pytest
from multiprocessing import Process

@pytest.mark.asyncio
async def test_http_file_upload():
    httpd = http_server.build_server(host, port)
    Process(target=httpd.serve_forever, args=(), daemon=True).start()
    await transfer_file(file, f"http://{host}:{port}{remote_path}")
    ...

Process will also terminate itself at the end of test.

felipe · Answer 2 · 2023-02-22T15:04:16.283

Inside of the root pytest folder add the following to conftest.py:

import asyncio

HOST = "127.0.0.1"
PORT = 8888

async def start():
    loop = asyncio.get_event_loop()

    try:
        # await server start here.
    except
        # await server stop here.
    finally:
        loop.close()
        
@pytest.fixture(scope="session")
def event_loop():
    return asyncio.get_event_loop()

@pytest.fixture(autouse=True, scope="session")
def server(event_loop):
    task = asyncio.ensure_future(start(), loop=event_loop)

    # Sleeps to allow the server boot-up.
    event_loop.run_until_complete(asyncio.sleep(1))

    try:
        yield
    finally:
        task.cancel()

The session scope ensures the same server instance is live until all tests run (as opposed to restarting on each individual test). Then in your tests, you can query the server;

import asyncio
import pytest

from .conftest import HOST, PORT

@pytest.mark.asyncio
async def test_http_request():
    # You can use anything to query the server - httpx, aiohttp, etc.
    # Using native code for the sake of example.

    reader, writer = await asyncio.open_connection(HOST, PORT)
    writer.write(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
    data = await reader.read(1024)
    assert data.startswith(b"HTTP/1.1 200 OK\r\n")
    writer.close()
    await writer.wait_closed()

The benefit of this solution is that coverage tools - like pytest-cov - will count all of the areas within your server code that is being tested. I used the above solution for a personal project, mitm, that uses asyncio.start_server.

How do I run an async background task during async pytest?

2 Answers2