python Locust个人学习记录

Locust学习记录

一、什么是locust？
- （一）安装Locust
二、Locust练习
- （一）登录功能性能测试脚本
- （二）注册功能性能测试脚本
- （三）查询功能性能测试脚本
- （四）查询个人信息功能性能场景测试
- （五）集合点功能的使用
- （六）多业务融合的业务场景测试
- （七）自行控制请求是fail还是success
- （八）locust命令行介绍

一、什么是locust？

Locust（蝗虫）是一个性能测试框架，可以并发生成多用户同时向我们的被测系统发出请求，就和一大群蝗虫一般，向我们的被测系统发起攻击，以此测试系统在高并发压力下是否能正常运转。且自带一个Web UI，用于定义用户模型，实时观察测试数据，错误统计等。

（一）安装Locust

（1）联网情况下：pip install locust，会自动下载相关的软件包。

（2）无网情况下：下载安装包安装（安装包可在PYPI官网进行下载https://pypi.org/）
（3）安装好后，在cmd窗口下执行locust --help 查看是否安装成功。

二、Locust练习（一）登录功能性能测试脚本

# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # 禁用安全请求警告
urllib3.disable_warnings(InsecureRequestWarning)

class MyLogin(TaskSet):
    @task(1)
    def login(self):
        # 登录时的请求url：http://127.0.0.1/iwebshop/index.php?controller=simple&action=login_act
        req = self.client.post("/iwebshop/index.php?controller=simple&action=login_act", json={"login_info": "testuser", "password": "123456"})#登录时，需要带上用户名和密码
        if req.status_code == 200:
            print("success")
        else:
            print("fails")

class websitUser(HttpLocust):
    host = "http://127.0.0.1"
    task_set = MyLogin #任务的类名
    wait_time = between(3, 6)

if __name__ == "__main__":
    import os
    os.system("locust -f main.py")#locust执行命令

启动程序后，可打开网页http://localhost:8089/，根据需要进行填写后点击Start swarming 按钮，开始运行性能测试。
Number of users to simulate：设置虚拟用户数
Hatch rate（users spawned/second）：表示产生模拟用户的速度
最后系统每秒能接受多少请求，看系统测试数据，Hatch rate并不是设置每秒请求速率，而是产生模拟用户的速率。

执行结果查看，涵盖了一些数据。如：共请求了多少次、失败次数、请求响应时间的中位数、90%的响应时间、平均响应时间、最小响应时间、最大响应时间、响应包平均大小、目前的RPS、目前的失败率。

在Charts中还有图表可供分析：

（二）注册功能性能测试脚本

from locust import HttpLocust, task, TaskSet, between
import os, re


class My_reg(TaskSet):

    @task
    def reg_test(self):
        for i in range(20, 40):#注册20个用户
            user = {
                'email': '11111{0}@qq.com'.format(i),
                'username': 'test{0}'.format(i),#从test20到test39
                'password': '123456',
                'repassword': '123456',
                'captcha': '123456'#这里我将我对应的测试网站的验证码固定设置为12346了
            }
            response = self.client.post('/index.php?controller=simple&action=reg_act',data=user)
            # print(response.content.decode('utf-8'))
            if response.content.decode('utf-8').find('恭喜') > 0:#对返回的response进行分析，当含有恭喜两个字时，表示注册成功
                print('success')
            else:
                print('fail')


class request_test(HttpLocust):
    task_set = My_reg
    host = 'http://127.0.0.1/iwebshop'
    min_wait = 10
    max_wait = 20
    wait_time = between(3, 6)


if __name__ == '__main__':
    os.system('locust -f reg_test.py')

注：对于非固定验证码的网站注册用户功能，目前本人还在探索中，后续持续更新……

（三）查询功能性能测试脚本

# coding = utf-8
from locust import HttpLocust, TaskSet, task, between
import csv
import random
import re


