Automated Testing

The automated_testing.py script is an example script on how to create a pipeline that can perform automated unit testing on processes. The main process should be designed with testing in mind and with testable subprocesses. A testing harness, which is a second process that contains the subprocess from the main deployable process, will be used to perform unit tests on the subprocesses. The testing harness process will contain multiple tests, each test will send data into the subprocess by a message shape and any required properties, and then the data flowing out of the subprocess will be evaluated against expected results. If it does not pass, an exception will be thrown and the execution will stop.

The following figure is an example main process that is configured to be deployed and used within production. The [SUB] Salesforce Query will contain a Salesforce query. The [SUB] NetSuite Account Create subprocess will contain a NetSuite create operation. Both of these will not be tested. Within the middle, the [SUB] SF to NS Transformation Logic will be tested.

Figure 1. Example main process that is designed to be deployed to production.

Figure 2 is an example of a Testing Harness process. It contains multiple tests against the [SUB] SF to NS Transformation Logic subprocess. Each branch performs a different test. Each branch starts with a message shape that contains input data. If dynamic document properties or dynamic process properties were required, they can also be set at this point. After the data passes through the subprocess that is being tested, the output is validated. If the output data does not pass the validation tests, which is a decision shape in this example, then an exception is thrown. Boomi developers can easily run this process within the build tab. However, it can additionally be run and tested within a release pipeline. The script below shows an execution of the testing harness within a lower environment and will raise an exception if any tests fail.

Figure 2. Example test harness used for unit testing.

The script below is an example that can be used to execute the test harness process and if successful, deploy to the Development environment.

import boomi_cicd


# Open release json
releases = boomi_cicd.set_release()

environment_id = boomi_cicd.query_environment(boomi_cicd.ENVIRONMENT_NAME)
atom_id = boomi_cicd.query_atom(boomi_cicd.ATOM_NAME)

for release in releases["pipelines"]:
    process_name = release["processName"]
    component_id = release["componentId"]
    automated_test_component_id = release["automatedTestId"]
    package_version = release["packageVersion"]
    notes = release.get("notes")

    # The same package version for the main process will be used to keep the versions in parallel.
    automated_test_package_id = boomi_cicd.query_packaged_component(
        automated_test_component_id, package_version
    )
    package_id = boomi_cicd.query_packaged_component(component_id, package_version)

    # Only perform deployments and tests on newly packaged components.
    if not package_id and not automated_test_package_id:
        automated_test_package_id = boomi_cicd.create_packaged_component(
            automated_test_component_id, package_version, notes
        )

        # Deploy the automated test harness
        boomi_cicd.create_deployed_package(
            release, automated_test_package_id, environment_id
        )

        # Execute the test process
        request_id = boomi_cicd.create_execution_request(
            atom_id, automated_test_component_id
        )

        # Check the completed status. Anything other than COMPLETE means it did not pass.
        execution_response = boomi_cicd.get_completed_execution_status(request_id)

        # Raise an exception if the status of the execution is anything other than COMPLETE
        if execution_response["status"] != "COMPLETE":
            raise AssertionError(
                f"Automation test for {process_name}. Error message: {execution_response['message']}"
            )

        # If successful the package and deploy the main process
        package_id = boomi_cicd.create_packaged_component(
            component_id, package_version, notes
        )
        boomi_cicd.create_deployed_package(release, package_id, environment_id)

Release JSON File

The first pipeline object contains an element called automatedTestId which is a component ID of the test process harness. It will also be deployed in a lower environment, executed, and the execution status checked.

{
  "pipelines": [
    {
      "processName": "An Example Batch Process",
      "packageVersion": "2.0",
      "componentId": "83d6013f-96f5-4a75-a97b-f4934b0ec2e8",
      "notes": "This is an example set of notes",
      "schedule": "0 0 1 * * * ; 30 0 2-7 * * *",
      "automatedTestId": "used within automated_testing.py"
    },
    {
      "processName": "An Example Listener Process",
      "packageVersion": "1.0",
      "componentId": "b24f310b-6a66-4e0d-97a3-26f1e812b79a",
      "notes": "This is an example set of notes",
      "listenerStatus": "RUNNING"
    },
    {
      "processName": "An Example Custom Library",
      "componentId": "7bd40730-6df3-4ba9-b4b2-ed9153dbca6d",
      "packageVersion": "1.0",
      "notes": "Initial deployment"
    }
  ]
}

Required Environment Variables

Required Environment Variables

Environment Variable	Description
BOOMI_ACCOUNT_ID	The Boomi account ID.
BOOMI_ATOM_NAME	The name of the Boomi Atom.
BOOMI_BASE_URL	The base URL for the Boomi API. https://api.boomi.com/api/rest/v1
BOOMI_ENVIRONMENT_NAME	The Boomi environment name.
BOOMI_PASSWORD	The Boomi password. Atomsphere API Token is recommended
BOOMI_RELEASE_FILE	The name of the release JSON file to use.
BOOMI_USERNAME	The Boomi username. Atomsphere API Token is recommended

