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))` — user56700, 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. — user56700, 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.

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

1 Answers1