class MySearchTask(TaskSet):

    @task
    def search(self):
        list_type = []
        with open('TypeName.csv', 'r+') as f:#通过读取csv文件，获取查询关键字
            type_list = csv.reader(f)
            for i in type_list:
                list_type = i[0:-1][:]
                break
        # print(list_type)
        i = random.randint(1, 10)#随机选取一个关键字
        response = self.client.post('/', data={
            'controller': 'site',
            'action': 'search_list',
            'word': list_type[i]
        })
        # print(list_type[i], response.content.decode('utf-8'))
        result = re.findall(r'搜索结果（(d+)）', response.content.decode('utf-8'))
        # print(result)
        if int(result[0]) > 0:#如果搜索有结果，则搜索成功
            print('success')
        else:
            print('fail')
        pass


class MyHttp(HttpLocust):

    task_set = MySearchTask
    host = 'http://127.0.0.1/iwebshop/index.php'
    wait_time = between(3, 6)


if __name__ == '__main__':

    import os
    os.system('locust -f searchLocust.py')

上方的csv文件可以通过爬虫脚本从网站上获取，并保存在csv文件中，参考脚本如下：

# coding=utf-8
import requests
from lxml import etree
import csv


def getAllTypeName(url):
    first_xpath = '/html/body/div[1]/div[6]/div[2]/div[3]'
    response = requests.get(url)
    document = etree.HTML(response.content)
    root_xpath = document.xpath(first_xpath)
    root = etree.ElementTree(root_xpath[0])
    result_list = []
    for one in root.findall('./table/caption/a'):
        result_list.append(one.text)
    for two in root.findall('./table/tbody/tr/th/a'):
        result_list.append(two.text)
    for three in root.findall('/table/tbody/tr/td/a'):
        if three.text != "回到顶部":
            result_list.append(three.text)
    return result_list
    pass


def writeToCSV(list_str):
    with open('TypeName.csv', 'w+', newline="") as f:
        csv_file = csv.writer(f)
        csv_file.writerow(list_str)
    pass


url = 'http://127.0.0.1/iwebshop/index.php?controller=site&action=sitemap'

if __name__ == '__main__':
    result = getAllTypeName(url)
    writeToCSV(result)

（四）查询个人信息功能性能场景测试

在查询个人信息前，必须先登录用户获取到cookie值，而在locust中可以使用on_start函数，该函数会在每个locust执行最开始前执行一次。

from locust import TaskSet, HttpLocust, between, task
import requests

class MyLocustTask(TaskSet):

    def __init__(self, parent):
        super().__init__(parent)
        self.to_str_cookies = {}

    def on_start(self):
        response = self.client.post("?controller=simple&action=login_act", data={"login_info": "test1", "password": "123456"})
        cookie = response.cookies
        dict_cookies = requests.utils.dict_from_cookiejar(cookie)
        str_cookies = ''
        for key, value in dict_cookies.items():
            str_cookies = str_cookies + key + '=' + value + ';'
        self.to_str_cookies = str_cookies


    @task
    def get_personal_information(self):
        # print(self.to_str_cookies)
        headers = {'Cookie': self.to_str_cookies}
        response = self.client.post('?controller=ucenter', headers=headers)
        # print(response.content.decode('utf-8'))
        if "test1您好，欢迎您来到iwebshop购物！" in response.content.decode('utf-8'):
            print("success")
        else:
            print('fail')


class MyHttpLocust(HttpLocust):
    wait_time = between(3, 6)
    task_set = MyLocustTask
    host = 'http://127.0.0.1/iwebshop/index.php'


if __name__ == '__main__':
    import os
    os.system("locust -f loginBeforeGetPersonInformation.py")

（五）集合点功能的使用

有时测试需要让所有并发用户完成初始化后再进行压力测试。在[测试计划]中，可能会要求系统能够承受100 人同时提交数据，在构造的用户还不到100 人时，前面已经初始化好的用户会进行等待，直至达到100 人时，然后100 人同时去提交数据，从而达到测试计划中的需求。

