Pruebas de Carga con Locust

¿Qué es Locust? #

Locust es una herramienta open-source de pruebas de carga escrita en Python. Su característica definitoria es que escribes tus pruebas como código Python puro, definiendo el comportamiento del usuario como clases Python. Si tú o tu equipo se sienten cómodos con Python, Locust ofrece la barrera de entrada más baja de cualquier herramienta de pruebas de carga.

Locust usa una arquitectura orientada a eventos (basada en gevent) en lugar de hilos, lo que permite que un solo proceso simule miles de usuarios concurrentes. Incluye una interfaz web integrada para monitorear pruebas en tiempo real y soporta testing distribuido entre múltiples máquinas.

El nombre “Locust” (langosta) proviene del comportamiento de enjambre de las langostas — defines comportamientos de usuario y Locust desata un enjambre de ellos sobre tu aplicación.

Cuándo Elegir Locust #

Elige Locust cuando: Tu equipo conoce Python, quieres una web UI en tiempo real, necesitas usar bibliotecas Python en tus pruebas (drivers de BD, ML, protocolos custom), o necesitas comportamiento de usuario altamente personalizable.

Instalación #

pip install locust

Verificar:

locust --version

Tu Primera Prueba con Locust #

Crea un archivo llamado locustfile.py:

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # esperar 1-3 segundos entre tareas

    @task(3)
    def view_products(self):
        self.client.get("/api/products")

    @task(1)
    def view_product_detail(self):
        self.client.get("/api/products/1")

Ejecútalo:

locust -f locustfile.py --host=https://api.example.com

Abre http://localhost:8089 en tu navegador para ver la web UI. Ingresa el número de usuarios, tasa de generación y haz clic en Start.

Conceptos Clave #

HttpUser: Una clase que representa un usuario virtual. Cada instancia simula un usuario.

wait_time: Controla la pausa entre tareas. between(1, 3) significa una espera aleatoria de 1-3 segundos. Otras opciones:

• constant(2) — siempre esperar 2 segundos
• constant_pacing(5) — asegurar que cada ciclo de tarea tome exactamente 5 segundos

Decorador @task: Marca métodos como tareas de usuario. El argumento numérico establece el peso — @task(3) significa que esta tarea se ejecuta 3 veces más frecuentemente que @task(1).

Pesos de Tareas y Tareas Secuenciales #

Tareas con Peso #

Los pesos de tareas modelan comportamiento realista del usuario. En una aplicación de e-commerce típica, navegar es mucho más común que comprar:

class EcommerceUser(HttpUser):
    wait_time = between(1, 5)

    @task(10)
    def browse_products(self):
        self.client.get("/api/products")

    @task(5)
    def search(self):
        self.client.get("/api/search?q=laptop")

    @task(3)
    def view_product(self):
        self.client.get("/api/products/42")

    @task(1)
    def add_to_cart(self):
        self.client.post("/api/cart", json={"product_id": 42, "qty": 1})

Aquí, navegar ocurre 10 veces más frecuentemente que agregar al carrito — lo cual refleja el comportamiento real del usuario.

Tareas Secuenciales (TaskSets) #

Para flujos ordenados, usa SequentialTaskSet:

from locust import HttpUser, SequentialTaskSet, task, between

class PurchaseFlow(SequentialTaskSet):
    @task
    def login(self):
        response = self.client.post("/api/auth/login", json={
            "username": "testuser",
            "password": "testpass"
        })
        self.token = response.json()["token"]

    @task
    def browse(self):
        self.client.get("/api/products", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task
    def add_to_cart(self):
        self.client.post("/api/cart", json={"product_id": 1, "qty": 1},
                         headers={"Authorization": f"Bearer {self.token}"})

    @task
    def checkout(self):
        self.client.post("/api/checkout",
                         headers={"Authorization": f"Bearer {self.token}"})
        self.interrupt()  # volver a la clase padre

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)
    tasks = [PurchaseFlow]

