pip3 install locust

locust -V
# locust 2.5.1

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

from locust import User, TaskSet, constant,task


class ForumSection(TaskSet):
    wait_time = constant(1)

    @task(10)
    def job1(self):
        print("job1----")

    @task
    def job2(self):
        print("job2+++++")

    @task
    def stop(self):
        self.interrupt()


class LoggedInUser(User):
    wait_time = constant(5)
    tasks = {ForumSection: 2}

    @task
    def my_task(self):
        print("my_task=======")

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

class MyUser(User):
    @task
    class MyTaskSet(TaskSet):
        ...

from locust import User, SequentialTaskSet, constant,task


class MySet(SequentialTaskSet):
    wait_time = constant(1)

    @task(10)
    def job1(self):
        print("job1----")

    @task
    def job2(self):
        print("job2+++++")

    @task
    def stop(self):
        print("stop")
        self.interrupt()


class LoggedInUser(User):
    wait_time = constant(5)
    tasks = {MySet: 2}

    @task
    def my_task(self):
        print("my_task=======")

locust -f test6.py --autostart --autoquit 0 -u 1 -r 3 --run-time 60s

from locust import User, SequentialTaskSet, constant, task


class ASet(SequentialTaskSet):
    wait_time = constant(1)

    @task(2)
    def job1(self):
        print("a-1")

    @task
    def job2(self):
        print("a-2")

    @task
    def stop(self):
        print("a-stop")
        self.interrupt()


class BSet(SequentialTaskSet):
    wait_time = constant(1)

    @task(2)
    def job1(self):
        print("b-1")

    @task
    def job2(self):
        print("b-2")

    @task
    def stop(self):
        print("b-stop")
        self.interrupt()


class LoggedInUser(User):
    wait_time = constant(1)

    tasks = {ASet: 2, BSet: 1}

    @task
    def my_task(self):
        print("c-1")

import random
from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 3)
    host  = "http://www.psvmc.cn"

    @task
    def index_page(self):
        self.client.get("/index.html")  # Адрес здесь необходимо исключить host часть
        self.client.get("/userlist.json")

    @task(3)
    def view_item(self):
        item_id = random.randint(1, 10000)
        self.client.get(f"/search.json?id={item_id}", name="/search.json")

    def on_start(self):
        self.client.post("/login.json", {"username":"foo", "password":"bar"})

locust

locust -f locust_files/my_locust_file.py

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