from locust import TaskSet, HttpLocust, between, task, events
import requests
from gevent._semaphore import Semaphore
all_locusts_semaphore = Semaphore()
all_locusts_semaphore.acquire()


def on_hatch_complete(**kwargs):
    all_locusts_semaphore.release()


events.hatch_complete += on_hatch_complete
# 挂载到locust钩子函数（所有的Locust实例产生完成时触发）


class MyLocustTask(TaskSet):

    def __init__(self, parent):
        super().__init__(parent)
        self.to_str_cookies = {}

    def on_start(self):
        response = self.client.post("?controller=simple&action=login_act", data={"login_info": "test1", "password": "123456"})
        cookie = response.cookies
        dict_cookies = requests.utils.dict_from_cookiejar(cookie)
        str_cookies = ''
        for key, value in dict_cookies.items():
            str_cookies = str_cookies + key + '=' + value + ';'
        self.to_str_cookies = str_cookies
        all_locusts_semaphore.wait()#限制在所有用户准备完成前处于等待状态



    @task
    def get_personal_information(self):
        # print(self.to_str_cookies)
        headers = {'Cookie': self.to_str_cookies}
        response = self.client.post('?controller=ucenter', headers=headers)
        # print(response.content.decode('utf-8'))
        if "test1您好，欢迎您来到iwebshop购物！" in response.content.decode('utf-8'):
            print("success")
        else:
            print('fail')


class MyHttpLocust(HttpLocust):
    wait_time = between(3, 6)
    task_set = MyLocustTask
    host = 'http://127.0.0.1/iwebshop/index.php'


if __name__ == '__main__':
    import os
    os.system("locust -f loginBeforeGetPersonInformation.py")

在All locusts hatched: MyHttpLocust: 100后，才会进行查询请求。

[2022-03-25 11:20:59,685] DESKTOP-QBDUANU/INFO/locust.main: Starting web monitor at http://*:8089
[2022-03-25 11:20:59,685] DESKTOP-QBDUANU/INFO/locust.main: Starting Locust 0.13.5
[2022-03-25 11:21:41,895] DESKTOP-QBDUANU/INFO/locust.runners: Hatching and swarming 100 clients at the rate 10 clients/s...
[2022-03-25 11:21:52,759] DESKTOP-QBDUANU/INFO/locust.runners: All locusts hatched: MyHttpLocust: 100
[2022-03-25 11:21:53,082] DESKTOP-QBDUANU/ERROR/stderr: Traceback (most recent call last):

（六）多业务融合的业务场景测试

可以通过@task()方法的参数用于指定该行为的执行权重。参数越大每次被虚拟用户执行的概率越高。如果不设置默认为1。或者在tasks（ tasks = {test1:1, test:2}）中设置。数值越大权重越大，执行的概率越大。

# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # 禁用安全请求警告
urllib3.disable_warnings(InsecureRequestWarning)

class MyTest(TaskSet):
    # 访问我的博客首页
    @task(1)
    def login(self):
        # 定义请求头http://127.0.0.1/iwebshop/index.php?controller=simple&action=login_act
        req = self.client.post("/iwebshop/index.php?controller=simple&action=login_act", json={"login_info": "liaoxm", "password": "123456"})
        if req.status_code == 200:
            print("success")
        else:
            print("fails")
     
    @task(3)
    def search(self):
        list_type = []
        with open('TypeName.csv', 'r+') as f:
            type_list = csv.reader(f)
            for i in type_list:
                list_type = i[0:-1][:]
                break
        # print(list_type)
        i = random.randint(1, 10)
        response = self.client.post('/', data={
            'controller': 'site',
            'action': 'search_list',
            'word': list_type[i]
        })
        # print(list_type[i], response.content.decode('utf-8'))
        result = re.findall(r'搜索结果（(d+)）', response.content.decode('utf-8'))
        # print(result)
        if int(result[0]) > 0:
            print('success')
        else:
            print('fail')
        pass

class websitUser(HttpLocust):
    host = "http://127.0.0.1"
    task_set = MyTest
    wait_time = between(3, 6)