Hooks del Ciclo de Vida #

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        """Se llama cuando un usuario inicia. Úsalo para login/setup."""
        response = self.client.post("/api/auth/login", json={
            "username": "user1", "password": "pass123"
        })
        self.token = response.json()["token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    def on_stop(self):
        """Se llama cuando un usuario se detiene. Úsalo para cleanup."""
        self.client.post("/api/auth/logout", headers=self.headers)

    @task
    def browse(self):
        self.client.get("/api/products", headers=self.headers)

Validación Personalizada #

@task
def get_products(self):
    with self.client.get("/api/products", catch_response=True) as response:
        if response.status_code != 200:
            response.failure(f"Status recibido {response.status_code}")
        elif "products" not in response.json():
            response.failure("Respuesta sin campo 'products'")
        elif len(response.json()["products"]) == 0:
            response.failure("Lista de productos vacía")
        else:
            response.success()

Testing Distribuido #

Locust soporta testing distribuido con arquitectura master/worker:

# Iniciar master
locust -f locustfile.py --master --host=https://api.example.com

# Iniciar workers (en la misma o diferentes máquinas)
locust -f locustfile.py --worker --master-host=192.168.1.100
locust -f locustfile.py --worker --master-host=192.168.1.100

El master coordina la prueba y agrega resultados. Los workers generan la carga real. Cada worker puede simular miles de usuarios.

La Web UI #

La web UI de Locust en http://localhost:8089 proporciona:

• Gráficos en tiempo real: Solicitudes por segundo, tiempos de respuesta, número de usuarios
• Tabla de estadísticas: Métricas por solicitud (mediana, p95, p99, max, tasa de fallos)
• Pestaña de fallos: Mensajes de error detallados
• Descarga de datos: Exportación CSV de resultados
• Detener/Reiniciar: Control de la prueba sin reiniciar

Para ejecución headless (CI/CD):

locust -f locustfile.py --headless -u 100 -r 10 --run-time 5m --host=https://api.example.com

Ejercicio: Prueba de Carga Multi-Comportamiento con Locust #

Escribe una prueba de Locust que simule tres tipos distintos de usuarios para una plataforma de contenido.

Escenario #

Una plataforma de contenido tiene tres tipos de usuarios:

1. Lectores (70%) — navegan artículos, leen contenido
2. Autores (20%) — crean y editan artículos
3. Admins (5%) — gestionan usuarios y ven analytics

Requisitos #

1. Crea clases de usuario separadas para cada tipo con pesos de tarea apropiados
2. Usa on_start para autenticación
3. Agrega validación personalizada para contenido de respuesta
4. Usa between() para think time realista
5. Haz que la clase de usuario admin se ejecute menos frecuentemente (usa weight en la clase)

class ReaderUser(HttpUser):
    weight = 70  # 70% de los usuarios son lectores
    wait_time = between(2, 5)

class AuthorUser(HttpUser):
    weight = 20  # 20% son autores
    wait_time = between(3, 8)

class AdminUser(HttpUser):
    weight = 5   # 5% son admins
    wait_time = between(5, 10)

from locust import HttpUser, task, between
import random

class ReaderUser(HttpUser):
    weight = 70
    wait_time = between(2, 5)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": f"reader_{random.randint(1, 1000)}",
            "password": "readerpass"
        })
        if response.status_code == 200:
            self.token = response.json()["token"]
            self.headers = {"Authorization": f"Bearer {self.token}"}
        else:
            self.headers = {}

    @task(5)
    def browse_articles(self):
        with self.client.get("/api/articles", headers=self.headers,
                             catch_response=True) as response:
            if response.status_code == 200:
                articles = response.json().get("articles", [])
                if len(articles) > 0:
                    response.success()
                else:
                    response.failure("No se retornaron artículos")
            else:
                response.failure(f"Status: {response.status_code}")

    @task(3)
    def read_article(self):
        article_id = random.randint(1, 100)
        with self.client.get(f"/api/articles/{article_id}",
                             headers=self.headers,
                             catch_response=True) as response:
            if response.status_code == 200:
                body = response.json()
                if "title" in body and "content" in body:
                    response.success()
                else:
                    response.failure("Artículo sin título o contenido")

    @task(1)
    def search_articles(self):
        queries = ["python", "testing", "qa", "automation", "ci/cd"]
        query = random.choice(queries)
        self.client.get(f"/api/search?q={query}", headers=self.headers)


class AuthorUser(HttpUser):
    weight = 20
    wait_time = between(3, 8)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": f"author_{random.randint(1, 50)}",
            "password": "authorpass"
        })
        self.token = response.json()["token"]
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }

    @task(3)
    def view_my_articles(self):
        self.client.get("/api/articles/mine", headers=self.headers)

    @task(2)
    def create_draft(self):
        self.client.post("/api/articles", headers=self.headers, json={
            "title": f"Artículo de Prueba {random.randint(1, 10000)}",
            "content": "Contenido de artículo de prueba para load testing.",
            "status": "draft"
        })

    @task(1)
    def edit_article(self):
        article_id = random.randint(1, 50)
        self.client.put(f"/api/articles/{article_id}", headers=self.headers, json={
            "title": "Título Actualizado",
            "content": "Contenido actualizado."
        })


class AdminUser(HttpUser):
    weight = 5
    wait_time = between(5, 10)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": "admin",
            "password": "adminpass"
        })
        self.token = response.json()["token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def view_analytics(self):
        self.client.get("/api/admin/analytics", headers=self.headers)

    @task(2)
    def list_users(self):
        self.client.get("/api/admin/users", headers=self.headers)

    @task(1)
    def view_system_health(self):
        self.client.get("/api/admin/health", headers=self.headers)

# Con web UI
locust -f locustfile.py --host=https://content-api.example.com

# Headless para CI/CD
locust -f locustfile.py --headless -u 200 -r 20 --run-time 10m \
  --host=https://content-api.example.com --csv=results

Tips Profesionales #

• Custom Load Shapes: Crea una clase que extienda LoadTestShape para definir patrones de carga complejos (spike, step, wave) con toda la flexibilidad de Python. Esto es más potente que el load shaping integrado de cualquier otra herramienta.
• Sistema de Eventos: Usa los hooks de eventos de Locust (@events.test_start.add_listener, @events.request.add_listener) para agregar logging personalizado, métricas o notificaciones durante la ejecución.
• FastHttpUser: Para máximo throughput, usa FastHttpUser en lugar de HttpUser. Usa un cliente HTTP basado en C (geventhttpclient) que es 5-6 veces más rápido para solicitudes simples.
• Filtrado por Tags: Usa el decorador @tag('smoke') y ejecuta con --tags smoke para ejecutar solo tareas etiquetadas — útil para ejecutar subconjuntos de una suite de pruebas grande.
• Docker Compose para Distribuido: Usa Docker Compose para levantar un master y múltiples workers con un solo comando, haciendo el testing distribuido repetible y fácil de escalar.