Locust 快速上手指南

    import random
    from locust import HttpUser, task, between, SequentialTaskSet, tag
    class MyTaskCase(SequentialTaskSet):
            # 初始化方法，相当于 setup
            def on_start(self):
                pass
    
        # @task python中的装饰器，告诉下面的方法是一个任务，
        # 这个装饰器和下面的方法被复制多次，改动一下，就能写出多个接口
        # 装饰器后面带上(数字)代表在所有任务中，执行比例
        # 要用这个装饰器，需要头部引入 从locust中，引入 task
        @task
        @tag("leave_1")
        def regist_(self):  # 一个方法， 方法名称可以自己改
            url = '/erp/regist'  # 接口请求的URL地址
            # 定义请求头为类变量，这样其他任务也可以调用该变量
            self.headers = {"Content-Type": "application/json"}
            self.username = "locust_" + str(random.randint(10000, 100000))
            self.pwd = '1234567890'
            # post请求的 请求体
            data = {"name": self.username, "pwd": self.pwd}
            # 使用self.client发起请求，请求的方法根据接口实际选,
            # catch_response 值为True 允许为失败 ， 
            # name 设置任务标签名称   -----可选参数
            with self.client.post(url,
                                  json=data,
                                  headers=self.headers,
                                  catch_response=True) as rsp:
                if rsp.status_code > 400:
                    print(rsp.text)
                    rsp.failure('regist_ 接口失败！')
    
        @task  # 装饰器，说明下面是一个任务
        def login_(self):
            url = '/erp/loginIn'  # 接口请求的URL地址
            data = {"name": self.username, "pwd": self.pwd}
            with self.client.post(url,
                                  json=data,
                                  headers=self.headers,
                                  catch_response=True) as rsp:
                # 提取响应json 中的信息，定义为 类变量
                self.token = rsp.json()['token']
                if rsp.status_code < 400 \
                        and rsp.json()['code'] == "200":
                    rsp.success()
                else:
                    rsp.failure('login_ 接口失败！')
    
        @task  # 装饰器，说明下面是一个任务
        def getuser_(self):
            url = '/erp/user'  # 接口请求的URL地址
            # 引用上一个任务的 类变量值   实现参数关联
            headers = {"Token": self.token}  
           # 使用self.client发起请求，请求的方法 选择 get
            with self.client.get(url, 
                                 headers=headers, 
                                 catch_response=True) as rsp: 
                if rsp.status_code < 400:
                    rsp.success()
                else:
                    rsp.failure('getuser_ 接口失败！')
    
        # 结束方法， 相当于teardown
        def on_stop(self):
            pass
    
    # 定义一个运行类 继承HttpUser类， 所以要从locust中引入 HttpUser类
    class UserRun(HttpUser):
        tasks = [MyTaskCase]
        # 设置运行过程中间隔时间 需要从locust中 引入 between
        wait_time = between(0.1, 3)  

     #!/usr/bin/env python
    # -*- coding: utf-8 -*-
    from locust import task,TaskSet,HttpUser
    
    class UserTasks(TaskSet):
        # 声明下面是一个任务
        @task
        def getIndex(self):
            # self.client是TaskSet的成员，相当于一个request对象
            self.client.get("/path")
            print("here")
    class WebUser(HttpUser):
        # 声明执行的任务集是哪个类
        tasks = [UserTasks]
        # 最小等待时间和最大等待时间   请求间的间隔时间
        min_wait = 1000
        max_wait = 2000

     # # -- coding: utf-8 --
    from locust import task,TaskSet,HttpUser,tag
    from random import randint
    import json
    class UserTasks(TaskSet):
 
        def on_start(self):
        # 准备测试，请求接受的大多是字典格式
            self.loginData = [{"username":"1","paswordword":"1"},
                              {"username":"2","paswordword":"2"},
                              {"username":"3","paswordword":"3"}]
        print("-----------on_start----------------")
        
        # 声明下面是一个登录任务
        @task(1)
        def doLogin(self):
            ranIndex = randint(1,len(self.loginData))
            # 发送请求并响应
            response = self.client.post("/path",data = self.loginData[ranIndex],catch_response=True)
            if "login-pass" in response.text:
                response.success()
            else:
                response.failure("Can not login!")
   
        # 声明下面是一个获取商品任务
        @task
        def get_goods(self):
            body = {"good_name" :"apple"}
            # 发送请求并响应
            response = self.client.post("/path2",data = body,catch_response=True)
            newText = json.loads(response.txt)
            if "game_id" in newText[0].keys():
                response.success()
            else:
                response.failure("Can not get!")
                
		#为任务分配权重
		tasks = {doLogin:1, get_goods:2}
    
    #增加 tag 标签，在执行时，可以用 -T \ --tags 指定标签执行、-E \ --exclude-tags 排除指定标签执行
	class WebUser(User):
	    @task
    	@tag("tag1", "tag2")
    	def my_task(self):
        	pass
        
    class WebUser(HttpUser):
        # 声明执行的任务集是哪个类，任务集中的任务按已分配的1:2权重执行
        tasks = [UserTasks]
        # 最小等待时间和最大等待时间   请求间的间隔时间
        min_wait = 1000
        max_wait = 2000
        
    # locust -f locust_test.py --host=http://localhost:8082    
    # Number of total users to simulate   模拟的用户数 
    # Spawn rate (users spawned/second)   每秒产生的用户数