Executing the auth token once per run - Locust

JD2775 0
Viewed 36

Ideally I want to grab the token one time (1 request) and then pass that token into the other 2 requests as they execute. When I run this code through Locust though...

from locust import HttpUser, constant, task
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class ProcessRequests(HttpUser):

    host = 'https://hostURL'
    wait_time = constant(1)

    def on_start(self):

        tenant_id = "tenant123"
        client_id = "client123"
        secret = "secret123"
        scope = "api://123/.default"

        body ="grant_type=client_credentials&client_id=" + client_id + "&client_secret=" + secret + "&scope=" + scope

        tokenResponse = self.client.post(
                f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token",
                body,
                headers = { "ContentType": "application/x-www-form-urlencoded"} 
            )

        response = tokenResponse.json()
        responseToken = response['access_token']
        

        self.headers = {'Authorization': 'Bearer ' + responseToken}


    @task
    def get_labware(self):
        self.client.get("/123", name="Labware",headers=self.headers)

    @task
    def get_instruments(self):
        self.client.get("/456", name="Instruments", headers=self.headers)

It ends up firing off multiple token requests that don't stop..

enter image description here

Any ideas how to fix this to make the token only run once?

locust load-testing python
1 Answers

In your case it runs once per user so my expectation is that you spawned 24 users and the number of Labware and Instruments is at least 2x times higher so it seems to work exactly according to the documentation.

Users (and TaskSets) can declare an on_start method and/or on_stop method. A User will call its on_start method when it starts running, and its on_stop method when it stops running. For a TaskSet, the on_start method is called when a simulated user starts executing that TaskSet, and on_stop is called when the simulated user stops executing that TaskSet (when interrupt() is called, or the user is killed).

If you want to get the token only once and then share it across all the virtual users you can go for the workaround from this question

@events.test_start.add_listener
def _(environment, **kwargs):
    global token
    token = get_token(environment.host)

and put into this get_token() function what you have in on_start() one.

More information: Locust with Python: Introduction, Correlating Variables, and Basic Assertions

Dmitri T0
Related