如何在异步 pytest 期间运行异步后台任务？答案

【问题标题】：How do I run an async background task during async pytest?如何在异步 pytest 期间运行异步后台任务？
【发布时间】：2022-02-11 16:31:10
【问题描述】：

我想使用 pytest 测试服务器 MyServer 和客户端 MyClient 之间的交互。服务器是异步的，并且有一个异步start() 方法。这是我的设置：

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

我需要在测试运行时启动服务器并在后台运行。这里有一个非常简单的解决方案：在运行pytest之前，只需运行一个调用run_server()的python文件，但是如果我可以在这个测试文件中启动服务器会更整洁，所以服务器最终会被破坏。

我的第一直觉是使用线程：

from threading import Thread

Thread(target=run_server).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!

添加time.sleep() 以便服务器有时间启动并没有什么不同，测试会立即运行。

我相信另一种选择是使用固定装置，即在每次测试过程中启动服务器。我尝试了以下方法：

@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

这会按预期启动服务器，但不会在后台启动：MyServer().start() 永远运行并阻止测试运行。

如何在后台启动我的服务器并在测试过程中运行它？

【问题讨论】：

我不知道您的类是如何制作的，但是您可以将服务器作为 Future 对象运行。 asyncio.create_task(MyServer().start(host=HOST, port=PORT))
你可以使用 setup/teardown : stackoverflow.com/questions/51984719/…
@DevangSanghani 将上面的 run_server 放入 setup_function 需要异步 def，并且测试在执行之前运行。
@user56700 Pytest 的事件循环仅存在于每个函数中。如果我将任务添加到该事件循环中，它会在最后添加，并且直到测试之后才开始。如果我 asyncio.wait() 那个任务，测试会立即运行。如果我在测试之外创建一个新的事件循环，则测试会被阻止并且不会运行。
@Student 你不能在事件循环中运行它吗？喜欢这个答案：stackoverflow.com/questions/26270681/… - 将循环发送到函数并包装它。

标签： python multithreading async-await pytest python-asyncio

【解决方案1】：

使用进程而不是线程。这是一个例子（不是完整的代码，只是为了说明）：

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}")
    ...

进程也会在测试结束时自行终止。

【讨论】：