python-3.x - Locust 负载测试 API 调用

Question

我已经为我的案例编写了一个用于 locust 负载测试的代码，我可以在其中进行 token 调用，然后按照以下代码进行功能调用。
这适用于单个 token 和 master 中提到的“n”个用户
我在类外进行了 token 调用，并将其作为参数发送给 ENV。用户类正在读取单个 token 并为所有用户使用相同的 token 。
我不想在类内进行 token 调用，因为它会在每次执行时生成新 token 。
我正在寻找是否有类似根据 master -u 上提到的用户数量进行 token 调用并仅在 User 类中使用这些 token 的方法。
请建议我是否有任何文档指针可供我引用此用例

#! /usr/bin/python3.6


import json
from locust import HttpUser, task, constant, tag, events
from locust.log import setup_logging
import os
from datetime import datetime

import requests

setup_logging("INFO", None)


@events.init_command_line_parser.add_listener
def init_parser(parser):
    parser.add_argument("--resp-delay", type=str, env_var="LOCUST_RESP_DELAY", default="", help="It's working")
    parser.add_argument("--resp-size", type=str, env_var="LOCUST_RESP_SIZE", default="", help="It's working")
    parser.add_argument("--env-endpoint", type=str, env_var="LOCUST_ENV_ENDPOINT", default="", help="It's working")


@events.init.add_listener
def _(environment, **kw):

    os.environ['resp-delay'] = environment.parsed_options.resp_delay
    os.environ['resp-size'] = environment.parsed_options.resp_size
    os.environ['env-endpoint'] = environment.parsed_options.env_endpoint

    with open("resources/data/" + environment.parsed_options.env_endpoint + '/data.json') as f:
        data = json.load(f)

    cal_transaction_id = datetime.now().strftime('%Y%m%dT%H%M%S')

    #retrive cliend id and client secret from bitbucket repo
    dict_car_app_all = data["data"]

    print("env-endpoint:" + os.environ.get("env-endpoint"))

    token_url = "https://ENDPOINT/auth/token"

    #retrive cliend id and client secret from bitbucket repo
    token_payload = "client_id=" + dict_car_app_all[0]["client_id"] + "&client_secret=" + dict_car_app_all[0]["client_secret"]

    token_headers = {
             'Content-Type': 'application/x-www-form-urlencoded'
    }

    response = requests.request("POST", token_url, data=token_payload, headers=token_headers,
                                    cert=( 'resources/certs/' + environment.parsed_options.env_endpoint + '/MyCERT.pem',
                                          'resources/certs/' + environment.parsed_options.env_endpoint + '/MYCERT.key'))

    result = json.loads(response.text)
    token = result["access_token"]

    os.environ['access_token'] = token
    os.environ['cal_transaction_id'] = cal_transaction_id


#class User_1(User):
class User_0(HttpUser):
    wait_time = constant(1)
    host = "host.com"

    @tag('tag1')
    @task
    def load_test_api_tag1(self):
        token_0 = os.environ.get('access_token')
        cal_transaction_id = os.environ.get('cal_transaction_id')
        env_endpoint = os.environ.get('env-endpoint')


        resp_delay = os.environ.get("resp-delay")
        resp_size = os.environ.get("resp-size")

        feature_headers = {
            'Authorization': "Bearer " + str(token_0),
            'sm_transactionID': cal_transaction_id
        }

        url = "https://ENDPOINT/SERVICE/mytestservice/first_test"

        querystring = {"response_delay": resp_delay, "data_size": resp_size}

        self.client.request("GET", url, headers=feature_headers, params=querystring,
                        cert = ('resources/certs/' + env_endpoint + '/MyCERT.pem',
                                 'resources/certs/' + env_endpoint + '/MyCERT.key'))

Answer 1

您可以在 User 类的 on_start 方法中生成 token ，以便每个用户在生成时生成一个新 token 。

class MyUser(User):


def on_start(self):
    #generate token here and assign an instance variable like self.token=abc
    super().on_start()

但是，这有一个缺点，如果您的用户数超过您的 token 生成服务可以处理的一些用户将无法生成，我在测试中所做的方式是如果 token 生成部分不是系统的一部分我正在测试中，我预先生成 token 并将其写入某个文件或某个外部数据库中，然后从那里读取它们。