Locust: нагрузочное тестирование

Подробнее на официальном сайте: https://docs.locust.io/en/latest/quickstart.html

Создание простого теста:

from locust import HttpUser, task, between

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

    @task(1)
    def auth(self):
        self.client.get("/exists")

Для проверки можно выполнить:

import requests as r
base_url = 'http://localhost:5007/release'
r.get(base_url + '/exists')

# <Response [200]>

Авторизация

from locust import HttpUser, task, between

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

    @task(1)
    def auth(i):
        i.client.post('/auth', { "UserName": "user", "Password": "1234", "Version": "0.0.0.0" })

Для проверки в консоли можно выполнить:

import requests as r
base_url = 'http://localhost:5007/release'
r.post(base_url + '/auth', { "UserName": "user", "Password": "1234", "Version": "0.0.0.0" }) 
#<Response [200]>
>>> response = r.post(base_url + '/auth', { "UserName": "user", "Password": "1234", "Version": "0.0.0.0" })
response.json()
# {'token': 'dXNlcjoxMjM0', 'user': {'id': 2, 'login': 'user', 'claims': '.user.', 'date': '2022-06-21T04:14:27.559Z'}, 'ip': '::1'}

Переопределение статуса ответа:

def exists(i):
    with i.client.get('/exists', catch_response=True) as response:
        if response.status_code == 200:
            response.success()
        else:
            response.failure(f'status code is {response.status_code}')

Параметр catch_response=True обязателен, иначе вывод ошибки в response.failure не будет.

Группировка запросов для оптимального просмотра:

with self.client.get(f'/photos/{photo_id}', catch_response=True, name='/photos/[id]') as response:
    if response.status_code == 200:
            response.success()
        else:
            response.failure(f'status code is {response.status_code}')

Параметр name=’/photos/[id]’ предназначен для группировки результатов по «маске«.

Создание собственного исключения:

class FlowException(Exception):
    pass
...
def check_flow(self):
    new_post = {...}
    response = self.client.post('/posts', json=new_post)
    if response.status_code != 201:
        raise FlowException('post not created')
    post_id = response.json().get('id')

• raise FlowException(‘post not created’) — остановка сценария. Можно в сценарии указать и return, но при этом во вкладке Exception мы не увидим ошибки

• response.json().get(‘id’) — чтение id из результата

Сценарий временного хранения:

from locust import HttpLocust, TaskSet, task
import random as r

# class UserBehavior(TaskSet): 
#     def __init__(self, parent):
#         super(UserBehavior, self).__init__(parent)
#         self.created_posts = list()

class UserBehavior(TaskSet):
    created_posts = []

    @task(1)
    def create_post(self):
        ...
        post_id = response.json().get('id')
        self.created_posts.append(host_id)

    @task(10)
    def read_post(self):
        if len(self.created_posts) == 0:
            return
        post_id = r.chioce(self.created_posts)

• r.chioce(self.created_posts) — рандомный выбор записи
• в закомментированной части указан пример реализации, чтобы набор постов для каждого пользователя был своим

Выполнение под сценариев:

class Todo(TaskSet):
    @task(3)
    def index(self)
        self.client.get('/todos')
    
    @task(1)
    def stop(self):
        self.interrupt()

class UserBehavior(TaskSet):
     tasks = {Todo: 1}
   
     @task(3)
     def index(self):
         self.client.get('/')

     @task(2)
     def posts(self):
         self.client.get('/posts')

• tasks = {Todo: 1} — объявление под задачи
• self.interrupt() — чтобы принудительно завершить тест

Передача заголовка авторизации:

...
import requests

class UserBehavior(TaskSet):
    def on_start(self):
        response = requests.post('/login', {...})
        self.client.headers.update({'Authorization': response.headers.get('RPC-Authorization')})
        # get token from cookies
        self.client.headers.update({'Authorization': response.cookies.get('RPC-Authorization')})
        # get token from body
        self.client.headers.update({'Authorization': str(response.content.decode().find('google'))})

     @task(21)
     def posts(self):
         ...

Распределенное тестирование.

• Мастер — экземпляр, который собирает данные со slave «машин»
• Slave — экземпляры, которые производят тестирование, и передают данные мастеру

locust -f test.py --master --host=http://localhost:3000

locust -f test.py --slave --master-host=http://localhost

Запуск теста без web-интерфейса:

locust -f test.py --host=http://localhost --no-web -c 10 -r 2 --run-time 1m --csv=test_res

• no-web — не создаем веб-интерфейс
• c — количество пользователей
• r — прирастание пользователей в секунду
• run-time — время теста в минутах (параметр времени)
• csv — результат теста

Игнорирование сертификата HTTPS:

self.client.post('/posts', data, verify=False)

verify=False — не будет проверки сертификата

Логирование:

locust -f main.py --host=http://localhost --logfile=locustfile.log

Подробнее тут https://docs.locust.io/en/stable/logging.html

Статья с Хабр

• https://habr.com/ru/post/433304/

Дополнительное видео: