int-layer/scripts/load_test.py

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "locust>=2.32.5",
# ]
# ///
"""Locust load testing script for capa-de-integracion service.

Usage:
    # Run with web UI (default port 8089)
    uv run scripts/load_test.py

    # Run headless with specific users and spawn rate
    uv run scripts/load_test.py --headless -u 100 -r 10

    # Run against specific host
    uv run scripts/load_test.py --host http://localhost:8080

    # Run for specific duration
    uv run scripts/load_test.py --headless -u 50 -r 5 --run-time 5m
"""

import random
from locust import HttpUser, between, task


class ConversationUser(HttpUser):
    """Simulate users interacting with the conversation API."""

    wait_time = between(1, 3)  # Wait 1-3 seconds between tasks
    host = "http://localhost:8080"

    # Sample data for realistic load testing
    phone_numbers = [
        f"555-{1000 + i:04d}" for i in range(100)
    ]  # 100 unique phone numbers

    conversation_messages = [
        "Hola",
        "¿Cuál es mi saldo?",
        "Necesito ayuda con mi tarjeta",
        "¿Dónde está mi sucursal más cercana?",
        "Quiero hacer una transferencia",
        "¿Cómo puedo activar mi tarjeta?",
        "Tengo un problema con mi cuenta",
        "¿Cuáles son los horarios de atención?",
    ]

    notification_messages = [
        "Tu tarjeta fue bloqueada por seguridad",
        "Se detectó un cargo de $1,500 en tu cuenta",
        "Tu préstamo fue aprobado",
        "Transferencia recibida: $5,000",
        "Recordatorio: Tu pago vence mañana",
    ]

    screen_contexts = [
        "home",
        "card_management",
        "account_details",
        "transfers",
        "help_center",
    ]

    def on_start(self):
        """Called when a simulated user starts."""
        # Assign a phone number to this user for the session
        self.phone = random.choice(self.phone_numbers)
        self.nombre = f"Usuario_{self.phone.replace('-', '')}"

    @task(5)
    def health_check(self):
        """Health check endpoint - most frequent task."""
        with self.client.get("/health", catch_response=True) as response:
            if response.status_code == 200:
                data = response.json()
                if data.get("status") == "healthy":
                    response.success()
                else:
                    response.failure("Health check returned unhealthy status")
            else:
                response.failure(f"Got status code {response.status_code}")

    @task(10)
    def detect_intent(self):
        """Test the main conversation endpoint."""
        payload = {
            "mensaje": random.choice(self.conversation_messages),
            "usuario": {
                "telefono": self.phone,
                "nickname": self.nombre,
            },
            "canal": "web",
            "pantallaContexto": random.choice(self.screen_contexts),
        }

        with self.client.post(
            "/api/v1/dialogflow/detect-intent",
            json=payload,
            catch_response=True,
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "responseId" in data or "queryResult" in data:
                    response.success()
                else:
                    response.failure("Response missing expected fields")
            elif response.status_code == 400:
                response.failure(f"Validation error: {response.text}")
            elif response.status_code == 500:
                response.failure(f"Internal server error: {response.text}")
            else:
                response.failure(f"Unexpected status code: {response.status_code}")

    @task(3)
    def send_notification(self):
        """Test the notification endpoint."""
        payload = {
            "texto": random.choice(self.notification_messages),
            "telefono": self.phone,
            "parametrosOcultos": {
                "transaction_id": f"TXN{random.randint(10000, 99999)}",
                "amount": random.randint(100, 10000),
            },
        }

        with self.client.post(
            "/api/v1/dialogflow/notification",
            json=payload,
            catch_response=True,
        ) as response:
            if response.status_code == 200:
                response.success()
            elif response.status_code == 400:
                response.failure(f"Validation error: {response.text}")
            elif response.status_code == 500:
                response.failure(f"Internal server error: {response.text}")
            else:
                response.failure(f"Unexpected status code: {response.status_code}")

    @task(4)
    def quick_reply_screen(self):
        """Test the quick reply screen endpoint."""
        payload = {
            "usuario": {
                "telefono": self.phone,
                "nombre": self.nombre,
            },
            "pantallaContexto": random.choice(self.screen_contexts),
        }

        with self.client.post(
            "/api/v1/quick-replies/screen",
            json=payload,
            catch_response=True,
        ) as response:
            if response.status_code == 200:
                data = response.json()
                if "responseId" in data and "quick_replies" in data:
                    response.success()
                else:
                    response.failure("Response missing expected fields")
            elif response.status_code == 400:
                response.failure(f"Validation error: {response.text}")
            elif response.status_code == 500:
                response.failure(f"Internal server error: {response.text}")
            else:
                response.failure(f"Unexpected status code: {response.status_code}")


class ConversationFlowUser(HttpUser):
    """Simulate realistic conversation flows with multiple interactions."""

    wait_time = between(2, 5)
    host = "http://localhost:8080"
    weight = 2  # This user class will be 2x more likely to be chosen

    def on_start(self):
        """Initialize user session."""
        self.phone = f"555-{random.randint(2000, 2999):04d}"
        self.nombre = f"Flow_User_{random.randint(1000, 9999)}"

    @task
    def complete_conversation_flow(self):
        """Simulate a complete conversation flow."""
        # Step 1: Start with quick replies
        screen_payload = {
            "usuario": {
                "telefono": self.phone,
                "nombre": self.nombre,
            },
            "pantallaContexto": "help_center",
        }
        self.client.post("/api/v1/quick-replies/screen", json=screen_payload)

        # Step 2: Have a conversation
        conversation_steps = [
            "Hola, necesito ayuda",
            "¿Cómo puedo verificar mi saldo?",
            "Gracias por la información",
        ]

        for mensaje in conversation_steps:
            payload = {
                "mensaje": mensaje,
                "usuario": {
                    "telefono": self.phone,
                    "nickname": self.nombre,
                },
                "canal": "mobile",
                "pantallaContexto": "help_center",
            }
            self.client.post("/api/v1/dialogflow/detect-intent", json=payload)
            # Small delay between messages
            self.wait()


if __name__ == "__main__":
    import os
    import sys

    # Set default host if not provided via command line
    if "--host" not in sys.argv and "HOST" not in os.environ:
        os.environ["HOST"] = "http://localhost:8080"

    # Import and run locust
    from locust import main as locust_main

    # Run locust with command line arguments
    sys.exit(locust_main.main())