if __name__ == "__main__":
    import os
    os.system("locust -f main.py")

使用tasks设置：

# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # 禁用安全请求警告
urllib3.disable_warnings(InsecureRequestWarning)

class MyTest(TaskSet):
    
    def login(self):
        # 定义请求头http://127.0.0.1/iwebshop/index.php?controller=simple&action=login_act
        req = self.client.post("/iwebshop/index.php?controller=simple&action=login_act", json={"login_info": "liaoxm", "password": "123456"})
        if req.status_code == 200:
            print("success")
        else:
            print("fails")
     
    def search(self):
        list_type = []
        with open('TypeName.csv', 'r+') as f:
            type_list = csv.reader(f)
            for i in type_list:
                list_type = i[0:-1][:]
                break
        # print(list_type)
        i = random.randint(1, 10)
        response = self.client.post('/', data={
            'controller': 'site',
            'action': 'search_list',
            'word': list_type[i]
        })
        # print(list_type[i], response.content.decode('utf-8'))
        result = re.findall(r'搜索结果（(d+)）', response.content.decode('utf-8'))
        # print(result)
        if int(result[0]) > 0:
            print('success')
        else:
            print('fail')
        pass

    tasks = {login:1,search:3}##需要确认一下

class websitUser(HttpLocust):
    host = "http://127.0.0.1"
    task_set = MyTest
    wait_time = between(3, 6)

if __name__ == "__main__":
    import os
    os.system("locust -f main.py")

（七）自行控制请求是fail还是success

通过使用catch_response参数和with语句，即使响应代码正确，也可以将请求标记为失败：

with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

样例：

# coding = utf-8
from locust import HttpLocust, TaskSet, task, between, events
from gevent._semaphore import Semaphore
import csv
import re

# 创建集合点
all_locust_semaphore = Semaphore()
all_locust_semaphore.acquire()


def on_hatch_complete(**kwargs):
    all_locust_semaphore.release()

# 注册事件
events.hatch_complete += on_hatch_complete
# 挂载到locust钩子函数（所有的Locust实例产生完成时触发）


class MySearchTask(TaskSet):

    def on_start(self):
        all_locust_semaphore.wait()# 限制在所有用户准备完成前处于等待状态

    @task
    def search(self):
        with open('testReport.txt', 'a+', encoding='utf-8') as f2:
            list_type = []
            with open('TypeName.csv', 'r+') as f:
                type_list = csv.reader(f)
                for i in type_list:
                    list_type = i[0:-1][:]
                    break
            for i in list_type:
                with self.client.post('/', data={
                    'controller': 'site',
                    'action': 'search_list',
                    'word': i
                }, catch_response=True) as response:

                    result = re.findall(r'搜索结果（(d+)）', response.content.decode('utf-8'))
                    if int(result[0]) > 0:
                        f2.write(str(i) + 'successn')
                        response.success()
                    else:
                        f2.write(str(i) + 'failn')
                        response.failure('fail')


class MyHttp(HttpLocust):

    task_set = MySearchTask
    host = 'http://127.0.0.1/iwebshop/index.php'
    wait_time = between(3, 6)
    # min_wait 模拟负载的任务之间执行时的最小等待时间，单位为毫秒
    # max_wait 模拟负载的任务之间执行时的最大等待时间，单位为毫秒


if __name__ == '__main__':

    import os
    os.system('locust -f homework.py')

（八）locust命令行介绍

命令行执行
locust-f test1.py --host=https://www.baidu.com --no-web -c 10 -r 2 -t 1m
启动参数：
test1.py是脚本文件名
–no-web 表示不使用Web界面运行测试。
-c 设置虚拟用户并发数。
-r 设置每秒启动虚拟用户数执行的次数。
-t 设置设置运行时间。

命令行更多详细的介绍，可以直接执行locust --help查看。

ps：以上是个人学习的简略总结，有很多不太详细的介绍，请包含~

python Locust个人学习记录

Python相关栏目本月热门文章