Как запустить Uvicorn + FastAPI в фоновом режиме при тестировании с PyTest

Question

У меня есть приложение REST-API, написанное с помощью Uvicorn+ FastAPI

Который я хочу проверить с помощью PyTest.

Я хочу запустить сервер в приборе при запуске тестов, поэтому, когда тест завершится, прибор убьет приложение.

FastAPI Testing показывает, как тестировать приложение 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

Как бы вы запустили приложение Uvicorn+FastAPI из PyTest, чтобы оно шло вверх и вниз с тестами?

Answer 1

На основе ответа @Gabriel C. Полностью объектно-ориентированный и асинхронный подход (с использованием превосходной среды asynctest).

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

Answer 2

Если вы хотите запустить сервер, вам придется сделать это в другом процессе / потоке, поскольку uvicorn.run() является блокирующим вызовом.

Тогда вместо использования TestClient вам придется использовать что-то вроде запросов для попадания по фактическому URL-адресу, который слушает ваш сервер.

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

Answer 3

Здесь у меня есть другое решение, которое запускает 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

Answer 4

В FastAPI есть тестовый клиент https://fastapi.tiangolo.com/tutorial/testing/ . Это своего рода реализация библиотеки запросов, поэтому вы можете использовать ее следующим образом:

      import unittest
from fastapi.testclient import TestClient
from engine.routes.base import app


class PostTest(unittest.TestCase):
    def setUp(self) -> None:
        self.client = TestClient(app)

    def test_home_page(self):
        response = self.client.get("/")
        assert response.status_code == 200

Answer 5

Углубившись в документацию, я наткнулся на https://fastapi.tiangolo.com/advanced/testing-events , где предлагается использоватьwith TestClient(app) as clientчтобы сделать асинхронные события для и запустить:

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

Это позволило мне правильно воспроизвести поведение приложения gunicorn также в pytest, не запуская никаких дополнительных фоновых процессов.

Дополнительная информация

Я только что столкнулся с этой проблемой, когда пытался адаптировать практику тестирования, предложенную в https://fastapi.tiangolo.com/tutorial/testing , т. е. без запуска фонового сервера, как предлагали другие.

Однако это не выполняет асинхронные события для@app.on_event("startup")и@app.on_event("shutdown"), которые правильно выполняются наgunicorn.

Например, следующее не будет напечатаноBlock reached

      from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    # This will not be reached in pytest, but in gunicorn it will
    print("Block reached")

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

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