Python3中性能測試工具Locust安裝使用

時間 2019-11-15

標籤 python3 python 性能測試工具 locust 安裝使用欄目 Python 简体版

原文原文鏈接

Locust安裝使用：

安裝： html

python3中 ---> pip3 install locust python

驗證是否安裝成功---> 終端中輸入 locust --help 顯示幫助信息表示安裝成功linux

locust官網 ---> https://www.locust.io/git

官網幫助文檔 ---> https://docs.locust.io/en/latest/installation.htmlgithub

大併發量測試時，建議在linux系統下進行；web

啓動：算法

終端中--->進入到代碼目錄： locust -f xxxoo.py --host=xxxxx.com json

-f 指定性能測試腳本文件
-host 被測試應用的URL地址【若是不填寫，讀取繼承（HttpLocust）類中定義的host】
經過瀏覽器訪問：http://localhost:8089(Locust啓動網絡監控器，默認爲端口號爲:8089）

Number of users to simulate 設置虛擬用戶數api

Hatch rate（users spawned/second）每秒產生(啓動）的虛擬用戶數，點擊Start swarming 按鈕，開始運行性能測試。瀏覽器

no-web模式運行啓動

終端中-->進入代碼目錄：>> locust -f xxoo.py --no-web -c10 -r2 -t 1m

啓動參數：

--no-web 表示不使用web界面運行測試。 -c 設置虛擬用戶數。 -r 設置每秒啓動虛擬用戶數。 -t 設置運行時間.。

no-web模式運行將測試結果保存到當前.py目錄中：locust -f xxoo.py --csv=起一個名字

例如：

locust -f test3.py --csv=foobar --no-web -c2 -t10s

分佈式壓測：

主從機中必須運行相同的測試代碼（把主機中代碼複製一份到多個從機中），主機負責收集測試數據，從機進行施壓測試；

在主機終端中-->進入代碼目錄：>> locust -f xxxoo.py --master

從機中終端中-->進入代碼目錄：>> locust -f xxxoo.py --slave --master-host=主機ip

分佈式壓測no-web模式保存結果到主機中當前運行.py的目錄中：>>locust -f test2.py --csv=foobartt --no-web -c2 -t10s --master

locust --help 查看幫助信息

概述

Locust寓意蝗蟲，蝗蟲過境，寸草不生；而Locust工具生成併發請求就和一大羣蝗蟲通常，向咱們的被測系統發起攻擊，以此測試系統在高併發壓力下是否能正常運轉。

Locust測試框架中，採用python進行開發，對常見的http(s)協議的系統，Locust採用request庫做爲客戶端，在發請求時和request庫使用方法同樣。

在模擬併發時，Locust採用協程、非阻塞IO來實現網絡層的併發請求，所以單臺壓力機也能產生數千併發請求，再加上對分佈式運行的支持，Locust能在使用較少壓力機的前提下支持極高的併發數測試。

實例腳本

僞代碼：

from locust import HttpLocust, TaskSet, task class WebsiteTasks(TaskSet): def on_start(self): #進行初始化的工做，每一個Locust用戶開始作的第一件事 payload = { "username": "test_user", "password": "123456", } header = { "User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36", } self.client.post("/login",data=payload,headers=header)#self.client屬性使用Python request庫的全部方法，調用和使用方法和requests徹底一致； @task(5) #經過@task()裝飾的方法爲一個事務，方法的參數用於指定該行爲的執行權重，參數越大每次被虛擬用戶執行的機率越高，默認爲1 def index(self): self.client.get("/") @task(1) def about(self): self.client.get("/about/") class WebsiteUser(HttpLocust): host = "https://github.com/" #被測系統的host，在終端中啓動locust時沒有指定--host參數時纔會用到 task_set = WebsiteTasks #TaskSet類，該類定義用戶任務信息，必填。這裏就是:WebsiteTasks類名,由於該類繼承TaskSet； min_wait = 5000 #每一個用戶執行兩個任務間隔時間的上下限（毫秒）,具體數值在上下限中隨機取值，若不指定默認間隔時間固定爲1秒 max_wait = 15000

僞代碼中對https://github.com/網站的測試場景，先模擬用戶登陸系統，而後隨機訪問首頁/和/about/，請求比例5:1，而且在測試過程當中，兩次請求的間隔時間1-5秒的隨機值；

on_start方法，在正式執行測試前執行一次，主要用於完成一些初始化的工做，例如登陸操做；

WebsiteTasks類中如何去調用 WebsiteUser(HttpLocust)類中定義的字段和方法呢？

經過在WebsiteTasks類中self.locust.xxoo xxoo就是咱們在WebsiteUser類中定義的字段或方法；

僞代碼：

from locust import HttpLocust, TaskSet, task
import hashlib
import queue

class WebsiteTasks(TaskSet):

    @task(5)
    def index(self):
        data = self.locust.user_data_queue  #獲取WebsiteUser裏面定義的ser_data_queue隊列
        md5_data=self.locust.md5_encryption() #獲取WebsiteUser裏面定義的md5_encryption（）方法
        self.client.get("/")

class WebsiteUser(HttpLocust):
    host     = "https://github.com/"
    task_set = WebsiteTasks
    min_wait = 5000
    max_wait = 15000
    user_data_queue = queue.Queue()

    def md5_encryption(self,star):
         '''md5加密方法'''
         obj    = hashlib.md5()
         obj.update(bytes(star,encoding="utf-8"))
         result = obj.hexdigest()
         return result

僞代碼中測試場景如何表達？

代碼主要包含兩個類：

WebsiteUser繼承（HttpLocust，而HttpLocust繼承自Locust）
WebsiteTasks繼承（TaskSet）

在Locust測試腳本中，全部業務測試場景都是在Locust和TaskSet兩個類的繼承子類中進行描述；

簡單說：Locust類就相似一羣蝗蟲，而每隻蝗蟲就是一個類的實例。TaskSet類就相似蝗蟲的大腦，控制蝗蟲的具體行爲，即實際業務場景測試對應的任務集；

源碼中：class Locust(object)和class HttpLocust(Locust)

class Locust(object):
    """
    Represents a "user" which is to be hatched and attack the system that is to be load tested.
    
    The behaviour of this user is defined by the task_set attribute, which should point to a 
    :py:class:`TaskSet <locust.core.TaskSet>` class.
    
    This class should usually be subclassed by a class that defines some kind of client. For 
    example when load testing an HTTP system, you probably want to use the 
    :py:class:`HttpLocust <locust.core.HttpLocust>` class.
    """
    
    host = None
    """Base hostname to swarm. i.e: http://127.0.0.1:1234"""
    
    min_wait = 1000
    """Minimum waiting time between the execution of locust tasks"""
    
    max_wait = 1000
    """Maximum waiting time between the execution of locust tasks"""
    
    task_set = None
    """TaskSet class that defines the execution behaviour of this locust"""
    
    stop_timeout = None
    """Number of seconds after which the Locust will die. If None it won't timeout."""

    weight = 10
    """Probability of locust being chosen. The higher the weight, the greater is the chance of it being chosen."""
        
    client = NoClientWarningRaiser()
    _catch_exceptions = True
    
    def __init__(self):
        super(Locust, self).__init__()
    
    def run(self):
        try:
            self.task_set(self).run()
        except StopLocust:
            pass
        except (RescheduleTask, RescheduleTaskImmediately) as e:

class HttpLocust(Locust):
    """
    Represents an HTTP "user" which is to be hatched and attack the system that is to be load tested.
    
    The behaviour of this user is defined by the task_set attribute, which should point to a 
    :py:class:`TaskSet <locust.core.TaskSet>` class.
    
    This class creates a *client* attribute on instantiation which is an HTTP client with support 
    for keeping a user session between requests.
    """
    
    client = None
    """
    Instance of HttpSession that is created upon instantiation of Locust. 
    The client support cookies, and therefore keeps the session between HTTP requests.
    """
    def __init__(self):
           super(HttpLocust, self).__init__()
           if self.host is None:
               raise LocustError("You must specify the base host. Either in the host attribute in the Locust class, or on the command line using the --host option.")
      self.client = HttpSession(base_url=self.host)

在Locust類中，靜態字段client即客戶端的請求方法，這裏的client字段沒有綁定客戶端請求方法，所以在使用Locust時，須要先繼承Locust類class HttpLocust（Locust），而後在self.client = HttpSession(base_url=self.host)綁定客戶端請求方法；

對於常見的HTTP(s)協議，Locust已經實現了HttpLocust類，其self.client=HttpSession(base_url=self.host)，而HttpSession繼承自requests.Session。所以在測試HTTP（s）的Locust腳本中，能夠經過client屬性來使用Python requests庫的所有方法，調用方式與 reqeusts徹底一致。另外，因爲requests.Session的使用，client的方法調用之間就自動具備了狀態記憶功能。常見的場景就是，在登陸系統後能夠維持登陸狀態的Session，從然後續HTTP請求操做都能帶上登陸狀態；

Locust類中，除了client屬性，還有幾個屬性須要關注：

task_set ---> 指向一個TaskSet類，TaskSet類定義了用戶的任務信息，該靜態字段爲必填；
max_wait/min_wait ---> 每一個用戶執行兩個任務間隔的上下限（毫秒），具體數值在上下限中隨機取值，若不指定則默認間隔時間爲1秒；
host --->被測試系統的host，當在終端中啓動locust時沒有指定--host參數時纔會用到；
weight--->同時運行多個Locust類時，用於控制不一樣類型的任務執行權重；

Locust流程，測試開始後，每一個虛擬用戶(Locust實例）運行邏輯都會遵照以下規律：

先執行WebsiteTasks中的on_start（只執行一次），做爲初始化；
從WebsiteTasks中隨機挑選（若是定義了任務間的權重關係，那麼就按照權重關係隨機挑選）一個任務執行；
根據Locust類中min_wait和max_wait定義的間隔時間範圍（若是TaskSet類中也定義了min_wait或者max_wait，以TaskSet中的優先），在時間範圍中隨機取一個值，休眠等待；
重複2~3步驟，直到測試任務終止；

class TaskSet

TaskSet類實現了虛擬用戶所執行任務的調度算法，包括規劃任務執行順序（schedule_task）、挑選下一個任務（execute_next_task）、執行任務（execute_task）、休眠等待（wait）、中斷控制（interrupt）等待。在此基礎上，就能夠在TaskSet子類中採用很是簡潔的方式來描述虛擬用戶的業務測試場景，對虛擬用戶的全部行爲進行組織和描述，並能夠對不一樣任務的權重進行配置。

@task

經過@task()裝飾的方法爲一個事務。方法的參數用於指定該行爲的執行權重。參數越大每次被虛擬用戶執行的機率越高。若是不設置默認爲1。

TaskSet子類中定義任務信息時，採起兩種方式：@task裝飾器和tasks屬性。

採用@task裝飾器定義任務信息時：

from locust import TaskSet, task class UserBehavior(TaskSet): @task(1) def test_job1(self): self.client.get('/test1') @task(3) def test_job2(self): self.client.get('/test2')

採用tasks屬性定義任務信息時

from locust import TaskSet def test_job1(obj): obj.client.get('/test1') def test_job2(obj): obj.client.get('/test2') class UserBehavior(TaskSet): tasks = {test_job1:1, test_job2:3} # tasks = [(test_job1,1), (test_job1,3)] # 兩種方式等價

上面兩種定義任務信息方式中，均設置了權重屬性，即執行test_job2的頻率是test_job1的兩倍。

若不指定，默認比例爲1:1。

關聯

在某些請求中，須要攜帶以前response中提取的參數，常見場景就是session_id。Python中可用經過re正則匹配，對於返回的html頁面，可用採用lxml庫來定位獲取須要的參數；

from locust import HttpLocust, TaskSet, task from lxml import etree class WebsiteTasks(TaskSet): def get_session(self,html): #關聯例子 tages = etree.HTML(html) return tages.xpath("//div[@class='btnbox']/input[@name='session']/@value")[0] def on_start(self): html = self.client.get('/index') session = self.get_session(html.text) payload = { "username": "test_user", "password": "123456", 'session' : session } header = { "User-Agent":"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36", } self.client.post("/login",data=payload,headers=header) @task(5) def index(self): self.client.get("/")
  assert response['ErrorCode']==0 #斷言 @task(1) def about(self): self.client.get("/about/") class WebsiteUser(HttpLocust): host = "https://github.com/" task_set = WebsiteTasks min_wait = 5000 max_wait = 15000

參數化

循環取數據，數據可重複使用

例如：模擬3個用戶併發請求網頁，共有100個URL地址，每一個虛擬用戶都會依次循環加載100個URL地址

#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = 'tian'
__data__ = '2019/8/9 11:02'

from locust import TaskSet,HttpLocust,task

class UserBehavior(TaskSet):
    '''併發用戶能夠重複使用數據'''
    def on_start(self):
        self.index = 0

    @task
    def test_visit(self):
        url = self.locust.share_data[self.index]
        print("這裏是visit url:{0}".format(url))
        self.index = (self.index +1)%len(self.locust.share_data)
        print("這index是多少啊；",self.index)
        self.client.get(url)

class WebsiteUser(HttpLocust):
     host = 'http://debugtalk.com'
     task_set = UserBehavior
     share_data = ['url1','url2','url3','url4','url5']
     min_wait = 1000
     max_wait = 3000

保證併發測試數據惟一性，不循環取數據；

全部併發虛擬用戶共享同一份測試數據，而且保證虛擬用戶使用的數據不重複；

例如：模擬3用戶併發註冊帳號，共有9個帳號，要求註冊帳號不重複，註冊完畢後結束測試：

採用隊列

from locust import TaskSet, task, HttpLocust
import queue
class UserBehavior(TaskSet):
    @task
    def test_register(self):
        try:
            data = self.locust.user_data_queue.get()
        except queue.Empty:
            print('account data run out, test ended.')
            exit(0)
        print('register with user: {}, pwd: {}'\
            .format(data['username'], data['password']))
        payload = {
            'username': data['username'],
            'password': data['password']
        }
        self.client.post('/register', data=payload)
class WebsiteUser(HttpLocust):
    host = 'http://debugtalk.com'
    task_set = UserBehavior
    user_data_queue = queue.Queue()
    for index in range(100):
        data = {
            "username": "test%04d" % index,
            "password": "pwd%04d" % index,
            "email": "test%04d@debugtalk.test" % index,
            "phone": "186%08d" % index,
        }
        user_data_queue.put_nowait(data)
    min_wait = 1000
    max_wait = 3000

保證併發測試數據惟一性，循環取數據；

全部併發虛擬用戶共享同一份測試數據，保證併發虛擬用戶使用的數據不重複，而且數據可循環重複使用；

例如：模擬3個用戶併發登陸帳號，總共有9個帳號，要求併發登陸帳號不相同，但數據可循環使用；

from locust import TaskSet, task, HttpLocust
import queue
class UserBehavior(TaskSet):
    @task
    def test_register(self):
        try:
            data = self.locust.user_data_queue.get()
        except queue.Empty:
            print('account data run out, test ended')
            exit(0)
        print('register with user: {0}, pwd: {1}' .format(data['username'], data['password']))
        payload = {
            'username': data['username'],
            'password': data['password']
        }
        self.client.post('/register', data=payload)
        self.locust.user_data_queue.put_nowait(data)
class WebsiteUser(HttpLocust):
    host = 'http://debugtalk.com'
    task_set = UserBehavior
    user_data_queue = queue.Queue()
    for index in range(100):
        data = {
            "username": "test%04d" % index,
            "password": "pwd%04d" % index,
            "email": "test%04d@debugtalk.test" % index,
            "phone": "186%08d" % index,
        }
        user_data_queue.put_nowait(data)
    min_wait = 1000
    max_wait = 3000

斷言（即檢查點）

經過with self.client.get("url地址",catch_response=True) as response的形式；

response.status_code獲取http響應碼進行判斷，失敗後會加到統計錯誤表中；

python自帶的斷言assert失敗後代碼就不會向下走，且失敗後不會被Locust報表統計進去；

默認不寫參數catch_response=False斷言無效，將catch_response=True才生效；

下面例子中：

首先使用python斷言對接口返回值進行判斷（python斷言不經過，代碼就不向下執行，get請求數爲0），經過後對該接口的http響應是否爲200進行判斷；

@task def all_interface(self): #豆瓣圖書api爲例子 with self.client.get("https://api.douban.com/v2/book/1220562",name="/LhcActivity/GetActConfig",catch_response=True) as response:
 assert response.json()['rating']['max']==10 #python斷言對接口返回值中的max字段進行斷言 if response.status_code ==200: #對http響應碼是否200進行判斷 response.success() else: response.failure("GetActConfig[Failed!]")