Locust（ grasshopper ） Is a performance testing framework , It can generate multiple users simultaneously and send requests to our system under test , Just like a swarm of locusts , Attack our system under test , To test whether the system can operate normally under high concurrent pressure . With one Web UI, Used to define the user model , Observe test data in real time , Error statistics, etc .
（1） In the case of networking ：pip install locust, Relevant software packages will be downloaded automatically .
（2） Without a net ： Download the installation package to install （ The installation package can be found in PYPI Download on the official website https://pypi.org/）
（3） After installed , stay cmd Execute... Under the window locust --help Check whether the installation is successful .
# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # Disable security request warning
urllib3.disable_warnings(InsecureRequestWarning)
class MyLogin(TaskSet):
@task(1)
def login(self):
# Login request 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"})# Login time , You need to bring your user name and password
if req.status_code == 200:
print("success")
else:
print("fails")
class websitUser(HttpLocust):
host = "http://127.0.0.1"
task_set = MyLogin # Class name of the task
wait_time = between(3, 6)
if __name__ == "__main__":
import os
os.system("locust -f main.py")#locust Carry out orders
After starting the program , You can open a web page http://localhost:8089/, Click... After filling in as required Start swarming Button , Start running the performance test .
Number of users to simulate： Set the number of virtual users
Hatch rate（users spawned/second）： Represents the speed at which simulated users are generated
How many requests can the system accept per second , Look at the system test data ,Hatch rate Instead of setting the request rate per second , It is the rate at which simulated users are generated .
Check the execution result , It covers some data . Such as ： How many times have you requested 、 Number of failures 、 Median request response time 、90% Response time of 、 Mean response time 、 Minimum response time 、 Maximum response time 、 Average response packet size 、 current RPS、 Current failure rate .
stay Charts There are also charts for analysis ：
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):# register 20 Users
user = {
'email': '11111{0}@qq.com'.format(i),
'username': 'test{0}'.format(i),# from test20 To test39
'password': '123456',
'repassword': '123456',
'captcha': '123456'# Here, I set the verification code of my corresponding test website to 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(' Congratulations ') > 0:# For returned response Analyze , When the word "Congratulations" is included , Indicates successful registration
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')
notes ： For the website with non fixed verification code, register the user function , At present, I am still exploring , Continue to update later ……
# 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:# By reading the csv file , Get query keywords
type_list = csv.reader(f)
for i in type_list:
list_type = i[0:-1][:]
break
# print(list_type)
i = random.randint(1, 10)# Pick a keyword at random
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' The search results （(\d+)）', response.content.decode('utf-8'))
# print(result)
if int(result[0]) > 0:# If the search results , Then the search is successful
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')
At the top of the csv The file can be obtained from the website through the crawler script , And keep it in csv In file , The reference script is as follows ：
# 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 != " Back to the top ":
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)
Before querying personal information , You must first log in to the user to get cookie value , And in the locust Can be used in on_start function , This function will be used in every locust Execute once before the beginning of execution .
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 Hello! , Welcome to iwebshop shopping ！" 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")
Sometimes the test needs all concurrent users to complete the initialization before the stress test . stay [ test plan ] in , The system may be required to withstand 100 People submit data at the same time , The number of users in the construction is less than 100 Human hour , The previously initialized user will wait , Until it reaches 100 Human hour , then 100 People submit data at the same time , So as to meet the requirements in the test plan .
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
# Mount to locust Hook function （ be-all Locust Triggered when the instance generation is completed ）
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()# Limit waiting until all users are ready
@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 Hello! , Welcome to iwebshop shopping ！" 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")
stay All locusts hatched: MyHttpLocust: 100 after , To make a query request .
[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):
Can pass @task() Method to specify the execution weight of the behavior . The larger the parameter, the higher the probability that it will be executed by the virtual user each time . If not set to default to 1. Or in tasks（ tasks = {test1:1, test:2}） Set in . The larger the value, the greater the weight , The greater the probability of execution .
# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # Disable security request warning
urllib3.disable_warnings(InsecureRequestWarning)
class MyTest(TaskSet):
# Visit my blog home page
@task(1)
def login(self):
# Define request headers 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' The search results （(\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")
Use tasks Set up ：
# coding=utf-8
import requests
from locust import HttpLocust, TaskSet, between, task
import urllib3
from urllib3.exceptions import InsecureRequestWarning
# # Disable security request warning
urllib3.disable_warnings(InsecureRequestWarning)
class MyTest(TaskSet):
def login(self):
# Define request headers 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' The search results （(\d+)）', response.content.decode('utf-8'))
# print(result)
if int(result[0]) > 0:
print('success')
else:
print('fail')
pass
tasks = {
login:1,search:3}## It needs to be confirmed that
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")
By using catch_response Parameters and with sentence , Even if the response code is correct , You can also mark a request as failed ：
with self.client.get("/does_not_exist/", catch_response=True) as response:
if response.status_code == 404:
response.success()
Examples ：
# coding = utf-8
from locust import HttpLocust, TaskSet, task, between, events
from gevent._semaphore import Semaphore
import csv
import re
# Create assembly point
all_locust_semaphore = Semaphore()
all_locust_semaphore.acquire()
def on_hatch_complete(**kwargs):
all_locust_semaphore.release()
# Registration events
events.hatch_complete += on_hatch_complete
# Mount to locust Hook function （ be-all Locust Triggered when the instance generation is completed ）
class MySearchTask(TaskSet):
def on_start(self):
all_locust_semaphore.wait()# Limit waiting until all users are ready
@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' The search results （(\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 Minimum waiting time between tasks with simulated load , The unit is millisecond
# max_wait Maximum waiting time between tasks with simulated load , The unit is millisecond
if __name__ == '__main__':
import os
os.system('locust -f homework.py')
Command line execution
locust-f test1.py --host=https://www.baidu.com --no-web -c 10 -r 2 -t 1m
Launch parameters ：
test1.py Is the script file name
–no-web No use Web Interface run test .
-c Set virtual user concurrency .
-r Set the number of times per second that virtual users are started to execute .
-t Set the run time .
Command line for more details , Can be executed directly locust --help see .
ps： The above is a brief summary of personal learning , There are a lot of less detailed introductions , Please include ~
Python -- variable, parameter passing, anonymous, recursive function (day05)
Catalog One 、 Basic type of f
Official recommendation: there are six ways for pandas to read excel, and the correct answers are written in the source code ~ its too convenient
Hello everyone , This is Wang
