pip install -U locust==2.5.1
#pip install -U locust==2.5.1 -i https://pypi.tuna.tsinghua.edu.cn/simple

from locust import HttpUser, task, between

# Эквивалентно моделированию пользователя
class WebUser(HttpUser):
    # Адрес сервера
    host = 'http://localhost:8000'

    # Время ожидания перед следующим заданием. Изучите время обдумывания для пользователя шаблона.
    wait_time = between(3, 5)  # Здесь используется случайный 3,5 секунды

    # Определите тестовую задачу, похожую на сэмплер.
    @task(1)
    def info(self):
        # Получить информацию о пользователе
        self.client.get('/info/')

    @task(10)
    def addresses(self):
        # Получить адрес пользователя
        self.client.get('/addresses/')

    def on_start(self):
        # тест Доосуществлятьиздействовать        print('Вход пользователя')
        auth = {'username': 'root', 'password': 'root1234'}
        self.client.post('/login/', json=auth)

    def on_stop(self):
        # тест Заканчиватьосуществлятьиздействовать        print('Пользователь вышел')
        self.client.delete('/logout/')

# Запускать напрямую
locust

# Укажите файл
locust -f xxx/locustfile.py

from locust import User, events, task, between
import random


def success_call(name, recvText, total_time):
    events.request_success.fire(
        request_type="[Success]",
        name=name,
        response_time=total_time,
        response_length=len(recvText)
    )


def fail_call(name, total_time):
    events.request_failure.fire(
        request_type="[Fail]",
        name=name,
        response_time=total_time,
        response_length=0,
        exception=Exception(),
    )


class MyUser(User):
    wait_time = between(1, 5)

    @task(1)
    def job1(self):
        print('This is job 1')
        self.sendMsg("job1", 'This is job 2')

    @task(2)
    def job2(self):
        print('This is job 2')
        self.sendMsg("job2", 'This is job 2')

    def sendMsg(self, methodname, msg):
        num = random.randint(0, 2)
        time = random.randint(10, 50)
        if (num == 0):
            fail_call(methodname, time)
        else:
            success_call(methodname, msg, time);

locust -f test4.py --autostart --autoquit 0 -u 1 -r 3 --run-time 10s

locust -f locustfile.py --master --headless --master-bind-host=127.0.0.1 --master-bind-port=8090 -t 15s -u 5 -r 1

locust -f locustfile.py --worker --master-host=127.0.0.1 --master-port=8090 --headless

import uuid

from locust import HttpUser, task, between, TaskSet

# зарегистрироваться
class RegistTaskSet(TaskSet):
    # Имя пользователя дублируется?
    @task
    def username_count(self):
        username = 'zhangsan'
        self.client.get(f'/usernames/{username}/count/')

    # Дублируется ли номер мобильного телефона?
    @task
    def mobile_count(self):
        mobile = '13712345678'
        self.client.get(f'/mobiles/{mobile}/count/')

    # Получить графический код подтверждения
    @task
    def image_codes(self):
        image_code_id = str(uuid.uuid4())
        self.client.get(f'/image_codes/{image_code_id}/')

    @task
    def stop(self):
        print('Выход из текущего набора действий')
        self.interrupt()

# Авторизоваться
class UserLoginTaskSet(TaskSet):
    # Вложенный пользователь зарегистрироваться
    tasks = [RegistTaskSet]

    @task
    def username_login(self):
        auth = {'username': 'root', 'password': 'root1234'}
        self.client.post('/login/', json=auth)

    @task
    def stop(self):
        print('Выход из текущего набора действий')
        self.interrupt()

# Выдавать себя за пользователя
class WebUser(HttpUser):
    # Сумка Содержит пользователей Авторизоваться
    tasks = [UserLoginTaskSet]
    # Адрес сервера
    host = 'http://localhost:8000'

    # Время ожидания перед следующим заданием. Изучите время обдумывания для пользователя шаблона.
    wait_time = between(3, 5)  # Здесь используется случайный 3,5 секунды

    @task
    def search(self):
        # поиск
        self.client.get('/search/')

from locust import HttpUser, task, between, SequentialTaskSet

