path: root/tools/pkt_gen/testcenter/testcenter-rfc2544-throughput.py

# Copyright 2015 Spirent Communications.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

'''
@author Spirent Comunications

This test automates the RFC2544 throughput test using the Spirent
TestCenter command sequencer. This test supports Python 2.7.

'''
from __future__ import print_function
import argparse
import os
from conf import settings


def create_dir(path):
    if not os.path.exists(path):
        try:
            os.makedirs(path)
        except Exception, e:
            print("Failed to create directory %s: %s" % (path, str(e)))
            raise


def write_query_results_to_csv(results_path, csv_results_file_prefix, query_results):
    create_dir(results_path)
    file = os.path.join(results_path, csv_results_file_prefix + ".csv")
    with open(file, "wb") as f:
        f.write(query_results["Columns"].replace(" ", ",") + "\n")
        for row in query_results["Output"].replace("} {", ",").replace("{", "").replace("}", "").split(","):
            f.write(row.replace(" ", ",") + "\n")


def destroy_session(stc, lab_server_addr, test_session_name):
    stc.perform("CSServerConnectCommand",
                Host=lab_server_addr)

    cs_server = stc.get("system1", "children-csserver")

    cs_test_sessions = stc.get(cs_server, "children-cstestsession").split()

    for session in cs_test_sessions:
        name = stc.get(session, "Name")
        if test_session_name in name:
            stc.perform("CSDestroyTestSession",
                        testSession=session)
            break


def create_session(stc, lab_server_addr, test_session_name):
    destroy_session(stc, lab_server_addr, test_session_name)
    stc.perform("CSTestSessionConnect",
                Host=lab_server_addr,
                CreateNewTestSession="TRUE",
                TestSessionName=test_session_name)


def positive_int(value):
    ivalue = int(value)
    if ivalue <= 0:
        raise argparse.ArgumentTypeError("%s is an invalid positive int value" % value)
    return ivalue


def percent_float(value):
    pvalue = float(value)
    if pvalue < 0.0 or pvalue > 100.0:
        raise argparse.ArgumentTypeError("%s not in range [0.0, 100.0]" % pvalue)
    return pvalue


