Locust Load Testing Tutorial: Руководство по Нагрузочному Тестированию на Python

pip install locust

locust --version

# locustfile.py
from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # Ожидание 1-3 секунды между задачами

    @task
    def homepage(self):
        self.client.get("/")

    @task(3)  # В 3 раза чаще чем homepage
    def api_endpoint(self):
        self.client.get("/api/products")

# Запуск с web UI
locust -f locustfile.py --host=https://api.example.com

# Открой http://localhost:8089
# Укажи количество пользователей и spawn rate
# Нажми Start

locust -f locustfile.py \
    --host=https://api.example.com \
    --users 100 \
    --spawn-rate 10 \
    --run-time 5m \
    --headless

from locust import HttpUser, task, between

class ApiUser(HttpUser):
    wait_time = between(0.5, 2)

    @task(5)
    def list_products(self):
        self.client.get("/api/products")

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

    @task(1)
    def create_order(self):
        self.client.post("/api/orders", json={
            "product_id": 1,
            "quantity": 2
        })

from locust import HttpUser, task, between

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

    def on_start(self):
        # Выполняется один раз при старте пользователя
        response = self.client.post("/api/login", json={
            "username": "test",
            "password": "password"
        })
        self.token = response.json()["token"]

    def on_stop(self):
        # Выполняется один раз при остановке пользователя
        self.client.post("/api/logout")

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

from locust import HttpUser, task, between, SequentialTaskSet

class UserFlow(SequentialTaskSet):
    @task
    def browse(self):
        self.client.get("/products")

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

    @task
    def checkout(self):
        self.client.post("/checkout")

class EcommerceUser(HttpUser):
    tasks = [UserFlow]
    wait_time = between(1, 3)

from locust import HttpUser, task

class ValidatingUser(HttpUser):
    @task
    def check_api(self):
        with self.client.get("/api/status", catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Got {response.status_code}")
            elif response.elapsed.total_seconds() > 2:
                response.failure("Too slow!")
            elif "ok" not in response.text:
                response.failure("Missing 'ok' in response")
            else:
                response.success()

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

# На каждой worker машине
locust -f locustfile.py --worker --master-host=192.168.1.100

# docker-compose.yml
version: '3'
services:
  master:
    image: locustio/locust
    ports:
      - "8089:8089"
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --master -H https://api.example.com

  worker:
    image: locustio/locust
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --worker --master-host master
    deploy:
      replicas: 4

locust -f locustfile.py \
    --headless \
    --users 100 \
    --spawn-rate 10 \
    --run-time 5m \
    --csv=results \
    --html=report.html

# .github/workflows/load-test.yml
name: Load Tests

on:
  schedule:
    - cron: '0 2 * * *'  # Ежедневно в 2:00

jobs:
  load-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install Locust
        run: pip install locust

      - name: Run Load Test
        run: |
          locust -f locustfile.py \
            --host=${{ secrets.API_URL }} \
            --users 50 \
            --spawn-rate 5 \
            --run-time 2m \
            --headless \
            --csv=results

      - name: Upload Results
        uses: actions/upload-artifact@v4
        with:
          name: load-test-results
          path: results*.csv