CODEWITHSUNDEEP
python-3.xlocust
r0ckyte
r0ckyte

Reputation: 23

Locust load testing For API calls

I have written a code for locust load testing for my case, where i can do a token call and then will make feature calls as per below code.

This is working fine with single token and 'n' number of users mentioned in master

I made a token call outside class and sending it as parameter to ENV. The userclass is reading the single token and using the same for all users.

I do not want to make token call inside the class, as it generates new token in every execution.

I'm looking if there is anyway like making token calls based on number of users mentioned on master -u and using only those tokens in User class.

Please suggest me if there is any documentation pointer which i can refer for this usecase

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

Upvotes: 2

Views: 1304

Answers (1)

Muhammed Tanır
Muhammed Tanır

Reputation: 197

You can generate tokens in User class's on_start method so each user generates a new token when spawning.

class MyUser(User):


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

there is a drawback to this though, if your user count is more than your token generating service can handle some users will not be able to spawn, the way I do in my tests is if token generating part is not a part of the system I am testing, I generate tokens beforehand and write it in some file or some external db and read them from there.

Upvotes: 2

Related Questions