Подробнее на официальном сайте: 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
Статья с Хабр
Дополнительное видео:
