1. # -*- coding: utf-8 -*-#
    2. # file:         test_fasthttp
    3. # Author:       ShunZhe
    4. # Date:         2021/7/24
    5. import os
    6. import queue
    8. from locust import TaskSet, constant, task, events
    9. from locust.contrib.fasthttp import FastHttpUser
    10. from locust.runners import MasterRunner
    12. """
    13. 1、test_start:开始新的负载测试时在每个节点上触发。如果在测试期间用户数量发生变化，则不会再次触发
    14. 2、init事件：在初始化 locust，发生在解析locust文件之后，但在测试接口开始之前，分布式执行则每个slave初始化一次
    15. 3、test_start：当负载测试停止时在每个节点上触发
    16. """
    17. ids = queue.Queue()
    19. @events.test_start.add_listener
    20. def on_test_start(**kwargs):
    21.     print(f'A new test is starting')
    24. @events.test_stop.add_listener
    25. def on_test_stop(**kwargs):
    26.     print('A new test is ending')
    29. @events.init_command_line_parser.add_listener
    30. def add_user_parser(parser, **kwargs):
    31.     """自定义locust命令参数"""
    32.     parser.add_argument(
    33.         "-fi",
    34.         "--first_index",
    35.         help="获取截取列表首索引"
    36.              "[fc:]",
    37.         default=0,
    38.         type=str
    39.     )
    40.     parser.add_argument(
    41.         "-li",
    42.         "--last_index",
    43.         help="获取截取列表末索引"
    44.              "[:li]",
    45.         default=0,
    46.         type=str
    47.     )
    48.     args = parser.parse_args()
    49.     if args.first_index:
    50.         os.environ['first_index'] = args.first_index
    51.     if args.last_index:
    52.         os.environ['last_index'] = args.last_index
    55. @events.init.add_listener
    56. def on_locust_init(environment, **kwargs):
    57.     print("当前环境变量first_index为： %s" % os.environ.get('first_index'))
    58.     print("当前环境变量last_index为： %s" % os.environ.get('last_index'))
    59.     for i in range(10000)[int(os.environ.get('first_index')): int(os.environ.get('last_index'))]:
    60.         ids.put_nowait(i)
    61.     print('初始化测试数据完毕')
    62.     if isinstance(environment.runner, MasterRunner):
    63.         print("master节点执行")
    64.     else:
    65.         print("worker节点执行")
    68. class ApiUser(TaskSet):
    70.     def on_start(self):
    71.         """每个虚拟用户运行时触发"""
    72.         print("---------------测试开始了---------------")
    74.     def on_stop(self):
    75.         """每个虚拟用户退出时触发"""
    76.         print("---------------测试结束了----------------")
    78.     @task(1)
    79.     def test_visit_baidu(self):
    80.         print(f'当前id的值为：{ids.get()}')
    81.         with self.client.get("/web/users/details", catch_response=True, name='users/details') as response:
    82.             # print(response)
    83.             if response.status_code == 401:
    84.                 response.success()
    85.             else:
    86.                 response.failure('not 401 code')
    89. # class UserTest(HttpUser):
    90. #     host = "https://backend-test.cn"
    91. #     wait_time = constant(1)
    92. #
    93. #     tasks = {ApiUser}
    96. class UserTest(FastHttpUser):
    97.     host = "https://backend-test.cn"
    98.     wait_time = constant(1)
    99.     tasks = {ApiUser}
    102. if __name__ == "__main__":
    103.     file_name = os.path.abspath(__file__)
    104.     os.system("locust -f %s --first_index 10 --last_index 20" % file_name)