Locust 负载测试 API 次调用

Question

我已经为我的案例编写了一个 Locust 负载测试代码，我可以在其中进行令牌调用，然后将按照以下代码进行功能调用。

这适用于单个令牌和 'n' master

中提到的用户数量

我在 class 之外进行了令牌调用，并将其作为参数发送给 ENV。用户class 正在读取单个令牌并为所有用户使用相同的令牌。

我不想在 class 中进行令牌调用，因为它会在每次执行时生成新令牌。

我正在寻找是否有基于 master -u 中提到的用户数量进行令牌调用并仅使用用户 class 中的那些令牌的方法。

如果有任何我可以参考此用例的文档指针，请建议我

#! /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

您可以在用户 class 的 on_start 方法中生成令牌，这样每个用户在生成时都会生成一个新令牌。

class MyUser(User):


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

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

Locust 负载测试 API 次调用

Locust load testing For API calls

python-3.x

locust