class RegistSeqTaskSet(SequentialTaskSet):
    # Имя пользователя дублируется?
    @task
    def username_count(self):
        print('Проверить имя пользователя')
        username = 'zhangsan'
        self.client.get(f'/usernames/{username}/count/')

    # Дублируется ли номер мобильного телефона?
    @task
    def mobile_count(self):
        print('Подтвердить номер мобильного телефона')
        mobile = '13712345678'
        self.client.get(f'/mobiles/{mobile}/count/')

    # Получить графический код подтверждения
    @task
    def image_codes(self):
        print('Получить графический код подтверждения')
        image_code_id = str(uuid.uuid4())
        self.client.get(f'/image_codes/{image_code_id}/')

# Выдавать себя за пользователя
class WebUser(HttpUser):
    # Сумка Содержит пользователей Авторизоваться
    tasks = [RegistSeqTaskSet]
    # Адрес сервера
    host = 'http://localhost:8000'

    # Время ожидания перед следующим заданием. Изучите время обдумывания для пользователя шаблона.
    wait_time = between(3, 5)  # Здесь используется случайный 3,5 секунды

from locust import HttpUser, task, between

class WebUser(HttpUser):
    wait_time = between(1, 5)
    host = 'http://localhost:8000'

    @task
    def info(self):
        with self.client.get('/info/', catch_response=True) as resp:
            res = resp.json()
            if res['code'] != 0:
                resp.failure(res['errmsg'])

pip install websockets

import asyncio
import websockets
import json
import random

async def mytest():
    async with websockets.connect('wss://sockettest.xhkjedu.com/ws') as websocket:
        num = random.randint(0, 10000000)
        msg = {
            "b": {"num": num},
            "c": 123456,
        }

        msgstr = json.dumps(msg)

        await websocket.send(msgstr)
        print(f"↑: {msgstr}")

        greeting = await websocket.recv()
        print(f"↓: {greeting}")

asyncio.get_event_loop().run_until_complete(mytest())

pip install websocket-client

from websocket import create_connection
import json
import random

ws = create_connection("wss://sockettest.xhkjedu.com/ws")

num = random.randint(0, 10000000)
msg = {
    "b": {"num": num},
    "c": 123456,
}

msgstr = json.dumps(msg)
print("Sending " + msgstr)
ws.send(msgstr)

result = ws.recv()
print("Received '%s'" % result)
ws.close()

import websocket

def on_message(ws, message):
    print(ws)
    print(message)

def on_error(ws, error):
    print(ws)
    print(error)

def on_close(ws):
    print(ws)
    print("### closed ###")

websocket.enableTrace(True)
ws = websocket.WebSocketApp("ws://127.0.0.1:8888/track",
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)

ws.run_forever()

from locust import User, task, events
import time
from websocket import create_connection
import json
import random

def success_call(name, recvText, total_time):
    events.request_success.fire(
        request_type="[Success]",
        name=name,
        response_time=total_time,
        response_length=len(recvText)
    )

def fail_call(name, total_time, e):
    events.request_failure.fire(
        request_type="[Fail]",
        name=name,
        response_time=total_time,
        response_length=0,
        exception=e,
    )

class WebSocketClient(object):
    def __init__(self, host):
        self.host = host
        self.ws = None

    def connect(self, burl):
        self.ws = create_connection(burl)

    def recv(self):
        return self.ws.recv()

    def send(self, msg):
        self.ws.send(msg)

class WebsocketUser(User):
    abstract = True

    def __init__(self, *args, **kwargs):
        super(WebsocketUser, self).__init__(*args, **kwargs)
        self.client = WebSocketClient(self.host)
        self.client._locust_environment = self.environment

class ApiUser(WebsocketUser):
    host = "wss://sockettest.xhkjedu.com/"

    @task(1)
    def pft(self):
        # wss адрес
        self.url = 'wss://sockettest.xhkjedu.com/ws'
        print("перед подключением")
        start_time = time.time()
        try:
            self.client.connect(self.url)
            print("После подключения")
            # Отправить запрос на подписку
            num = random.randint(0, 10000000)
            msg = {
                "b": {"num": num},
                "c": 123456,
            }

            msgstr = json.dumps(msg)

            self.client.send(msgstr)
            print(f"↑: {msgstr}")

            greeting = self.client.recv()
            print(f"↓: {greeting}")

        except Exception as e:
            total_time = int((time.time() - start_time) * 1000)
            fail_call("Send", total_time, e)
        else:
            total_time = int((time.time() - start_time) * 1000)
            success_call("Send", "success", total_time)

locust -f main.py -u 5000 -r 300

locust -f main.py --autostart --autoquit 0 -u 1 -r 3 --run-time 10s