Create CXTM API Pipeline Script

In this section of the lab, you will learn about CXTM's API and how to create Python code to interact with the API. The first thing you need to do is generate your own API token. That token will then be used as a secret environment variable in GitLab that you will access in the pipeline. This is a best practice as to not expose your API token.

CXTM's API is driven based on identifiers for objects in the API. There will be multiple code functions for you to develop in order to query for your specific project ID and associated batches that you created earlier in the lab. All of this will be done dynamically based on what you pass into the code. Lastly, you will write the fucntion to drive the desired batch and a very simple backoff and retry function to control if the pipeline should pass or fail in coorelation with your test cases.

Step 1 - Navigate to CICD Settings in Your GitLab Project

In preparation to store your CXTM API token as an environment variable in GitLab, navigate to CICD settings in your GitLab project:

Click Settings
Click CI/CD

Step 2 - Navigate to CICD Variables

In the center of your screen:

1. Locate Variables
2. Click Expand

Step 3 - Launch Add Variable Pop-UP

Launch the Add Variable pop-up in preparation to add your CXTM API token as a variable:

1. Scroll down until you can see the majority of the Variables section
2. Click Add Variable

Step 4 - Fillout Add Variable Pop-Up

In the Add Variable pop-up, fillout the fields as described below:

1. Name the variable Key: CXTM_API_KEY
2. Uncheck Protect variable
3. Check Mask variable

Step 5 - Return to CXTM and Navigate to Your User Profile

In your browser, return to CXTM:

1. In the top-right, click on your user icon
2. Click Manage Profile

Step 6 - Generate API Token Key

On the right-side of your screen,

1. Click REQUEST API KEY

Step 7 - Copy CXTM API Token

After generating your CXTM API token:

1. With your mouse, click and highlight the generated API key
2. Copy the API key you have selected and highlighted with the keyboard shortcut Ctrl+c or right-click, then click Copy




3. Save this API Key in a text file for use later in the Bonus: CXTM REST API section later

Step 8 - Return to GitLab with Your CXTM API Token Copied

Return to GitLab:

1. Paste you user's CXTM generated API key into the Value field
2. Click Add Variable

Step 9 - Confirm CXTM API Key is Added as a Variable in GitLab

Placeholder

1. Confirm CXTM_API_KEY variable was added to your project repo
2. Return to home project repo page

Step 10 - Create a New Python File in Visual Studio Code For CXTM Pipeline Script

Return to your VSCode Terminal and use the code keyword to open a YAML file called cxtm-pipeline-script.py.

    
        code -r /home/cisco/CiscoLive/LTROPS-2711/tests/cxtm-pipeline-script.py
    

Step 11 - Create CXTM API Pipeline Script

The script below defines one such method of driving CXTM's REST AP, specifcally projects and batch API endpoints, using the Python requests library. Additionally, there a few other Python packages leveraged along with requests:

