int-layer/locustfile.py

"""Locust load testing for capa-de-integracion service.

Usage:
    # Run with web UI (default port 8089)
    locust --host http://localhost:8080

    # Run headless with specific users and spawn rate
    locust --host http://localhost:8080 --headless -u 100 -r 10

    # Run for specific duration
    locust --host http://localhost:8080 --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)

    phone_numbers = [
        f"555-{1000 + i:04d}" for i in range(100)
    ]

    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",
        "pagos",
        "transferencia",
        "prestamos",
        "inversiones",
        "lealtad",
        "finanzas",
        "capsulas",
        "descubre",
        "retiro-sin-tarjeta",
        "detalle-tdc",
        "detalle-tdd",
    ]

    def on_start(self):
        """Called when a simulated user starts."""
        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)
    weight = 2

    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."""
        screen_payload = {
            "usuario": {
                "telefono": self.phone,
                "nombre": self.nombre,
            },
            "pantallaContexto": "home",
        }
        self.client.post("/api/v1/quick-replies/screen", json=screen_payload)

        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": "home",
            }
            self.client.post("/api/v1/dialogflow/detect-intent", json=payload)
            self.wait()