def main():
    parser = argparse.ArgumentParser()
    # Required parameters
    requirednamed = parser.add_argument_group("required named arguments")
    requirednamed.add_argument("--lab_server_addr",
                               required=True,
                               help="The IP address of the Spirent Lab Server",
                               dest="lab_server_addr")
    requirednamed.add_argument("--license_server_addr",
                               required=True,
                               help="The IP address of the Spirent License Server",
                               dest="license_server_addr")
    requirednamed.add_argument("--east_chassis_addr",
                               required=True,
                               help="The TestCenter chassis IP address to use for the east test port",
                               dest="east_chassis_addr")
    requirednamed.add_argument("--east_slot_num",
                               type=positive_int,
                               required=True,
                               help="The TestCenter slot number to use for the east test port",
                               dest="east_slot_num")
    requirednamed.add_argument("--east_port_num",
                               type=positive_int,
                               required=True,
                               help="The TestCenter port number to use for the east test port",
                               dest="east_port_num")
    requirednamed.add_argument("--west_chassis_addr",
                               required=True,
                               help="The TestCenter chassis IP address to use for the west test port",
                               dest="west_chassis_addr")
    requirednamed.add_argument("--west_slot_num",
                               type=positive_int,
                               required=True,
                               help="The TestCenter slot number to use for the west test port",
                               dest="west_slot_num")
    requirednamed.add_argument("--west_port_num",
                               type=positive_int,
                               required=True,
                               help="The TestCenter port number to use for the west test port",
                               dest="west_port_num")
    # Optional parameters
    optionalnamed = parser.add_argument_group("optional named arguments")
    optionalnamed.add_argument("--test_session_name",
                               required=False,
                               default="RFC2544 East-West Throughput",
                               help="The friendly name to identify the Spirent Lab Server test session",
                               dest="test_session_name")
    optionalnamed.add_argument("--results_dir",
                               required=False,
                               default=settings.getValue("TRAFFICGEN_STC_RESULTS_DIR"),
                               help="The directory to copy results to",
                               dest="results_dir")
    optionalnamed.add_argument("--csv_results_file_prefix",
                               required=False,
                               default="Rfc2544Tput",
                               help="The prefix for the CSV results files",
                               dest="csv_results_file_prefix")
    optionalnamed.add_argument("--num_trials",
                               type=positive_int,
                               required=False,
                               default=1,
                               help="The number of trials to execute during the test",
                               dest="num_trials")
    optionalnamed.add_argument("--trial_duration_sec",
                               type=positive_int,
                               required=False,
                               default=60,
                               help="The duration of each trial executed during the test",
                               dest="trial_duration_sec")
    optionalnamed.add_argument("--traffic_pattern",
                               required=False,
                               choices=["BACKBONE", "MESH", "PAIR"],
                               default="PAIR",
                               help="The traffic pattern between endpoints",
                               dest="traffic_pattern")
    optionalnamed.add_argument("--search_mode",
                               required=False,
                               choices=["COMBO", "STEP", "BINARY"],
                               default="BINARY",
                               help="The search mode used to find the throughput rate",
                               dest="search_mode")
    optionalnamed.add_argument("--learning_mode",
                               required=False,
                               choices=["AUTO", "L2_LEARNING", "L3_LEARNING", "NONE"],
                               default="AUTO",
                               help="The learning mode used during the test, default = 'NONE'",
                               dest="learning_mode")
    optionalnamed.add_argument("--rate_lower_limit_pct",
                               type=percent_float,
                               required=False,
                               default=1.0,
                               help="The minimum percent line rate that will be used during the test",
                               dest="rate_lower_limit_pct")
    optionalnamed.add_argument("--rate_upper_limit_pct",
                               type=percent_float,
                               required=False,
                               default=99.0,
                               help="The maximum percent line rate that will be used during the test",
                               dest="rate_upper_limit_pct")
    optionalnamed.add_argument("--rate_initial_pct",
                               type=percent_float,
                               required=False,
                               default=99.0,
                               help="If Search Mode is BINARY, the percent line rate that will be used at the start of the test",
                               dest="rate_initial_pct")
    optionalnamed.add_argument("--rate_step_pct",
                               type=percent_float,
                               required=False,
                               default=10.0,
                               help="If SearchMode is STEP, the percent load increase per step",
                               dest="rate_step_pct")
    optionalnamed.add_argument("--resolution_pct",
                               type=percent_float,
                               required=False,
                               default=1.0,
                               help="The minimum percentage of load adjustment between iterations",
                               dest="resolution_pct")
    optionalnamed.add_argument("--frame_size_list",
                               type=lambda s: [int(item) for item in s.split(',')],
                               required=False,
                               default=[256],
                               help="A comma-delimited list of frame sizes",
                               dest="frame_size_list")
    optionalnamed.add_argument("--acceptable_frame_loss_pct",
                               type=percent_float,
                               required=False,
                               default=0.0,
                               help="The maximum acceptable frame loss percent in any iteration",
                               dest="acceptable_frame_loss_pct")
    optionalnamed.add_argument("--east_intf_addr",
                               required=False,
                               default="192.85.1.3",
                               help="The address to assign to the first emulated device interface on the first east port",
                               dest="east_intf_addr")
    optionalnamed.add_argument("--east_intf_gateway_addr",
                               required=False,
                               default="192.85.1.53",
                               help="The gateway address to assign to the first emulated device interface on the first east port",
                               dest="east_intf_gateway_addr")
    optionalnamed.add_argument("--west_intf_addr",
                               required=False,
                               default="192.85.1.53",
                               help="The address to assign to the first emulated device interface on the first west port",
                               dest="west_intf_addr")
    optionalnamed.add_argument("--west_intf_gateway_addr",
                               required=False,
                               default="192.85.1.53",
                               help="The gateway address to assign to the first emulated device interface on the first west port",
                               dest="west_intf_gateway_addr")
    parser.add_argument("-v",
                        "--verbose",
                        required=False,
                        default=False,
                        help="More output during operation when present",
                        action="store_true",
                        dest="verbose")
    args = parser.parse_args()

    if args.verbose:
        print("Make sure results directory exists")
    create_dir(args.results_dir)

    # Load Spirent Test Center library
    from StcPython import StcPython
    stc = StcPython()

    tx_port_loc = "//%s/%s/%s" % (args.east_chassis_addr,
                                  args.east_slot_num,
                                  args.east_port_num)
    rx_port_loc = "//%s/%s/%s" % (args.west_chassis_addr,
                                  args.west_slot_num,
                                  args.west_port_num)

    # This line will write the TestCenter commands to a log file
    stc.config("automationoptions", logto="test-op", loglevel="DEBUG")

    # Retrieve and display the current API version.
    if args.verbose:
        print ("SpirentTestCenter system version: %s" % stc.get("system1", "version"))

    create_session(stc, args.lab_server_addr, args.test_session_name)

    try:
        device_list = []

        if args.verbose:
            print("Bring up license server")
        license_mgr = stc.get("system1", "children-licenseservermanager")
        if args.verbose:
            print("license_mgr = %s" % license_mgr)
        stc.create("LicenseServer", under=license_mgr, server=args.license_server_addr)

        # Create the root project object
        if args.verbose:
            print ("Creating project ...")
        project = stc.get("System1", "children-Project")

        # Create ports
        if args.verbose:
            print ("Creating ports ...")
        east_chassis_port = stc.create("port",
                                       under=project,
                                       location=tx_port_loc,
                                       useDefaultHost="False")
        west_chassis_port = stc.create("port",
                                       under=project,
                                       location=rx_port_loc,
                                       useDefaultHost="False")

        if args.verbose:
            print ("Creating devices...")
        # Create emulated genparam for east port
        east_device_gen_params = stc.create("EmulatedDeviceGenParams",
                                            under=project,
                                            Port=east_chassis_port)
        # Create the DeviceGenEthIIIfParams object
        stc.create("DeviceGenEthIIIfParams",
                   under=east_device_gen_params)
        # Configuring Ipv4 interfaces
        stc.create("DeviceGenIpv4IfParams",
                   under=east_device_gen_params,
                   Addr=args.east_intf_addr,
                   Gateway=args.east_intf_gateway_addr)
        # Create Devices using the Device Wizard
        device_gen_config = stc.perform("DeviceGenConfigExpand",
                                        DeleteExisting="No",
                                        GenParams=east_device_gen_params)
        # Append to the device list
        device_list.append(device_gen_config['ReturnList'])

        # Create emulated genparam for west port
        west_device_gen_params = stc.create("EmulatedDeviceGenParams",
                                            under=project,
                                            Port=west_chassis_port)
        # Create the DeviceGenEthIIIfParams object
        stc.create("DeviceGenEthIIIfParams",
                   under=west_device_gen_params)
        # Configuring Ipv4 interfaces
        stc.create("DeviceGenIpv4IfParams",
                   under=west_device_gen_params,
                   Addr=args.west_intf_addr,
                   Gateway=args.west_intf_gateway_addr)
        # Create Devices using the Device Wizard
        device_gen_config = stc.perform("DeviceGenConfigExpand",
                                        DeleteExisting="No",
                                        GenParams=west_device_gen_params)
        # Append to the device list
        device_list.append(device_gen_config['ReturnList'])

        if args.verbose:
            print(device_list)

        # Create the RFC 2544 throughput test
        if args.verbose:
            print ("Set up the RFC2544 throughput test...")
        stc.perform("Rfc2544SetupThroughputTestCommand",
                    AcceptableFrameLoss=args.acceptable_frame_loss_pct,
                    Duration=args.trial_duration_sec,
                    FrameSizeList=args.frame_size_list,
                    LearningMode=args.learning_mode,
                    NumOfTrials=args.num_trials,
                    RateInitial=args.rate_initial_pct,
                    RateLowerLimit=args.rate_lower_limit_pct,
                    RateStep=args.rate_step_pct,
                    RateUpperLimit=args.rate_upper_limit_pct,
                    Resolution=args.resolution_pct,
                    SearchMode=args.search_mode,
                    TrafficPattern=args.traffic_pattern)

        # Save the configuration
        stc.perform("SaveToTcc", Filename="2544.tcc")

        # Connect to the hardware...
        stc.perform("AttachPorts", portList=stc.get("system1.project", "children-port"), autoConnect="TRUE")

        # Apply configuration.
        if args.verbose:
            print("Apply configuration...")
        stc.apply

        if args.verbose:
            print("Starting the sequencer...")
        stc.perform("SequencerStart")

        # Wait for sequencer to finish
        print("Starting test... Please wait for the test to complete...")
        stc.waitUntilComplete()
        print("The test has completed... Saving results...")

        # Determine what the results database filename is...
        lab_server_resultsdb = stc.get("system1.project.TestResultSetting", "CurrentResultFileName")

        if args.verbose:
            print("The lab server results database is %s" % lab_server_resultsdb)

        stc.perform("CSSynchronizeFiles",
                    DefaultDownloadDir=args.results_dir)

        resultsdb = args.results_dir + lab_server_resultsdb.split("/Results")[1]

        print("The local summary DB file has been saved to %s" % resultsdb)

        # The returns the "RFC2544ThroughputTestResultDetailedSummaryView" table view from the results database.
        # There are other views available.
        resultsdict = stc.perform("QueryResult",
                                  DatabaseConnectionString=resultsdb,
                                  ResultPath="RFC2544ThroughputTestResultDetailedSummaryView")
        if args.verbose:
            print("resultsdict[\"Columns\"]: %s" % resultsdict["Columns"])
            print("resultsdict[\"Output\"]: %s" % resultsdict["Output"])

        if args.verbose:
            print("Result paths: %s" % stc.perform("GetTestResultSettingPaths"))

        # Write results to csv
        if args.verbose:
            print("Writing CSV file to results directory %s" % args.results_dir)
        write_query_results_to_csv(args.results_dir, args.csv_results_file_prefix, resultsdict)

    except Exception, e:
        print("Exception: %s" % str(e))
        # destroy_session(stc, args.lab_server_addr, args.test_session_name)
        raise

    if args.verbose:
        print("Disconnect from lab server")
    stc.perform("CSTestSessionDisconnect")

    if args.verbose:
        print("Destroy session on lab server")
    # destroy_session(stc, args.lab_server_addr, args.test_session_name)
    print("Test complete!")


if __name__ == "__main__":
    main()