Load Testing OData endpoints with Python

Today I thought it would be fun to look at using an open-source framework to perform load testing on Dynamics 365 Finance and Operations OData endpoints. OData is a standard protocol for creating and consuming data in a synchronous manner. The purpose of OData is to provide a protocol that is based on Representational State Transfer (REST) for create, read, update, and delete (CRUD) operations. 

You can read more about OData here:

https://docs.microsoft.com/en-us/dynamics365/fin-ops-core/dev-itpro/data-entities/odata

We will be using a Python Library named Locust to run the load tests and a Python Library named Faker to generate data.

Read more about Locust here.

Locust

Read more about Faker here.

Faker

 A few notes before we get started.

Do not execute this against your production environment, or any environment where you do not want simulated data created. In this example, I map the Application ID to the Finance and Operation admin account. I do not recommend using the Admin account. Also, be sure to change the default company (DataAreaID) for whatever account you use to a company that has been configured and has data. The DAT company will not work.

 In this blog I will show the following:

• Create a virtual environment on Linux
• Install Locust 
• Install Faker
• Create an app registration in Azure Active Directory
• Map the app registration to a user in Finance and Operations
• Create a python script using the Locust and Faker libraries
• Executing a load test

Create a virtual environment on Linux

I used Ubuntu Linux version 18.04.5 LTS deployed in Azure. There may be small differences if you choose to attempt this on a different Linux distribution. You could install Python and other supporting libraries on Windows but the paths and screenshots will of course not match this blog. This blog is not intended to get into the setup of Python and a virtual environment.

The commands below will create a virtual environment on Linux using Python3.

Create a directory named locust.

mkdir locust

Create a virtual environment.

python3 -m venv locust/

Now navigate to the locust directory.

cd locust

Activate the virtual environment

source bin/activate

Create a directory for our script.

mkdir D365

Change into the D365 directory.

cd D365

Your screen should like this where your username@hostname should be your server and host of course.

(locust) tells us we are in the virtual environment we created.

Install Locust and Faker

Use the following command to install the Locust and Faker libraries. Make sure your virtual environment is still activated from the previous step.

pip install locust

pip install faker

 *You may need to install other packages or libraries depending on the state of you environment.

You should see output like the following. 

Create Application Registration and Client Secret in Azure Active Directory

Go to Azure portal > AAD > App registrations > New registration (this section was re-used from a previous blog post. That is why you see the name of the application registration be strange for the topic)

Click on New Registration and provide a name for the application and click Register.

Once the Application ID is generated note the ID that was created from Application (client) ID in the Azure portal. You will need the Application ID later.

Click on Certificates & Secrets. Click on New client secret.

Provide a Description and an Expires on option and click Add. Take note of the client secret that is generated because you will not be able to retrieve the secret again and we will need the client secret and the Application ID to configure Finance and Operations later.

Now you should have an Application ID and a client secret in your notes.

Map the app registration to a user in Finance and Operations

 In Finance and Operations, go to Modules > System administration > Setup > Azure Active Directory applications.

Click New to add a new Azure Active Directory application. In the Client Id field, enter the Application ID that was created above. Select Admin for the User ID. Then click Save.

 Create a python script using the Locust and Faker libraries

Open an editor and create a file as follows. Make sure you are in the virtual environment that was created earlier. I named the file D365.py. First, we will start with the includes, global variables, 1 class, and an authentication method.

There are changes that will need to be made for you environment.

D365 = ‘Your D365 URL’

clientid = ‘Your Application ID from Azure Active Directory’

secret = ‘The secret your created in your Application ID in Azure Active Directory’

tenant = ‘Your tenant’

Script so far.

from locust import HttpUser, task, between, TaskSet
import time,sys,json
from faker import Faker

# variables.
#just declare these global to make things easy.
global d365,clientid,secret,tokendpoint

d365 = 'https://jaldev09017a96d09905665265devaos.cloudax.dynamics.com'
clientid = 'bb1d7169-f85a-43a1-9f23-7b99461cf0f5'
secret = '9Ngq4Gx.IRyv89BmajB1~Ta-d5yR-B83ma' 
tenant = 'jlarsondemo.onmicrosoft.com'

tokenendpoint = f'https://login.microsoftonline.com/{tenant}/oauth2/token'  #oauth token endpoint
tokenpost = {
	'client_id':clientid,
	'client_secret':secret,
	'grant_type':'client_credentials',
	'resource':d365
	}

class QuickstartUser(HttpUser):
    wait_time = between(1, 2)
	
	class QuickstartUser(HttpUser):
    wait_time = between(1, 2)
    @task
    def done(self):
        sys.exit()

    def on_start(self):
        print("logging in")
        token = self.client.post(tokenendpoint, tokenpost)
        global accesstoken
        accesstoken = token.json()['access_token']
        print(accesstoken)
        global requestheaders
        requestheaders = {
            'Authorization': 'Bearer '   accesstoken,
            'content-type': 'application/json'
        }

To test this script from the Linux shell execute

 (locust)$ locust -f D365.py --headless -u 1 –host yourdynamicsurl

 You should see the Bearer token printed to the screen and then the script will exit.

