问API调用的Locust负载测试

EN

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