Alternative Testing Strategies Using JUnit XML Format

The concepts within this section build upon the previous section but demonstrate an additional way to implement automated unit testing. After the initial test harness is built out, the assertion steps after validation can be formatted to a JUnit XML file and configured to be viewed at a return document shape. The JUnit XML format is important because it is common for CICD tools to automatically parse this format and display the results in a more user-friendly way. Below is an example JUnit XML that has two test cases and the second test case failed. The JUnit XML schema can be found at JUnit XML Schema.

Example JUnit XML format

 <?xml version='1.0' encoding='UTF-8'?>
 <testsuites>
   <testsuite name="[TEST] Salesforce to NetSuite Account ADO" tests="3" time="3275">
     <testcase name="Test Case 1 - Value is set to 1" time="2241" />
     <testcase name="Test Case 2 - Value is set to 2" time="432" />
       <failure message="Data did not match expected results." />
     </testcase>
   </testsuite>
 </testsuites>

The figure below has two tests that are being executed. The results of each test are put into a Document Cache. The final branch retrieved all documents from the cache and combines the documents into a single document. The fully combined document will be available in the Return Document shape within Process Reporting.

Figure 3. Example test harness used for unit testing with JUnit XML format.

The script below is an example script that can be used to execute the test harness process, retrieve the automatically JUnit XML document at the return document shape, and write the JUnit XML to a file. Update the script to fit any specific requirements.

import boomi_cicd
from boomi_cicd.util import logging

# Open release json
releases = boomi_cicd.set_release()

environment_id = boomi_cicd.query_environment(boomi_cicd.ENVIRONMENT_NAME)
atom_id = boomi_cicd.query_atom(boomi_cicd.ATOM_NAME)

for release in releases["pipelines"]:
    process_name = release["processName"]
    component_id = release["componentId"]
    automated_test_component_id = release.get("automatedTestId")
    package_version = release["packageVersion"]
    notes = release.get("notes")

    # Check if automated testing is required
    if automated_test_component_id:
        # The same package version for the main process will be used to keep the versions in parallel.
        automated_test_package_id = boomi_cicd.query_packaged_component(
            automated_test_component_id, package_version
        )
        package_id = boomi_cicd.query_packaged_component(component_id, package_version)

        # Only perform deployments and tests on newly packaged components.
        if not package_id and not automated_test_package_id:
            automated_test_package_id = boomi_cicd.create_packaged_component(
                automated_test_component_id, package_version, notes
            )
            package_id = boomi_cicd.query_packaged_component(
                component_id, package_version
            )

            # Deploy the automated test harness
            boomi_cicd.create_deployed_package(
                release, automated_test_package_id, environment_id
            )

            # Execute the test process
            request_id = boomi_cicd.create_execution_request(
                atom_id, automated_test_component_id
            )

            # Check the execution's completion status.
            execution_response = boomi_cicd.get_completed_execution_status(request_id)

            # Log the results of the test.  Anything other than COMPLETE means the executed failed.
            if execution_response["status"] == "COMPLETE":
                # Since the process executed successfully,
                # query the execution to get the document at the return document shape.
                # That document should be in a JUnit format and writen to disk.
                execution_record_id = boomi_cicd.query_execution_connector(
                    execution_response["executionId"]
                )

                generic_connector_record = boomi_cicd.query_generic_connector_record(
                    execution_response["executionId"],
                    execution_record_id["result"][0]["id"],
                )

                get_connector_document = boomi_cicd.get_connector_document(
                    generic_connector_record["result"][0]["id"]
                )

                cleaned_document = get_connector_document.replace("\r\n", "\n").strip()
                with open(f"test-{process_name}.xml", "w", encoding="utf-8") as f:
                    f.write(cleaned_document)
            else:
                raise AssertionError(
                    "One or more automated tests failed. Please see logs for more information."
                )
        else:
            logging.warning(
                f"Automation test for {process_name} was skipped. Package already exists."
            )
    else:
        logging.warning(f"Automation test for {process_name} was skipped.")


# Once the all tests have completed, the release_pipeline.py script can be used to deploy the processes.
# JUnit test results can often throw errors within the deployment tool when errors are present.

The following is an example configuration of how Azure DevOps can be configured to read the JUnit XML file within a Release Pipeline. If any test fails, the release pipeline will fail. The percentage of tests that passed can be viewed within the pipeline.

Figure 4. Example JUnit Test Results in Azure DevOps.

The image below is from an Azure DevOps Release Pipeline and shows that 88% of the tests passed within the release pipeline. Additional details can be seen by clicking on the test results section of the stage.

Figure 4. Example JUnit Test Results in Azure DevOps.

Additional Resources

• Boomi Community: Boomi Harness

• Boomi Community: Testing Strategies