使用 PyTest 进行测试时如何在后台启动 Uvicorn FastAPI

Question

我有一个用 Uvicorn + FastAPI 编写的 REST-API 应用程序

我想使用 PyTest 进行测试。

我想在开始测试时在夹具中启动服务器，因此当测试完成时，夹具将终止应用程序。

FastAPI 测试 展示了如何测试 API 应用程序，

 from fastapi import FastAPI
from starlette.testclient import TestClient

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

这不会以通常的方式使服务器联机。似乎由 client.get 命令触发的特定功能是唯一运行的东西。

我找到了这些额外的资源，但我无法让它们为我工作：

https://medium.com/@hmajid2301/pytest-with-background-thread-fixtures-f0dc34ee3c46

如何将服务器作为 py.test 的夹具运行

您将如何从 PyTest 运行 Uvicorn+FastAPI 应用程序，使其随着测试上下波动？

原文由 RaamEE 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 1

灵感来自@Gabriel C 的回答。完全面向对象的异步方法（使用优秀的异步测试框架）。

 import logging
from fastapi import FastAPI

class App:
    """ Core application to test. """

    def __init__(self):
        self.api = FastAPI()
        # register endpoints
        self.api.get("/")(self.read_root)
        self.api.on_event("shutdown")(self.close)

    async def close(self):
        """ Gracefull shutdown. """
        logging.warning("Shutting down the app.")

    async def read_root(self):
        """ Read the root. """
        return {"Hello": "World"}

""" Testing part."""
from multiprocessing import Process
import asynctest
import asyncio
import aiohttp
import uvicorn

class TestApp(asynctest.TestCase):
    """ Test the app class. """

    async def setUp(self):
        """ Bring server up. """
        app = App()
        self.proc = Process(target=uvicorn.run,
                            args=(app.api,),
                            kwargs={
                                "host": "127.0.0.1",
                                "port": 5000,
                                "log_level": "info"},
                            daemon=True)
        self.proc.start()
        await asyncio.sleep(0.1)  # time for the server to start

    async def tearDown(self):
        """ Shutdown the app. """
        self.proc.terminate()

    async def test_read_root(self):
        """ Fetch an endpoint from the app. """
        async with aiohttp.ClientSession() as session:
            async with session.get("http://127.0.0.1:5000/") as resp:
                data = await resp.json()
        self.assertEqual(data, {"Hello": "World"})

原文由 Constantin De La Roche 发布，翻译遵循 CC BY-SA 4.0 许可协议

Answer 2

如果你想启动服务器，你将不得不在不同的进程/线程中进行，因为 uvicorn.run() 是一个阻塞调用。

然后，您将不得不使用请求之类的东西来代替您的服务器正在侦听的实际 URL，而不是使用 TestClient。

 from multiprocessing import Process

import pytest
import requests
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

def run_server():
    uvicorn.run(app)

@pytest.fixture
def server():
    proc = Process(target=run_server, args=(), daemon=True)
    proc.start()
    yield
    proc.kill() # Cleanup after test

def test_read_main(server):
    response = requests.get("http://localhost:8000/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

原文由 Gabriel Cappelli 发布，翻译遵循 CC BY-SA 4.0 许可协议