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

У меня есть приложение 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, чтобы оно двигалось вверх и вниз вместе с тестами?


person RaamEE    schedule 08.08.2019    source источник


Ответы (3)


На основе ответа @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"})
person Constantin De La Roche    schedule 06.09.2019
comment
какие дополнительные значения дает asynctest модульное тестирование? Я понимаю, что это может быть важно для сквозного тестирования, нагрузочного теста и т. Д., Но для модульного теста я этого не понял. - person Baskaya; 30.05.2020
comment
asynctest - это среда тестирования выше unittest, удобная для тестирования сопрограмм. Unittest смог проверить только функции синхронизации, но, возможно, с тех пор он изменился. - person Constantin De La Roche; 31.05.2020
comment
Я не думаю, что вам нужен набор тестов async для тестирования функций async FastAPI. Вот руководство: fastapi.tiangolo.com/tutorial/testing. - person Baskaya; 01.06.2020
comment
Здесь у меня есть другое решение, которое раскручивает сервер в тот же процесс и плавно завершает работу. - person erny; 21.10.2020
comment
С моей точки зрения новичка, если вы введете новые элементы, которые не требуются OP, такие как асинхронный подход, будет оценено объяснение того, почему это требуется или какие преимущества он имеет, если он не требуется. Это также поможет людям, которые не могут заставить его работать без этого подхода, узнать, является ли это требованием или нет и почему. - person Btc Sources; 07.03.2021

Если вы хотите запустить сервер, вам придется сделать это в другом процессе / потоке, поскольку 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"}
person Gabriel Cappelli    schedule 06.09.2019
comment
Это не работает с pytest ›= 4.0, так как он больше не поддерживает yield - person M.Winkens; 26.05.2020
comment
Я только что протестировал pytest 4.0.0 и 5.4.2, и yield все еще работает. В документации даже говорится, что вам следует использовать этот подход. - person Gabriel Cappelli; 26.05.2020
comment
Здесь в документации говорится, что yield-test устарел. . В моем случае я не мог запустить его с yield. Сервер не остановился без этого - person M.Winkens; 26.05.2020
comment
С вашим вкладом я запустил его с fixture(scope="module") и yield proc (вместо просто yield). Большое спасибо! - person M.Winkens; 26.05.2020
comment
@ M.Winkens, вы говорите о yields внутри тестовых функций, которые устарели. В этом примере yield находится в фикстуре, что вовсе не является устаревшим. Вот и все: docs.pytest.org/en/2.8.7/ yieldfixture.html # yieldfixture - person Baskaya; 30.05.2020

Здесь у меня есть другое решение, которое запускает 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
person erny    schedule 21.10.2020
comment
У меня не получилось. Сервер просто завис и не ответил. Пытался дозвониться до хоста в браузере: та же история. Пришлось убить его с помощью kill -9. Python 3.7.5 - person GlaIZier; 22.10.2020
comment
@GlaIZier, ваш тестовый код тоже должен быть асинхронным. Вы использовали запросы? Вам следует использовать aiohttp. Если вы хотите, чтобы сервер отвечал, все должно быть неблокирующим. - person erny; 23.10.2020
comment
Помогло! Большое спасибо! - person GlaIZier; 16.12.2020
comment
Это отлично сработало для меня! - person Michal K; 21.01.2021