Locust入门

环境

1. python：3.7
2. locust：1.3.2

# -*- coding: utf-8 -*-
# @Author  : Lee
# @File    : demo.py

import os
import sys
# 将当前项目根目录添加到运行环境中，目的是引用自定义包
sys.path.append(os.path.join(os.getcwd(), "../.."))  # 需根据当前文件所在目录层级调整

from pathlib import Path
from locust.runners import MasterRunner # 设置主从运行时导入
from locust import HttpUser, task, between, tag, events # Locust框架相关

class LocustDemo(HttpUser):
    wait_time = between(1, 5)

    # 每启动一个用户调用一次
    def on_start(self):
        print("on_start")

    def on_stop(self):
        print("on_stop")
  
    @task
    @tag("tag", "tag1")
    def task_demo_tag1(self):
        response = self.client.post(url, params=params, data=data, catch_response=True)
        # 执行成功
        response.success()
        # 执行失败
        response.failure(“error message”)

    @task
    @tag("tag2")
    def task_demo_tag2(self):
        response = self.client.post(url, params=params, data=data, catch_response=True)
        # 执行成功
        response.success()
        # 执行失败
        response.failure(“error message”)

    @task
    @tag("tag3")
    def task_demo_tag3(self):
        response = self.client.get(url, params=params, catch_response=True)
        # 执行成功
        response.success()
        # 执行失败
        response.failure(“error message”)

    @task
    @tag("tag4")
    def task_demo_tag4(self):
        response = self.client.get(url, params=params, catch_response=True)
        # 执行成功
        response.success()
        # 执行失败
        response.failure(“error message”)

     # 测试开始前执行，只执行一次
     @events.test_start.add_listener
     def on_test_start(**kwargs):
         print("A new test is starting")
    
     # 测试结束后执行
     @events.test_stop.add_listener
     def on_test_stop(**kwargs):
         print("A new test is ending")
    
     # 设置主从时在主机上执行
     @events.init.add_listener
     def on_locust_init(environment, **kwargs):
         if isinstance(environment.runner, MasterRunner):
             print("I'm on master node")
         else:
             print("I'm on a worker or standalone node")


if __name__ == '__main__':
    res = os.system("locust -f {} --host={} --tags tag4 tag1".format(Path(__file__).name, configr.cp.get("api", "pre_url")))