Python API testing

Question

I have written a rudimentary code in Python:

1. That reads the API collection from Swagger/Postman and extracts the payload, link, and method
2. It creates the response with the token(Generated from the token link)
3. It records the response in a CSV.

Need help in further formatting the code in terms of

1. creating functions.

2. Reducing lines of code.

3. Performance tuning using multithreading

    from urllib.request import urlopen
    import requests
    from bs4 import BeautifulSoup
    import json
    from jsonpath_ng import jsonpath, parse
   
   
    def GenerateToken():
        link = "<link>"
        page = urlopen(link)  # Link to generate token
        contents = page.read()
        soup = BeautifulSoup(contents)
        shred = list(soup.stripped_strings)
        token = shred[2]  # gets the token
        return token
   
   
    with open("TestResults.csv", "w") as f_object:
        fields = ["API_Endpoint", "Method", "Status_Code", "Message", "Timestamp"]
        writer = csv.DictWriter(f_object, fieldnames=fields)
        writer.writeheader()
        headers = {"Authorization": "Bearer {}".format(GenerateToken())}
        with open("postman_collection.json") as f:
            data = json.load(f)
            for res in data["item"]:
   
                for d in res["request"]:
   
                    if "GET" in res["request"]["method"]:
                        payload = {}
                        url = res["request"]["url"]["raw"]
                        method = res["request"]["method"]
   
                        try:
                            response = requests.get(url, headers=headers, data=payload)
   
                            if response.status_code == 200:
                                rows1 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": "OK",
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows1)
                            # writer.writerows(rows1)
                            else:
                                rows2 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": response.text,
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows2)
                        # print(response.json().get("message"))
                        except requests.ConnectionError:
                            print("failed to connect")
   
                    elif "POST" in res["request"]["method"]:
                        url = res["request"]["url"]["raw"]
                        method = res["request"]["method"]
                        payload = res["request"]["body"]["raw"]
                        try:
                            # response = requests.request("GET",url,headers=headers, data=payload)
                            response = requests.request(
                                "POST", url, headers=headers, data=payload
                            )
                            # print(response.json())
                            if response.status_code == 200:
                                rows3 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": "OK",
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows3)
                            else:
                                rows4 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": response.text,
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows4)
                                # print(response.json().get("message"))
                        except requests.ConnectionError:
                            print("failed to connect")
   
                    else:
                        url = res["request"]["url"]["raw"]
                        method = res["request"]["method"]
                        payload = res["request"]["body"]["raw"]
                        try:
                            # response = requests.request("GET",url,headers=headers, data=payload)
                            response = requests.request(
                                "POST", url, headers=headers, data=payload
                            )
                            # print(response.json())
                            if response.status_code == 200:
                                rows5 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": "OK",
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows5)
                            else:
                                rows6 = [
                                    {
                                        "API_Endpoint": url,
                                        "Method": method,
                                        "Status_Code": response.status_code,
                                        "Message": response.text,
                                        "Timestamp": datetime.datetime.utcnow(),
                                    }
                                ]
                                writer.writerows(rows6)
                        # print(response.json().get("message"))
                        except requests.ConnectionError:
                            print("failed to connect")
    f_object.close()
   

Answer 1

There is indeed a lot of repetition in this code.

You have 3 cases: GET, POST, other. So we have:

 if "GET" in res["request"]["method"]:
     payload = {}
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
 elif "POST" in res["request"]["method"]:
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
     payload = res["request"]["body"]["raw"]
 else:
     url = res["request"]["url"]["raw"]
     method = res["request"]["method"]
     payload = res["request"]["body"]["raw"]

Now that we have an simple overview we see that the whole block can be reduced as follows:

url = res["request"]["url"]["raw"]
method = res["request"]["method"]
if "GET" in res["request"]["method"]:
    payload = {}
else:
    payload = res["request"]["body"]["raw"]

There is only a slight difference between methods: the payload.

Then we proceed to the request proper:

# GET
response = requests.get(url, headers=headers, data=payload)

# POST
response = requests.request(
 "POST", url, headers=headers, data=payload
)

# other
response = requests.request(
 "POST", url, headers=headers, data=payload
)

Again, all very similar. We can build an all-purpose statement like this:

method = res["request"]["method"]
response = requests.request(method, url, headers=headers, data=payload)

There is no need for 3 different statements since they all pass the same arguments. The only thing that is different is the method, and you can pass that as an argument too.

So now that you have a single statement to issue the request, we need to look at the result. It's a simple condition of 200 or other. Thus:

if response.status_code == 200:
    result = "OK"
else:
    result = response.text

rows = [
    {
    "API_Endpoint": url,
    "Method": method,
    "Status_Code": response.status_code,
    "Message": result,
    "Timestamp": datetime.datetime.utcnow(),
    }
]
writer.writerows(rows)

As far as I can see you are repeating the same stuff 6 times. The only thing that varies is the method + the HTTP status code and the resulting response message. That can be easily simplified.

The final f_object.close() is not required since you are using the context manager (with).

Accordingly, your code can already be shortened like this (untested but the concision is hopefully becoming perceptible):

with open("TestResults.csv", "w") as f_object:
    fields = ["API_Endpoint", "Method", "Status_Code", "Message", "Timestamp"]
    writer = csv.DictWriter(f_object, fieldnames=fields)
    writer.writeheader()
    headers = {"Authorization": "Bearer {}".format(GenerateToken())}
    with open("postman_collection.json") as f:
        data = json.load(f)
        for res in data["item"]:

            for d in res["request"]:

                # common for all methods
                url = res["request"]["url"]["raw"]
                method = res["request"]["method"]

                # empty payload for GET
                if "GET" in res["request"]["method"]:
                    payload = {}
                else:
                    payload = res["request"]["body"]["raw"]

                try:
                    response = requests.request(method, url, headers=headers, data=payload)

                    if response.status_code == 200:
                        result = "OK"
                    else:
                        result = response.text

                    rows = [
                        {
                        "API_Endpoint": url,
                        "Method": method,
                        "Status_Code": response.status_code,
                        "Message": result,
                        "Timestamp": datetime.datetime.utcnow(),
                        }
                    ]
                    writer.writerows(rows)
                    # print(response.json().get("message"))
                except requests.ConnectionError:
                    print("failed to connect")

So the code is now becoming more manageable after removing duplicate code. Basically we went from 118 lines to 43.

That block of code could fit into a function, which name should reflect the actual purpose (that we don't know). Then it would be good to pass file names as arguments, rather than have them hardcoded inside the function.

If you are going to perform repeated requests against a host then you should use requests.session to improve performance and simplifying handling of state, cookies, headers etc.

To address some of your concerns: if the intention is to perform parallel processing or multithreading, there is more than one way to do that in Python. Which is best depends on your circumstances. May I suggest one link as a tutorial: Python Multithreading and Multiprocessing Tutorial.

But you have two distinct tasks: writing to a CSV file and sending data to an API. So it makes sense to split the code in two functions. If you will be doing parallel processing, then basically you would read the JSON one item at a time and feed a "queue". So you need some kind of function that accepts a URL, with optional payload and return whatever you want, that could be the just the response message based on the HTTP status code. Perhaps you can build something around existing code if you look at the link quoted above, and adapt it to your need.

I might edit this post later to add more suggestions on that.

I also recommend the book Expert Python Programming by Tarek Jaworski & Michal Ziade if you want to get more in depth knowledge on such topics.