Next, we will add a method in our QuickstartUser class to read data from the customer groups entity.

New Method

@task
def readcustomergroups(self):
    print("get CustomerGroups")
    CustomerGroups = self.client.get("/data/CustomerGroups", headers=requestheaders)
    CustomerGroups_response_dict = CustomerGroups.json()
    print("CustomerGroups",CustomerGroups_response_dict)

Full script so far

from locust import HttpUser, task, between, TaskSet
import time,sys,json
from faker import Faker

# variables.
#just declare these global to make things easy.
global d365,clientid,secret,tokendpoint

d365 = 'https://jaldev09017a96d09905665265devaos.cloudax.dynamics.com'
clientid = 'bb1d7169-f85a-43a1-9f23-7b99461cf0f5'
secret = '9Ngq4Gx.IRyv89BmajB1~Ta-d5yR-B83ma' 
tenant = 'jlarsondemo.onmicrosoft.com'

tokenendpoint = f'https://login.microsoftonline.com/{tenant}/oauth2/token'  #oauth token endpoint
tokenpost = {
	'client_id':clientid,
	'client_secret':secret,
	'grant_type':'client_credentials',
	'resource':d365
	}

class QuickstartUser(HttpUser):
    wait_time = between(1, 2)
	
    @task
    def readcustomergroups(self):
        print("get CustomerGroups")
        CustomerGroups = self.client.get("/data/CustomerGroups", headers=requestheaders)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    def on_start(self):
        print("logging in")
        token = self.client.post(tokenendpoint, tokenpost)
        global accesstoken
        accesstoken = token.json()['access_token']
        print(accesstoken)
        global requestheaders
        requestheaders = {
            'Authorization': 'Bearer '   accesstoken,
            'content-type': 'application/json'
        }

The following method is removed from the script. We will not use it any longer.

    @task
    def done(self):
        sys.exit() 

Now we are going to run the script with a new flag -t 5s. This tells Locust to execute the script picking tasks for 5 seconds.

Execute the script from the virtual environment with the following command.

(locust)$ locust -f D365.py --headless -u 1 -t 5s –host yourdynamicsurl

Next, we will add a method in our QuickstartUser class to create data in Finance and Operations using the customer groups entity.

    @task
    def createcustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

Notice here we are using the Faker library for the first time. We are just going to create customer groups with a name generated from Faker.

Full script up to this point

from locust import HttpUser, task, between, TaskSet
import time,sys,json
from faker import Faker

# variables.
#just declare these global to make things easy.
global d365,clientid,secret,tokendpoint

d365 = 'https://jaldev09017a96d09905665265devaos.cloudax.dynamics.com'
clientid = 'bb1d7169-f85a-43a1-9f23-7b99461cf0f5'
secret = '9Ngq4Gx.IRyv89BmajB1~Ta-d5yR-B83ma' 
tenant = 'jlarsondemo.onmicrosoft.com'

tokenendpoint = f'https://login.microsoftonline.com/{tenant}/oauth2/token'  #oauth token endpoint
tokenpost = {
	'client_id':clientid,
	'client_secret':secret,
	'grant_type':'client_credentials',
	'resource':d365
	}

class QuickstartUser(HttpUser):
    wait_time = between(1, 2)

    @task
    def readcustomergroups(self):
        print("get CustomerGroups")
        CustomerGroups = self.client.get("/data/CustomerGroups", headers=requestheaders)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    @task
    def createcustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    def on_start(self):
        print("logging in")
        token = self.client.post(tokenendpoint, tokenpost)
        global accesstoken
        accesstoken = token.json()['access_token']
        print(accesstoken)
        global requestheaders
        requestheaders = {
            'Authorization': 'Bearer '   accesstoken,
            'content-type': 'application/json'
        }

Execute the script from the virtual environment with the following command.

(locust)$ locust -f D365.py --headless -u 1 -t 10s –host yourdynamicsurl

Finance and Operations Customer Groups prior execution.

Finance and Operations Customer Groups post execution.

Next, we will add a method in our QuickstartUser class to update data in Finance and Operations using the customer groups entity. This will create and update a customer group. The update will contain the description updated.\

    @task
    def updatecustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)
        #start the update
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description': name,
        }
        body = json.dumps({
            "Description": "updated"
        })
        print(f"Updating customer group {name}")
        CustomerGroups = self.client.patch(f"/data/CustomerGroups(dataAreaId=\'usmf\',CustomerGroupId=\'{name}\')", headers=requestheaders,data=body)

 Full script up to this point

from locust import HttpUser, task, between, TaskSet
import time,sys,json
from faker import Faker

# variables.
#just declare these global to make things easy.
global d365,clientid,secret,tokendpoint

d365 = 'https://jaldev09017a96d09905665265devaos.cloudax.dynamics.com'
clientid = 'bb1d7169-f85a-43a1-9f23-7b99461cf0f5'
secret = '9Ngq4Gx.IRyv89BmajB1~Ta-d5yR-B83ma' 
tenant = 'jlarsondemo.onmicrosoft.com'

