Parte 6: Tests en FastAPI

¡Muy buenas!, llegamos al final de esta serie 😢. Espero que os haya gustado el proyecto y los tutoriales tanto como yo he disfrutado haciéndolos 😊. Como último tutorial de esta serie, vamos a integrar tests en nuestra API con FastAPI para verificar que funciona correctamente.

Si os perdisteis los tutoriales anteriores de esta serie os los dejo aquí 👇

• Parte 1: Cómo crear una API REST COMPLETA con FastAPI, instalación y estructura
• Parte 2: Conexiones a bases de datos y creación de modelos con FastAPI
• Parte 3: Creación de modelos de Pydantic y nuestro primer usuario con FastAPI
• Parte 4: Autenticación con JWT en FastAPI
• Parte 5: Cómo crear un CRUD con FastAPI

¿Cómo vamos a plantear los tests?

Para este sistema he pensado que la mejor opción sería la de imitar lo que hace el framework Django en sus tests. Este framework crea una base de datos de prueba al inicio de los tests y la elimina al final por lo que necesitaremos que el usuario de la base de datos que usamos para autenticarnos en la API tenga permisos para crear una nueva base de datos.

Lo he planteado de esta manera porque me parece la forma más fiel de replicar todo el funcionamiento de la API, pero eso no quiere decir que haya otras formas de hacerlo como por ejemplo utilizando mockups. En mi opinión siempre deberíamos adaptarnos según el proyecto que estemos realizando y no tener siempre una idea fija de como hacer las cosas.

Cambios en el settings

Actualmente, nuestro archivo db.py el cual se encarga de la conexión a la base de datos recibe el parámetro db_name de la clase Settings que creamos en esta serie anteriormente. Pues bien, cuando estemos lanzando un test no vamos a querer que se conecte a nuestra base de datos real así que vamos a realizar una pequeña modificación en el archivo /app/v1/utils/settings.py para que se conecte a la base de datos de prueba así que lo abrimos y añadimos el siguiente código:

import os

from pydantic import BaseSettings
from dotenv import load_dotenv
load_dotenv()


class Settings(BaseSettings):

    _db_name: str = os.getenv('DB_NAME')
    db_user: str = os.getenv('DB_USER')
    db_pass: str = os.getenv('DB_PASS')
    db_host: str = os.getenv('DB_HOST')
    db_port: str = os.getenv('DB_PORT')

    secret_key: str = os.getenv('SECRET_KEY')
    token_expire: int = os.getenv('ACCESS_TOKEN_EXPIRE_MINUTES')

    @property
    def db_name(self):
        if os.getenv('RUN_ENV') == 'test':
            return 'test_' + self._db_name

        return self._db_name

Ahora la variable db_name tiene el prefijo _ por lo que será una variable protegida. Para acceder a ella he creado un getter y en él hacemos una comprobación, si la variable de entorno RUN_ENV (que definiremos más adelante) es igual a test, queremos que le añadas el prefijo "test_" al nombre de la base de datos, si no, devolverá el nombre real.

Configuración de los tests

Para ejecutar nuestros tests vamos a emplear la librería pytest así que el primer paso será instalarla en nuestro proyecto. También necesitaremos la librería requests la cual se encargará de realizar las peticiones a nuestra API cuando estemos realizando los tests así que vamos a instalar ambas:

pip install -U pytest
pip install requests

El siguiente paso es generar la carpeta que contendrá nuestros tests, para ello vamos al directorio raíz y creamos una carpeta que se llamará tests.

Dentro de la carpeta tests crearemos un archivo llamado __init__.py, ya que lo necesitaremos para la correcta importación de las clases en los tests.

Una vez hecho esto, vamos a producir el sistema que se encargará de crear y eliminar la base de datos de pruebas, para ello nos vamos a valer de las funciones pytest_sessionstart y pytest_sessionfinish. La primera se iniciará antes de realizar los tests y la segunda al finalizar todos los tests.