• argparse: used to write a simple command-line interface to accept arguments to drive specific test batches in a given project
• os: used to lookup a CXTM user's API key from an environment variable following best practices
• sys: used to interact with the Python runtime environment
• requests: used to interact directly with CXTM's REST API
• json: used as an encoder and decoder of JSON payloads and responses to/from CXTM's REST API
• time: used as part of a simple backoff and retry routine when checking state of test cases
• random: used as a random number generator as part of simple backoff and retry routine when checking state of test cases
• datetime: used for print logging the date and time information during test case execution

    
        import argparse
        import os
        import sys
        import requests
        import json
        import time
        import random
        from datetime import datetime
        
        
        requests.packages.urllib3.disable_warnings()
        
        
        CXTM_API_KEY = os.environ.get('CXTM_API_KEY')
        
        HEADERS = {
            'X-TM2-API-KEY': CXTM_API_KEY,
            'Content-Type': 'application-json',
            'Accept': '*/*'
            }
        
        FORM_HEADERS = {
            'X-TM2-API-KEY': CXTM_API_KEY,
            'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
            'Accept': '*/*'
            }
        
        
        def get_project_id(project_name):
            try:
                url = f"{CXTM_BASE_URL}/projects/writable"
                response = requests.request("GET", url, headers=HEADERS, verify=False)
                response_dict = json.loads(response.text)
                projects = response_dict['data']
                for project in projects:
                    if project['name'] == project_name:
                        project_id = project['id']
                        return project_id
            except Exception:
                print(f"Project {project_name} is invalid!")
                sys.exit(1)
        
        
        def get_batch_id(project_id, batch_name):
            try:
                url = f"{CXTM_BASE_URL}/projects/{project_id}/batches"
                response = requests.request("GET", url, headers=HEADERS, verify=False)
                response_dict = json.loads(response.text)
                batches = response_dict['data']
                for batch in batches:
                    if batch['title'] in batch_name:
                        batch_id = batch['id']
                        return batch_id
            except Exception:
                print(f"Batch {batch_name} is invalid!")
                sys.exit(1)
        
        
        def run_batch(project_id, batch_name, batch_id):
            tprint(f"Starting {batch_name} test cases...")
            print()
        
            url = f"{CXTM_BASE_URL}/batches/{batch_id}/run"
            raw_data = 'is_default_topology=true'
            response = requests.request("POST", url, headers=FORM_HEADERS, data=raw_data, verify=False)
            response_dict = json.loads(response.text)
            batch_run_id = response_dict[0]['batchrun_id']
        
            retry_with_backoff(project_id, batch_id, batch_run_id)
        
            print()
            tprint("Testing ended...")
        
        
        def tprint(msg):
            timestamp = datetime.now().strftime('%M:%S.%f')[:-3]
            print(timestamp, msg)
        
        
        def retry_with_backoff(project_id, batch_id, batch_run_id, backoff_in_seconds=30):
            url = f"{CXTM_BASE_URL}/projects/{project_id}/batches/{batch_id}/runs/{batch_run_id}"
            x = 0
            testcase_status_list = [None]
            while None in testcase_status_list:
                testcase_status_list = []
                response = requests.request("GET", url, headers=HEADERS, verify=False)
                response_dict = json.loads(response.text)
                batch_testcases = response_dict['data']
                for batch_testcase in batch_testcases:
                    testcase_status_list.append(batch_testcase['result_rollup'])
                    if batch_testcase['result_rollup'] is None:
                        tprint(f"Test case {batch_testcase['name']} is QUEUED")
                    elif batch_testcase['result_rollup'] in ('passed', 'aborted'):
                        tprint(f"Test case {batch_testcase['name']} {batch_testcase['result_rollup']} execution")
                    elif batch_testcase['result_rollup'] in ('failed', 'errored'):
                        raise Exception(f"Test case {batch_testcase['name']} {batch_testcase['result_rollup']} execution\n")
                if None in testcase_status_list:
                    sleep = (backoff_in_seconds * 2**x + random.uniform(0, 1))
                    tprint(f"Backoff: {sleep}")
                    time.sleep(sleep)
        
        
        def main():
            global CXTM_BASE_URL
        
            parser = argparse.ArgumentParser(
                description='CXTM Batch Pipeline Script',
                formatter_class=argparse.RawDescriptionHelpFormatter,
            )
            parser.add_argument('-c', '--cxtm', action='store', help='CXTM IP Address')
            parser.add_argument('-p', '--project', action='store', help='CXTM Project Name')
            parser.add_argument('-b', '--batch', action='store', help='CXTM Batch Name')
            args = parser.parse_args()
        
            CXTM_BASE_URL = f"https://{args.cxtm}/api/v1"
        
            project_id = get_project_id(args.project)
            batch_id = get_batch_id(project_id, args.batch)
            run_batch(project_id, args.batch, batch_id)
        
        
        if __name__ == '__main__':
            sys.exit(main())                
        

    

Continue to the next section to create your GitLab CI pipeline fill which will specify what your pipeline does.