tokenendpoint = f'https://login.microsoftonline.com/{tenant}/oauth2/token'  #oauth token endpoint
tokenpost = {
	'client_id':clientid,
	'client_secret':secret,
	'grant_type':'client_credentials',
	'resource':d365
	}

class QuickstartUser(HttpUser):
    wait_time = between(1, 2)

    @task
    def readcustomergroups(self):
        print("get CustomerGroups")
        CustomerGroups = self.client.get("/data/CustomerGroups", headers=requestheaders)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    @task
    def createcustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    @task
    def updatecustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)
        #start the update
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description': name,
        }
        body = json.dumps({
            "Description": "updated"
        })
        print(f"Updating customer group {name}")
        CustomerGroups = self.client.patch(f"/data/CustomerGroups(dataAreaId=\'usmf\',CustomerGroupId=\'{name}\')", headers=requestheaders,data=body)

    def on_start(self):
        print("logging in")
        token = self.client.post(tokenendpoint, tokenpost)
        global accesstoken
        accesstoken = token.json()['access_token']
        print(accesstoken)
        global requestheaders
        requestheaders = {
            'Authorization': 'Bearer '   accesstoken,
            'content-type': 'application/json'
        }

Execute the script from the virtual environment with the following command. Notice the time was increased to 20 seconds.

(locust)$ locust -f D365.py --headless -u 1 -t 20s –host yourdynamicsurl

Customer Groups in Finance and Operations with updates.

Next, we will add a method in our QuickstartUser class to delete data in Finance and Operations using the customer groups entity.

    @task
    def deletecustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)
        self.client.delete(f"/data/CustomerGroups(dataAreaId=\'usmf\',CustomerGroupId=\'{name}\')", headers=requestheaders)
        print("deleted %s" % (name))

Execute the script from the virtual environment with the following command. Notice the time was increased to 20 seconds.

(locust)$ locust -f D365.py --headless -u 1 -t 20s –host yourdynamicsurl

Deletes are hard to grab a screenshot of .

Full Script

from locust import HttpUser, task, between, TaskSet
import time,sys,json
from faker import Faker

# variables.
#just declare these global to make things easy.
global d365,clientid,secret,tokendpoint

d365 = 'https://jaldev09017a96d09905665265devaos.cloudax.dynamics.com'
clientid = 'bb1d7169-f85a-43a1-9f23-7b99461cf0f5'
secret = '9Ngq4Gx.IRyv89BmajB1~Ta-d5yR-B83ma' 
tenant = 'jlarsondemo.onmicrosoft.com'

tokenendpoint = f'https://login.microsoftonline.com/{tenant}/oauth2/token'  #oauth token endpoint
tokenpost = {
	'client_id':clientid,
	'client_secret':secret,
	'grant_type':'client_credentials',
	'resource':d365
	}

class QuickstartUser(HttpUser):
    wait_time = between(1, 2)

    @task
    def readcustomergroups(self):
        print("get CustomerGroups")
        CustomerGroups = self.client.get("/data/CustomerGroups", headers=requestheaders)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    @task
    def createcustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)

    @task
    def updatecustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)
        #start the update
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description': name,
        }
        body = json.dumps({
            "Description": "updated"
        })
        print(f"Updating customer group {name}")
        CustomerGroups = self.client.patch(f"/data/CustomerGroups(dataAreaId=\'usmf\',CustomerGroupId=\'{name}\')", headers=requestheaders,data=body)

    @task
    def deletecustomergroup(self):
        faker = Faker()
        name = faker.first_name()
        print("add CustomerGroups")
        CustomerGroup = {
            'dataAreaId':'usmf',
            'CustomerGroupId':name,
            'Description':name
        }
        CustomerGroups = self.client.post("/data/CustomerGroups", headers=requestheaders, json=CustomerGroup)
        CustomerGroups_response_dict = CustomerGroups.json()
        print("CustomerGroups",CustomerGroups_response_dict)
        self.client.delete(f"/data/CustomerGroups(dataAreaId=\'usmf\',CustomerGroupId=\'{name}\')", headers=requestheaders)
        print("deleted %s" % (name))

    def on_start(self):
        print("logging in")
        token = self.client.post(tokenendpoint, tokenpost)
        global accesstoken
        accesstoken = token.json()['access_token']
        print(accesstoken)
        global requestheaders
        requestheaders = {
            'Authorization': 'Bearer '   accesstoken,
            'content-type': 'application/json'
        }

Locust also has a web interface we can use to run tests. If I execute the python script with the following command

 (locust)$ locust -f D365.py -u 1 --web-port PORTNUMBER --host yourdynamicsurl

Now open a browser to the IP address and port of your host running locust. 

*if you are using IASS in Azure you may need to create a new inbound port rule.

Click Start Swarming

Notice now we can see Statistics, Charts, Failures, Exceptions, and Download Data.

If you want to learn more about Finance and Operations Integrations which is what we are using to perform these tests please reach out to your account representative at Microsoft. The Dynamics 365 Finance and Operation team provides a 3 day workshop named Dynamics 365 for Finance and Operations Integration.