Estas funciones deberán ir dentro de un archivo llamado conftest.py dentro de la carpeta /tests así que nos dirigimos ahí y creamos el archivo.

Una vez creado, vamos a añadir el siguiente código:

import os
os.environ['RUN_ENV'] = 'test'

from app.v1.model import user_model, todo_model
from app.v1.utils.settings import Settings

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT

settings = Settings()

Como podéis observar, lo primero que hacemos es importar la librería os y después creamos la variable de entorno RUN_ENV con el valor test. Esta es la variable que estamos utilizando en nuestro archivo settings.py para saber si nos estamos conectando a la base de datos real o a la de pruebas.

Posteriormente, importamos los modelos de peewee de usuarios y to-dos, ya que los necesitaremos para generar las tablas.

También importamos Settings y creamos una instancia, ya que necesitamos la información de la conexión para crear y borrar la base de datos.

Y por último importamos psycopg2 que es la librería que se encarga de conectarse con PostgreSQL.

El siguiente paso es crear las funciones que se encargarán de realizar la conexión a la base de datos, crearla y eliminarla. Para ello añadimos las siguientes funciones en nuestro archivo conftest.py:

def postgresql_connection():
    con = psycopg2.connect(f"user='{settings.db_user}' password='{settings.db_pass}'")
    con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
    return con

def delete_database():

    if not settings.db_name.startswith("test_"):
        raise Exception(f'Invalid name for database = {settings.db_name}')

    sql_drop_db = f"DROP DATABASE IF EXISTS {settings.db_name}"
    con = postgresql_connection()
    cursor = con.cursor()
    cursor.execute(f"SELECT pg_terminate_backend(pg_stat_activity.pid) FROM pg_stat_activity WHERE  pg_stat_activity.datname = '{settings.db_name}' AND pid <> pg_backend_pid();")
    cursor.execute(sql_drop_db)
    con.close()

def create_database():
    sql_create_db = f"CREATE DATABASE {settings.db_name} WITH OWNER = {settings.db_user} ENCODING = 'UTF8' CONNECTION LIMIT = -1;"

    con = postgresql_connection()
    cursor = con.cursor()
    cursor.execute(sql_create_db)
    con.close()

La función postgresql_connection se encargará de efectuar y retornar la conexión a la base de datos.

Seguidamente, tenemos la función delete_database. Esta función tiene un control para verificar que el nombre de la base de datos empieza por la palabra test_, de esta forma nos aseguramos de borrar la base de datos correcta (nos puede pasar que hagamos un cambio en la clase Settings y no nos devuelva la base de datos que esperamos).

Luego creamos la conexión y lanzamos una query para eliminar las conexiones abiertas a la base de datos, ya que si se queda alguna conexión abierta no nos dejará eliminarla.

Por último lanzamos la sentencia DROP para eliminarla.

La siguiente función es create_database que se encargará de generar la base de datos.

Una vez hecho esto, vamos a añadir las funciones que se encargarán de lanzar las funciones para crear y borrar las bases de datos. Al final del archivo conftest.py añadimos el siguiente código:

def pytest_sessionstart(session):

    delete_database()
    create_database()

    from app.v1.utils.db import db

    with db:
        db.create_tables([user_model.User, todo_model.Todo])


def pytest_sessionfinish(session, exitstatus):
    delete_database()

Como comentamos anteriormente, la función pytest_sessionstart se iniciará antes de empezar el proceso de ejecución de los tests. En ella primero eliminamos la base de datos de prueba (si existe, por si se nos ha quedado colgado y no se eliminó al finalizar los tests) y después la generamos. Una vez hecho esto, creamos las tablas.

Al finalizar todos los tests, se ejecutará la función pytest_sessionfinish la cual llamará a la función delete_database para destruir la base de datos.

Testando los endpoints de usuarios

Como primer paso vamos a testar que los endpoints de creación y autenticación de usuarios funcionan correctamente. Para ello vamos a crear un archivo llamado test_user.py dentro de la carpeta tests. Pytest requiere que todos los archivos que contengan tests tendrán que comenzar por la palabra test_ y estar dentro de la carpeta tests para que este los ejecute.

