您现在的位置：程式師世界 >> 編程語言 > >> 更多編程語言 >> Python

python Locust個人學習記錄

編輯：Python

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) + 'success\n')
response.success()
else:
f2.write(str(i) + 'fail\n')
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：以上是個人學習的簡略總結，有很多不太詳細的介紹，請包含~