How to start a Uvicorn + FastAPI in background when testing with PyTest
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"})
If you want to bring the server up you will have to do it in a different process/thread, since uvicorn.run() is a blocking call.
Then instead of using the TestClient you will have to use something like requests to hit the actual URL your server is listening to.
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"}
Here I have another solution which runs uvicorn in the same process (tested with 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