How to start a Uvicorn + FastAPI in background when testing with PyTest

RaamEE picture RaamEE · Aug 8, 2019 · Viewed 9k times · Source

I have an REST-API app written with Uvicorn+FastAPI

Which I want to test using PyTest.

I want to start the server in a fixture when I start the tests, so when the test complete, the fixture will kill the app.

FastAPI Testing shows how to test the API app,

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

This doesn't bring the server online in the usual way. It seems that the specific functionality that is triggered by the client.get command is the only thing that runs.

I found these additional resources, but I can't make them work for me:

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

How to run server as fixture for py.test

How would you run the Uvicorn+FastAPI app from PyTest, so it goes up and down with the tests?

Answer

Constantin De La Roche picture Constantin De La Roche · Sep 7, 2019

Inspired from @Gabriel C answer. A fully object oriented and async approach (using the excellent asynctest framework).

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