Python locust 压力测试代码

我就是我 2022-03-18 14:22 261阅读 0赞

#!/usr/bin/env python
    # encoding: utf-8
    """
    @author: morgan
    @time: 2/21/19 9:53 AM
    """
    import datetime
    import time
    from _decimal import Decimal
     
    from bson import ObjectId
    from locust import HttpLocust, TaskSet, task
    import subprocess
    import json
     
     
    def __default(obj):
        if isinstance(obj, datetime.datetime):
            try:
                return int(time.mktime(obj.timetuple()))
            except:
                return 0
        elif isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        elif isinstance(obj, Decimal):
            return float(obj)
        elif isinstance(obj, ObjectId):
            return str(obj)
        else:
            raise TypeError('%r is not JSON serializable' % obj)
     
     
    def db_to_json(obj):
        json_str = json.dumps(obj, ensure_ascii=False, default=__default).encode('utf-8')
        # json_str = dumps(obj, ensure_ascii = False).encode('utf-8')
        return json_str
     
     
    # 性能测试任务类 TaskSet.
    class UserBehavior(TaskSet):
        # 开始
        def on_start(self):
            pass
     
        # 任务
        @task(1)
        def getTagVals(self):
            u"""
            request_url：请求路径
            request_params：请求头参数
            request_json：请求json参数
            """
            secret = "18851057d710a7f767a84233905079c5fd1bded8"
            request_url = "/wecastsvr"
            request_params = {
                "functioncode": "add_question",
                "team_id": "59597468ed8000f104304c2594999f2264fa205f",
                "title": "我是帅哥" + str(time.time()),
                "question_type": "question_type",
                "creator": "45456",
                "option_list": db_to_json(
                    [
                        {"title": "选项1", "answer": 1},
                        {"title": "选项2", "answer": 0},
                        {"title": "新的选项", "answer": 0}
                    ]
                ),
                "target": "CLNJ01",
                "timestamp": 1507860000,
                "apiId": "EC",
                "apiSign": "D41D8CD98F00B204E9800998ECF8427E",
                "api_key": "dd6se5157292",
            }
     
            request_json = {
                "tagKey": 25
            }
            response = self.client.post(
                url=request_url,
                params=request_params,
                json=request_json
            )
            # sign = get_data_sign(request_params, secret).lower()
            if response.status_code != 200:
                print('返回异常')
                print("请求返回状态码:", response.status_code)
            elif response.status_code == 200:
                print('返回正常')
                print(response)
     
            # 这里可以编写自己需要校验的返回内容
            # content = json.loads(response.content)["content"]
            # if content["tagKey"] == 25:
            #     print u"校验成功"
            #     print json.dumps(content, encoding="UTF-8", ensure_ascii=False)
     
     
    # 性能测试配置
    class MobileUserLocust(HttpLocust):
        u"""
        min_wait ：用户执行任务之间等待时间的下界，单位：毫秒。
        max_wait ：用户执行任务之间等待时间的上界，单位：毫秒。
        """
        # weight = 3
        task_set = UserBehavior
        host = "http://127.0.0.1:8080"
        min_wait = 3000
        max_wait = 6000
     
     
    if __name__ == "__main__":
        subprocess.Popen("locust -f main_pt.py", shell=True)

python main\_pt.py

运行成功后即可访问 locust的 web UI界面 ：[http://localhost:8089][http_localhost_8089]

设置即可

[http_localhost_8089]: http://localhost:8089