Una vez hecho esto, abrimos el archivo que acabamos de crear y añadimos el siguiente código:

from fastapi.testclient import TestClient
from main import app

def test_create_user_ok():
    client = TestClient(app)

    user = {
        'email': 'test_create_user_ok@cosasdedevs.com',
        'username': 'test_create_user_ok',
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 201, response.text
    data = response.json()
    assert data['email'] == user['email']
    assert data['username'] == user['username']

Como veis, primero importamos la clase TestClient de FastAPI. Gracias a esta librería podremos realizar las peticiones a nuestra API y comprobar el resultado.

Luego importamos la instancia de FastAPI llamada app que creamos al principio de esta serie dentro del archivo main.py. La tenemos que importar, ya que TestClient la necesita como parámetro para crear el objeto con el cual podremos ejecutar las peticiones.

Por último tenemos la función test_create_user_ok con nuestro primer test. Todos los tests deberán tener el prefijo test_ para que pytest los ejecute. Si no los ignorará.

Esta función se encargará de crear un usuario y lo primero que hacemos es crear una instancia de TestClient. Después, estamos definiendo un diccionario con los datos del usuario a generar.

El siguiente paso es realizar la petición la cual guardará su resultado dentro de la variable response. Utilizamos el método post de client, ya que la creación es una petición de tipo post y como parámetro enviamos la url al endpoint de creación de usuarios y dentro del parámetro json le pasamos la variable user. Ya la librería se encargará de convertirlo automáticamente a JSON y enviarlo.

Para verificar que el resultado de la API es correcto, vamos a utilizar los assert, Estos nos permiten realizar validaciones, si no cumplen la condición, lanzará una excepción que será capturada por pytest y nos devolverá el test como erróneo.

En este caso, primero verificamos que el status_code de la respuesta es igual a 201 que es el estado que definimos si todo había ido bien y después comprobamos la respuesta que definimos en nuestra la cual retornaba un JSON con la información del usuario recién creado.

Comprobamos que coinciden con los datos del usuario que hemos enviado y listo, ya tenemos nuestro primer test.

Ahora vamos a crear dos nuevos tests que se encargarán de comprobar que se envía un error cuando intentamos crear un usuario con un username o email ya existente:

def test_create_user_duplicate_email():
    client = TestClient(app)

    user = {
        'email': 'test_create_user_duplicate_email@cosasdedevs.com',
        'username': 'test_create_user_duplicate_email',
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 201, response.text

    user['username'] = 'test_create_user_duplicate_email2'

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 400, response.text
    data = response.json()
    assert data['detail'] == 'Email already registered'


def test_create_user_duplicate_username():
    client = TestClient(app)

    user = {
        'email': 'test_create_user_duplicate_username@cosasdedevs.com',
        'username': 'test_create_user_duplicate_username',
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 201, response.text

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 400, response.text
    data = response.json()
    assert data['detail'] == 'Username already registered'

En estos tests, primero creamos el usuario y seguidamente volvemos a intentarlo con el username o email ya existente. Si nos devuelve el error 400 y el mensaje que definimos es que todo ha ido como esperábamos.

Por último vamos a testar que el login funciona correctamente. Para ello vamos a añadir el siguiente test:

def test_login():
    client = TestClient(app)

    user = {
        'email': 'test_login@cosasdedevs.com',
        'username': 'test_login',
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/user/',
        json=user,
    )
    assert response.status_code == 201, response.text

    login = {
        'username': 'test_login',
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/login/',
        data=login,
        headers={
            'Content-Type': 'application/x-www-form-urlencoded'
        },
        allow_redirects=True
    )

    assert response.status_code == 200, response.text
    data = response.json()
    assert len(data['access_token']) > 0
    assert data['token_type'] == 'bearer'

Este test crea un usuario y posteriormente realizamos una petición al endpoint de login. En él enviamos por parámetro en data los datos de autenticación y en headers la cabecera que se utiliza cuando enviamos un formulario. Por último enviamos el parámetro allow_redirects igual a True para obtener la respuesta final.

Si todo ha ido bien la respuesta nos devolverá el status 200 y el token de acceso.

Para verificar que funcionan nuestros tests, vamos a la terminal y en la raíz de nuestro proyecto lanzamos el siguiente comando:

pytest

Si todo ha ido bien, nos aparecerá un mensaje en la consola afirmando que todos los tests han pasado.

Testando los endpoints de To-dos

Ahora que ya tenemos los tests para los usuarios, vamos a testar los endpoints de las tareas por hacer. Para ello dentro de la carpeta /tests crearemos un archivo llamado test_todo.py con el siguiente contenido:

from fastapi.testclient import TestClient
from main import app


def create_user_and_make_login(username: str):
    client = TestClient(app)

    user = {
        'email': f'{username}@cosasdedevs.com',
        'username': username,
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/user/',
        json=user,
    )

    login = {
        'username': username,
        'password': 'admin123'
    }

    response = client.post(
        '/api/v1/login/',
        data=login,
        headers={
            'Content-Type': 'application/x-www-form-urlencoded'
        },
        allow_redirects=True
    )

    data = response.json()
    return data['access_token']

Al igual que en los tests de usuarios importamos TestClient y app y aquí hemos creado una función que no empieza por la palabra test_, por lo tanto, no se ejecutará al menos que la llamemos. Esta función la he preparado para que nos permita crear un usuario y realizar la autenticación y así evitar duplicar código.

Ahora sí, vamos a añadir nuestro primer test. Para ello añadimos la siguiente función:

def test_create_todo_ok():
    token = create_user_and_make_login('test_create_todo_ok')

    client = TestClient(app)

    todo = {
        'title': 'My first task'
    }

    response = client.post(
        '/api/v1/to-do/',
        json=todo,
        headers={
            'Authorization': f'Bearer {token}'
        }
    )

    assert response.status_code == 201, response.text
    data = response.json()
    assert data['title'] == todo['title']
    assert data['is_done'] == False

Como veis, esta función crea un usuario y recupera el token de autenticación, después empleamos client para lanzar la petición que se encarga de crear la tarea. También fijaos en que en headers estamos enviando el token de autenticación.

Por último comprobamos el status_code y el resultado para confirmar que todo ha funcionado correctamente.

Podríamos añadir muchos más test para testar la creación y demás endpoints, como la actualización de estados y eliminación de tareas o verificar que un usuario no tiene acceso a las tareas de otro, pero con las bases que os he dejado pienso que podréis hacerlo sin mi ayuda perfectamente 💪 y si no me preguntáis por los comentarios 😅. 

Y listo, ya hemos completado nuestro proyecto, espero que hayáis aprendido mucho con esta serie y nada, ahora a darle caña a FastAPI 🤘.

Como en tutoriales anteriores, os dejo repositorio por si necesitáis echarle un vistazo https://github.com/albertorc87/fastapi-todo-api/tree/tutorial-6-tests.

Edit:

Actualmente, según las métricas de la web, solo el 8% de los usuarios que empezaron la serie de FastAPI la han terminado completamente. Si eres uno de ellos y has llegado hasta aquí, enhorabuena 🙌. Si quieres responde este tweet https://twitter.com/alberrc87/status/1476820985932238849 con el emoji 🐍 para que sepa que eres uno de ellos.

Espero que este post te ayude y como siempre, te recomiendo seguirme en Twitter para estar al tanto de los nuevo contenido. Ahora también puedes seguirme en Instagram donde estoy subiendo tips, tutoriales en vídeo e información sobre herramientas para developers.

Por último os dejo mi guía para aprender a trabajar con APIs donde explico todo el funcionamiento de una API, el protocolo HTTP y veremos como construir una API con arquitectura REST.

Nos leemos 👋.