16

我有一个用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 应用程序，所以它会随着测试而上升和下降？

4

3 回答 3

21

灵感来自@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"})
于 2019-09-06T23:29:18.140 回答
12

如果要启动服务器，则必须在不同的进程/线程中进行，因为 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"}
于 2019-09-06T06:12:22.420 回答
4

在这里，我有另一个在同一进程中运行 uvicorn 的解决方案（使用 Python 3.7.9 测试）：

from typing import List, Optional
import asyncio

import pytest

import uvicorn

PORT = 8000


class UvicornTestServer(uvicorn.Server):
    """Uvicorn test server

    Usage:
        @pytest.fixture
        server = UvicornTestServer()
        await server.up()
        yield
        await server.down()
    """

    def __init__(self, app, host='127.0.0.1', port=PORT):
        """Create a Uvicorn test server

        Args:
            app (FastAPI, optional): the FastAPI app. Defaults to main.app.
            host (str, optional): the host ip. Defaults to '127.0.0.1'.
            port (int, optional): the port. Defaults to PORT.
        """
        self._startup_done = asyncio.Event()
        super().__init__(config=uvicorn.Config(app, host=host, port=port))

    async def startup(self, sockets: Optional[List] = None) -> None:
        """Override uvicorn startup"""
        await super().startup(sockets=sockets)
        self.config.setup_event_loop()
        self._startup_done.set()

    async def up(self) -> None:
        """Start up server asynchronously"""
        self._serve_task = asyncio.create_task(self.serve())
        await self._startup_done.wait()

    async def down(self) -> None:
        """Shut down server asynchronously"""
        self.should_exit = True
        await self._serve_task


@pytest.fixture
async def startup_and_shutdown_server():
    """Start server as test fixture and tear down after test"""
    server = UvicornTestServer()
    await server.up()
    yield
    await server.down()


@pytest.mark.asyncio
async def test_chat_simple(startup_and_shutdown_server):
    """A simple websocket test"""
    # any test code here
于 2020-10-21T00:03:43.707 回答