Luego de investigar con expertos en esta materia, programadores de diversas áreas y profesores hemos dado con la respuesta al problema y la dejamos plasmada en esta publicación.
Solución:
Inspirado en la respuesta de @Gabriel C. Un enfoque totalmente orientado a objetos y asincrónico (utilizando el excelente marco de prueba asincrónica).
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")
Si desea activar el servidor, tendrá que hacerlo en un proceso / hilo diferente, ya que uvicorn.run () es una llamada de bloqueo.
Luego, en lugar de usar TestClient, tendrá que usar algo como solicitudes para acceder a la URL real que su servidor está escuchando.
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"
Aquí tengo otra solución que ejecuta uvicorn en el mismo proceso (probado con 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
Sección de Reseñas y Valoraciones
Te invitamos a añadir valor a nuestro contenido informacional aportando tu experiencia en las notas.