trex_path=/opt/nsb_bin/trex/scripts
bin_path=/opt/nsb_bin
trex_client_lib=/opt/nsb_bin/trex_client/stl
+# Path to 32-bit Java installation, for Spirent Landslide TG API
+jre_path_i386=/usr/lib/jvm/java-8-openjdk-i386/jre
class ValueCheckError(YardstickException):
message = 'Constraint "%(value1)s %(operator)s %(value2)s" does not hold'
+
+
+class LandslideTclException(RuntimeError):
+ def __init__(self, message):
+ self._message = message
+ super(LandslideTclException, self).__init__(message)
'yardstick.network_services.traffic_profile.prox_ramp',
'yardstick.network_services.traffic_profile.rfc2544',
'yardstick.network_services.traffic_profile.pktgen',
+ 'yardstick.network_services.traffic_profile.landslide_profile',
]
for module in modules:
--- /dev/null
+# Copyright (c) 2018 Intel Corporation
+#
+# 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.
+""" Spirent Landslide traffic profile definitions """
+
+from yardstick.network_services.traffic_profile import base
+
+
+class LandslideProfile(base.TrafficProfile):
+ """
+ This traffic profile handles attributes of Landslide data stream
+ """
+
+ def __init__(self, tp_config):
+ super(LandslideProfile, self).__init__(tp_config)
+
+ # for backward compatibility support dict and list of dicts
+ if isinstance(tp_config["dmf_config"], dict):
+ self.dmf_config = [tp_config["dmf_config"]]
+ else:
+ self.dmf_config = tp_config["dmf_config"]
+
+ def execute(self, traffic_generator):
+ pass
+
+ def update_dmf(self, options):
+ if 'dmf' in options:
+ if isinstance(options['dmf'], dict):
+ _dmfs = [options['dmf']]
+ else:
+ _dmfs = options['dmf']
+
+ for index, _dmf in enumerate(_dmfs):
+ try:
+ self.dmf_config[index].update(_dmf)
+ except IndexError:
+ pass
cfg.StrOpt('trex_client_lib',
default=os.path.join(NSB_ROOT, 'trex_client/stl'),
help='trex python library path.'),
+ cfg.StrOpt('jre_path_i386',
+ default='',
+ help='path to installation of 32-bit Java 1.7+.'),
]
CONF.register_opts(OPTS, group="nsb")
--- /dev/null
+# Copyright (c) 2018 Intel Corporation
+#
+# 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.
+
+import logging
+
+from yardstick.common import exceptions
+from yardstick.common import utils as common_utils
+from yardstick.network_services import utils as net_serv_utils
+from yardstick.network_services.vnf_generic.vnf import sample_vnf
+
+try:
+ from lsapi import LsApi
+except ImportError:
+ LsApi = common_utils.ErrorClass
+
+LOG = logging.getLogger(__name__)
+
+
+class LandslideResourceHelper(sample_vnf.ClientResourceHelper):
+ """Landslide TG helper class"""
+
+ REST_STATUS_CODES = {'OK': 200, 'CREATED': 201, 'NO CHANGE': 409}
+ REST_API_CODES = {'NOT MODIFIED': 500810}
+
+ def __init__(self, setup_helper):
+ super(LandslideResourceHelper, self).__init__(setup_helper)
+ self._result = {}
+ self.vnfd_helper = setup_helper.vnfd_helper
+ self.scenario_helper = setup_helper.scenario_helper
+
+ # TCL session initialization
+ self._tcl = LandslideTclClient(LsTclHandler(), self)
+
+ def terminate(self):
+ raise NotImplementedError()
+
+ def collect_kpi(self):
+ raise NotImplementedError()
+
+
+class LandslideTclClient(object):
+ """Landslide TG TCL client class"""
+
+ DEFAULT_TEST_NODE = {
+ 'ethStatsEnabled': True,
+ 'forcedEthInterface': '',
+ 'innerVlanId': 0,
+ 'ip': '',
+ 'mac': '',
+ 'mtu': 1500,
+ 'nextHop': '',
+ 'numLinksOrNodes': 1,
+ 'numVlan': 1,
+ 'phy': '',
+ 'uniqueVlanAddr': False,
+ 'vlanDynamic': 0,
+ 'vlanId': 0,
+ 'vlanUserPriority': 0,
+ 'vlanTagType': 0
+ }
+
+ TEST_NODE_CMD = \
+ 'ls::create -TestNode-{} -under $p_ -Type "eth"' \
+ ' -Phy "{phy}" -Ip "{ip}" -NumLinksOrNodes {numLinksOrNodes}' \
+ ' -NextHop "{nextHop}" -Mac "{mac}" -MTU {mtu} ' \
+ ' -ForcedEthInterface "{forcedEthInterface}"' \
+ ' -EthStatsEnabled {ethStatsEnabled}' \
+ ' -VlanId {vlanId} -VlanUserPriority {vlanUserPriority}' \
+ ' -NumVlan {numVlan} -UniqueVlanAddr {uniqueVlanAddr}' \
+ ';'
+
+ def __init__(self, tcl_handler, ts_context):
+ self.tcl_server_ip = None
+ self._user = None
+ self._library_id = None
+ self._basic_library_id = None
+ self._tcl = tcl_handler
+ self._ts_context = ts_context
+ self.ts_ids = set()
+
+ # Test types names expected in session profile, test case and pod files
+ self._tc_types = {"SGW_Nodal", "SGW_Node", "MME_Nodal", "PGW_Node",
+ "PCRF_Node"}
+
+ self._class_param_config_handler = {
+ "Array": self._configure_array_param,
+ "TestNode": self._configure_test_node_param,
+ "Sut": self._configure_sut_param,
+ "Dmf": self._configure_dmf_param
+ }
+
+ def connect(self, tcl_server_ip, username, password):
+ """ Connect to TCL server with username and password
+
+ :param tcl_server_ip: TCL server IP address
+ :type tcl_server_ip: str
+ :param username: existing username on TCL server
+ :type username: str
+ :param password: password related to username on TCL server
+ :type password: str
+ """
+ LOG.info("connect: server='%s' user='%s'", tcl_server_ip, username)
+ res = self._tcl.execute(
+ "ls::login {} {} {}".format(tcl_server_ip, username, password))
+ if 'java0x' not in res: # handle assignment reflects login success
+ raise exceptions.LandslideTclException(
+ "connect: login failed ='{}'.".format(res))
+ self._library_id = self._tcl.execute(
+ "ls::get [ls::query LibraryInfo -userLibraryName {}] -Id".format(
+ username))
+ self._basic_library_id = self._get_library_id('Basic')
+ self.tcl_server_ip = tcl_server_ip
+ self._user = username
+ LOG.debug("connect: user='%s' me='%s' basic='%s'", self._user,
+ self._library_id,
+ self._basic_library_id)
+
+ def disconnect(self):
+ """ Disconnect from TCL server. Drop TCL connection configuration """
+ LOG.info("disconnect: server='%s' user='%s'",
+ self.tcl_server_ip, self._user)
+ self._tcl.execute("ls::logout")
+ self.tcl_server_ip = None
+ self._user = None
+ self._library_id = None
+ self._basic_library_id = None
+
+ def _add_test_server(self, name, ip):
+ try:
+ # Check if test server exists with name equal to _ts_name
+ ts_id = int(self.resolve_test_server_name(name))
+ except ValueError:
+ # Such test server does not exist. Attempt to create it
+ ts_id = self._tcl.execute(
+ 'ls::perform AddTs -Name "{}" -Ip "{}"'.format(name, ip))
+ try:
+ int(ts_id)
+ except ValueError:
+ # Failed to create test server, e.g. limit reached
+ raise RuntimeError(
+ 'Failed to create test server: "{}". {}'.format(name,
+ ts_id))
+ return ts_id
+
+ def _update_license(self, name):
+ """ Setup/update test server license
+
+ :param name: test server name
+ :type name: str
+ """
+ # Retrieve current TsInfo configuration, result stored in handle "ts"
+ self._tcl.execute(
+ 'set ts [ls::retrieve TsInfo -Name "{}"]'.format(name))
+
+ # Set license ID, if it differs from current one, update test server
+ _curr_lic_id = self._tcl.execute('ls::get $ts -RequestedLicense')
+ if _curr_lic_id != self._ts_context.license_data['lic_id']:
+ self._tcl.execute('ls::config $ts -RequestedLicense {}'.format(
+ self._ts_context.license_data['lic_id']))
+ self._tcl.execute('ls::perform ModifyTs $ts')
+
+ def _set_thread_model(self, name, thread_model):
+ # Retrieve test server configuration, store it in handle "tsc"
+ _cfguser_password = self._ts_context.vnfd_helper['mgmt-interface'][
+ 'cfguser_password']
+ self._tcl.execute(
+ 'set tsc [ls::perform RetrieveTsConfiguration '
+ '-name "{}" {}]'.format(name, _cfguser_password))
+ # Configure ThreadModel, if it differs from current one
+ thread_model_map = {'Legacy': 'V0',
+ 'Max': 'V1',
+ 'Fireball': 'V1_FB3'}
+ _model = thread_model_map[thread_model]
+ _curr_model = self._tcl.execute('ls::get $tsc -ThreadModel')
+ if _curr_model != _model:
+ self._tcl.execute(
+ 'ls::config $tsc -ThreadModel "{}"'.format(_model))
+ self._tcl.execute(
+ 'ls::perform ApplyTsConfiguration $tsc {}'.format(
+ _cfguser_password))
+
+ def create_test_server(self, test_server):
+ _ts_thread_model = test_server.get('thread_model')
+ _ts_name = test_server['name']
+
+ ts_id = self._add_test_server(_ts_name, test_server['ip'])
+
+ self._update_license(_ts_name)
+
+ # Skip below code modifying thread_model if it is not defined
+ if _ts_thread_model:
+ self._set_thread_model(_ts_name, _ts_thread_model)
+
+ return ts_id
+
+ def create_test_session(self, test_session):
+ """ Create, configure and save Landslide test session object.
+
+ :param test_session: Landslide TestSession object
+ :type test_session: dict
+ """
+ LOG.info("create_test_session: name='%s'", test_session['name'])
+ self._tcl.execute('set test_ [ls::create TestSession]')
+ self._tcl.execute('ls::config $test_ -Library {} -Name "{}"'.format(
+ self._library_id, test_session['name']))
+ self._tcl.execute('ls::config $test_ -Description "{}"'.format(
+ test_session['description']))
+ if 'keywords' in test_session:
+ self._tcl.execute('ls::config $test_ -Keywords "{}"'.format(
+ test_session['keywords']))
+ if 'duration' in test_session:
+ self._tcl.execute('ls::config $test_ -Duration "{}"'.format(
+ test_session['duration']))
+ if 'iterations' in test_session:
+ self._tcl.execute('ls::config $test_ -Iterations "{}"'.format(
+ test_session['iterations']))
+ if 'reservePorts' in test_session:
+ if test_session['reservePorts']:
+ self._tcl.execute('ls::config $test_ -Reserve Ports')
+
+ if 'reservations' in test_session:
+ for _reservation in test_session['reservations']:
+ self._configure_reservation(_reservation)
+
+ if 'reportOptions' in test_session:
+ self._configure_report_options(test_session['reportOptions'])
+
+ for _index, _group in enumerate(test_session['tsGroups']):
+ self._configure_ts_group(_group, _index)
+
+ self._save_test_session()
+
+ def create_dmf(self, dmf):
+ """ Create, configure and save Landslide Data Message Flow object.
+
+ :param dmf: Landslide Data Message Flow object
+ :type: dmf: dict
+ """
+ self._tcl.execute('set dmf_ [ls::create Dmf]')
+ _lib_id = self._get_library_id(dmf['dmf']['library'])
+ self._tcl.execute('ls::config $dmf_ -Library {} -Name "{}"'.format(
+ _lib_id,
+ dmf['dmf']['name']))
+ for _param_key in dmf:
+ if _param_key == 'dmf':
+ continue
+ _param_value = dmf[_param_key]
+ if isinstance(_param_value, dict):
+ # Configure complex parameter
+ _tcl_cmd = 'ls::config $dmf_'
+ for _sub_param_key in _param_value:
+ _sub_param_value = _param_value[_sub_param_key]
+ if isinstance(_sub_param_value, str):
+ _tcl_cmd += ' -{} "{}"'.format(_sub_param_key,
+ _sub_param_value)
+ else:
+ _tcl_cmd += ' -{} {}'.format(_sub_param_key,
+ _sub_param_value)
+
+ self._tcl.execute(_tcl_cmd)
+ else:
+ # Configure simple parameter
+ if isinstance(_param_value, str):
+ self._tcl.execute(
+ 'ls::config $dmf_ -{} "{}"'.format(_param_key,
+ _param_value))
+ else:
+ self._tcl.execute(
+ 'ls::config $dmf_ -{} {}'.format(_param_key,
+ _param_value))
+ self._save_dmf()
+
+ def configure_dmf(self, dmf):
+ # Use create to reconfigure and overwrite existing dmf
+ self.create_dmf(dmf)
+
+ def delete_dmf(self, dmf):
+ raise NotImplementedError
+
+ def _save_dmf(self):
+ # Call 'Validate' to set default values for missing parameters
+ res = self._tcl.execute('ls::perform Validate -Dmf $dmf_')
+ if res == 'Invalid':
+ res = self._tcl.execute('ls::get $dmf_ -ErrorsAndWarnings')
+ LOG.error("_save_dmf: %s", res)
+ raise exceptions.LandslideTclException("_save_dmf: {}".format(res))
+ else:
+ res = self._tcl.execute('ls::save $dmf_ -overwrite')
+ LOG.debug("_save_dmf: result (%s)", res)
+
+ def _configure_report_options(self, options):
+ for _option_key in options:
+ _option_value = options[_option_key]
+ if _option_key == 'format':
+ _format = 0
+ if _option_value == 'CSV':
+ _format = 1
+ self._tcl.execute(
+ 'ls::config $test_.ReportOptions -Format {} '
+ '-Ts -3 -Tc -3'.format(_format))
+ else:
+ self._tcl.execute(
+ 'ls::config $test_.ReportOptions -{} {}'.format(
+ _option_key,
+ _option_value))
+
+ def _configure_ts_group(self, ts_group, ts_group_index):
+ try:
+ _ts_id = int(self.resolve_test_server_name(ts_group['tsId']))
+ except ValueError:
+ raise RuntimeError('Test server name "{}" does not exist.'.format(
+ ts_group['tsId']))
+ if _ts_id not in self.ts_ids:
+ self._tcl.execute(
+ 'set tss_ [ls::create TsGroup -under $test_ -tsId {} ]'.format(
+ _ts_id))
+ self.ts_ids.add(_ts_id)
+ for _case in ts_group.get('testCases', []):
+ self._configure_tc_type(_case, ts_group_index)
+
+ self._configure_preresolved_arp(ts_group.get('preResolvedArpAddress'))
+
+ def _configure_tc_type(self, tc, ts_group_index):
+ if tc['type'] not in self._tc_types:
+ raise RuntimeError('Test type {} not supported.'.format(
+ tc['type']))
+ tc['type'] = tc['type'].replace('_', ' ')
+ res = self._tcl.execute(
+ 'set tc_ [ls::retrieve testcase -libraryId {0} "{1}"]'.format(
+ self._basic_library_id, tc['type']))
+ if 'Invalid' in res:
+ raise RuntimeError('Test type {} not found in "Basic" '
+ 'library.'.format(tc['type']))
+ self._tcl.execute(
+ 'ls::config $test_.TsGroup({}) -children-Tc $tc_'.format(
+ ts_group_index))
+ self._tcl.execute('ls::config $tc_ -Library {0} -Name "{1}"'.format(
+ self._basic_library_id, tc['name']))
+ self._tcl.execute(
+ 'ls::config $tc_ -Description "{}"'.format(tc['type']))
+ self._tcl.execute(
+ 'ls::config $tc_ -Keywords "GTP LTE {}"'.format(tc['type']))
+ if 'linked' in tc:
+ self._tcl.execute(
+ 'ls::config $tc_ -Linked {}'.format(tc['linked']))
+ if 'AssociatedPhys' in tc:
+ self._tcl.execute('ls::config $tc_ -AssociatedPhys "{}"'.format(
+ tc['AssociatedPhys']))
+ if 'parameters' in tc:
+ self._configure_parameters(tc['parameters'])
+
+ def _configure_parameters(self, params):
+ self._tcl.execute('set p_ [ls::get $tc_ -children-Parameters(0)]')
+ for _param_key in sorted(params):
+ _param_value = params[_param_key]
+ if isinstance(_param_value, dict):
+ # Configure complex parameter
+ if _param_value['class'] in self._class_param_config_handler:
+ self._class_param_config_handler[_param_value['class']](
+ _param_key,
+ _param_value)
+ else:
+ # Configure simple parameter
+ self._tcl.execute(
+ 'ls::create {} -under $p_ -Value "{}"'.format(
+ _param_key,
+ _param_value))
+
+ def _configure_array_param(self, name, params):
+ self._tcl.execute('ls::create -Array-{} -under $p_ ;'.format(name))
+ for param in params['array']:
+ self._tcl.execute(
+ 'ls::create ArrayItem -under $p_.{} -Value "{}"'.format(name,
+ param))
+
+ def _configure_test_node_param(self, name, params):
+ _params = self.DEFAULT_TEST_NODE
+ _params.update(params)
+
+ # TCL command expects lower case 'true' or 'false'
+ _params['ethStatsEnabled'] = str(_params['ethStatsEnabled']).lower()
+ _params['uniqueVlanAddr'] = str(_params['uniqueVlanAddr']).lower()
+
+ cmd = self.TEST_NODE_CMD.format(name, **_params)
+ self._tcl.execute(cmd)
+
+ def _configure_sut_param(self, name, params):
+ self._tcl.execute(
+ 'ls::create -Sut-{} -under $p_ -Name "{}";'.format(name,
+ params['name']))
+
+ def _configure_dmf_param(self, name, params):
+ self._tcl.execute('ls::create -Dmf-{} -under $p_ ;'.format(name))
+
+ for _flow_index, _flow in enumerate(params['mainflows']):
+ _lib_id = self._get_library_id(_flow['library'])
+ self._tcl.execute(
+ 'ls::perform AddDmfMainflow $p_.Dmf {} "{}"'.format(
+ _lib_id,
+ _flow['name']))
+
+ if not params.get('instanceGroups'):
+ return
+
+ _instance_group = params['instanceGroups'][_flow_index]
+
+ # Traffic Mixer parameters handling
+ for _key in ['mixType', 'rate']:
+ if _key in _instance_group:
+ self._tcl.execute(
+ 'ls::config $p_.Dmf.InstanceGroup({}) -{} {}'.format(
+ _flow_index, _key, _instance_group[_key]))
+
+ # Assignments parameters handling
+ for _row_id, _row in enumerate(_instance_group.get('rows', [])):
+ self._tcl.execute(
+ 'ls::config $p_.Dmf.InstanceGroup({}).Row({}) -Node {} '
+ '-OverridePort {} -ClientPort {} -Context {} -Role {} '
+ '-PreferredTransport {} -RatingGroup {} '
+ '-ServiceID {}'.format(
+ _flow_index, _row_id, _row['node'],
+ _row['overridePort'], _row['clientPort'],
+ _row['context'], _row['role'], _row['transport'],
+ _row['ratingGroup'], _row['serviceId']))
+
+ def _configure_reservation(self, reservation):
+ _ts_id = self.resolve_test_server_name(reservation['tsId'])
+ self._tcl.execute(
+ 'set reservation_ [ls::create Reservation -under $test_]')
+ self._tcl.execute(
+ 'ls::config $reservation_ -TsIndex {} -TsId {} '
+ '-TsName "{}"'.format(reservation['tsIndex'],
+ _ts_id,
+ reservation['tsName']))
+ for _subnet in reservation['phySubnets']:
+ self._tcl.execute(
+ 'set physubnet_ [ls::create PhySubnet -under $reservation_]')
+ self._tcl.execute(
+ 'ls::config $physubnet_ -Name "{}" -Base "{}" -Mask "{}" '
+ '-NumIps {}'.format(_subnet['name'], _subnet['base'],
+ _subnet['mask'], _subnet['numIps']))
+
+ def _configure_preresolved_arp(self, pre_resolved_arp):
+ if not pre_resolved_arp: # Pre-resolved ARP configuration not found
+ return
+ for _entry in pre_resolved_arp:
+ # TsGroup handle name should correspond in _configure_ts_group()
+ self._tcl.execute(
+ 'ls::create PreResolvedArpAddress -under $tss_ '
+ '-StartingAddress "{StartingAddress}" '
+ '-NumNodes {NumNodes}'.format(**_entry))
+
+ def delete_test_session(self, test_session):
+ raise NotImplementedError
+
+ def _save_test_session(self):
+ # Call 'Validate' to set default values for missing parameters
+ res = self._tcl.execute('ls::perform Validate -TestSession $test_')
+ if res == 'Invalid':
+ res = self._tcl.execute('ls::get $test_ -ErrorsAndWarnings')
+ raise exceptions.LandslideTclException(
+ "_save_test_session: {}".format(res))
+ else:
+ res = self._tcl.execute('ls::save $test_ -overwrite')
+ LOG.debug("_save_test_session: result (%s)", res)
+
+ def _get_library_id(self, library):
+ _library_id = self._tcl.execute(
+ "ls::get [ls::query LibraryInfo -systemLibraryName {}] -Id".format(
+ library))
+ try:
+ int(_library_id)
+ return _library_id
+ except ValueError:
+ pass
+
+ _library_id = self._tcl.execute(
+ "ls::get [ls::query LibraryInfo -userLibraryName {}] -Id".format(
+ library))
+ try:
+ int(_library_id)
+ except ValueError:
+ LOG.error("_get_library_id: library='%s' not found.", library)
+ raise exceptions.LandslideTclException(
+ "_get_library_id: library='{}' not found.".format(
+ library))
+
+ return _library_id
+
+ def resolve_test_server_name(self, ts_name):
+ return self._tcl.execute("ls::query TsId {}".format(ts_name))
+
+
+class LsTclHandler(object):
+ """Landslide TCL Handler class"""
+
+ LS_OK = "ls_ok"
+ JRE_PATH = net_serv_utils.get_nsb_option('jre_path_i386')
+
+ def __init__(self):
+ self.tcl_cmds = {}
+ self._ls = LsApi(jre_path=self.JRE_PATH)
+ self._ls.tcl(
+ "ls::config ApiOptions -NoReturnSuccessResponseString '{}'".format(
+ self.LS_OK))
+
+ def execute(self, command):
+ res = self._ls.tcl(command)
+ self.tcl_cmds[command] = res
+ return res
--- /dev/null
+# Copyright (c) 2018 Intel Corporation
+#
+# 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.
+
+import copy
+import unittest
+
+from yardstick.network_services.traffic_profile import landslide_profile
+
+TP_CONFIG = {
+ 'schema': "nsb:traffic_profile:0.1",
+ 'name': 'LandslideProfile',
+ 'description': 'Spirent Landslide traffic profile (Data Message Flow)',
+ 'traffic_profile': {
+ 'traffic_type': 'LandslideProfile'
+ },
+ 'dmf_config': {
+ 'dmf': {
+ 'library': 'test',
+ 'name': 'Fireball UDP',
+ 'description': "Basic data flow using UDP/IP (Fireball DMF)",
+ 'keywords': 'UDP ',
+ 'dataProtocol': 'fb_udp',
+ 'burstCount': 1,
+ 'clientPort': {
+ 'clientPort': 2002,
+ 'isClientPortRange': 'false'
+ },
+ 'serverPort': 2003,
+ 'connection': {
+ 'initiatingSide': 'Client',
+ 'disconnectSide': 'Client',
+ 'underlyingProtocol': 'none',
+ 'persistentConnection': 'false'
+ },
+ 'protocolId': 0,
+ 'persistentConnection': 'false',
+ 'transactionRate': 8.0,
+ 'transactions': {
+ 'totalTransactions': 0,
+ 'retries': 0,
+ 'dataResponseTime': 60000,
+ 'packetSize': 64
+ },
+ 'segment': {
+ 'segmentSize': 64000,
+ 'maxSegmentSize': 0
+ },
+ 'size': {
+ 'sizeDistribution': 'Fixed',
+ 'sizeDeviation': 10
+ },
+ 'interval': {
+ 'intervalDistribution': 'Fixed',
+ 'intervalDeviation': 10
+ },
+ 'ipHeader': {
+ 'typeOfService': 0,
+ 'timeToLive': 64
+ },
+ 'tcpConnection': {
+ 'force3Way': 'false',
+ 'fixedRetryTime': 0,
+ 'maxPacketsToForceAck': 0
+ },
+ 'tcp': {
+ 'windowSize': 32768,
+ 'windowScaling': -1,
+ 'disableFinAckWait': 'false'
+ },
+ 'disconnectType': 'FIN',
+ 'slowStart': 'false',
+ 'connectOnly': 'false',
+ 'vtag': {
+ 'VTagMask': '0x0',
+ 'VTagValue': '0x0'
+ },
+ 'sctpPayloadProtocolId': 0,
+ 'billingIncludeSyn': 'true',
+ 'billingIncludeSubflow': 'true',
+ 'billingRecordPerTransaction': 'false',
+ 'tcpPush': 'false',
+ 'hostDataExpansionRatio': 1
+ }
+ }
+}
+DMF_OPTIONS = {
+ 'dmf': {
+ 'transactionRate': 5,
+ 'packetSize': 512,
+ 'burstCount': 1
+ }
+}
+
+
+class TestLandslideProfile(unittest.TestCase):
+
+ def test___init__(self):
+ ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
+ self.assertListEqual([TP_CONFIG["dmf_config"]],
+ ls_traffic_profile.dmf_config)
+
+ def test___init__config_not_a_dict(self):
+ _tp_config = copy.deepcopy(TP_CONFIG)
+ _tp_config['dmf_config'] = [_tp_config['dmf_config']]
+ ls_traffic_profile = landslide_profile.LandslideProfile(_tp_config)
+ self.assertListEqual(_tp_config['dmf_config'],
+ ls_traffic_profile.dmf_config)
+
+ def test_execute(self):
+ ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
+ self.assertIsNone(ls_traffic_profile.execute(None))
+
+ def test_update_dmf_options_dict(self):
+ ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
+ ls_traffic_profile.update_dmf(DMF_OPTIONS)
+ self.assertDictContainsSubset(DMF_OPTIONS['dmf'],
+ ls_traffic_profile.dmf_config[0])
+
+ def test_update_dmf_options_list(self):
+ ls_traffic_profile = landslide_profile.LandslideProfile(TP_CONFIG)
+ _dmf_options = copy.deepcopy(DMF_OPTIONS)
+ _dmf_options['dmf'] = [_dmf_options['dmf']]
+ ls_traffic_profile.update_dmf(_dmf_options)
+ self.assertTrue(all([x in ls_traffic_profile.dmf_config[0]
+ for x in DMF_OPTIONS['dmf']]))
--- /dev/null
+# Copyright (c) 2018 Intel Corporation
+#
+# 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.
+#
+
+import mock
+import unittest
+
+from yardstick.common import exceptions
+from yardstick.network_services.vnf_generic.vnf import tg_landslide
+
+NAME = "tg__0"
+
+EXAMPLE_URL = 'http://example.com/'
+TCL_SUCCESS_RESPONSE = 'ls_ok'
+
+TEST_SERVERS = [
+ {'ip': '192.168.122.101',
+ 'phySubnets': [
+ {'mask': '/24',
+ 'base': '10.42.32.100',
+ 'numIps': 20,
+ 'name': 'eth1'}
+ ],
+ 'role': 'SGW_Node',
+ 'name': 'TestServer_1'},
+ {'ip': '192.168.122.102',
+ 'phySubnets': [
+ {'mask': '/24',
+ 'base': '10.42.32.1',
+ 'numIps': 100,
+ 'name': 'eth1'
+ },
+ {'mask': '/24',
+ 'base': '10.42.33.1',
+ 'numIps': 100,
+ 'name': 'eth2'}
+ ],
+ 'preResolvedArpAddress': [
+ {'NumNodes': 1,
+ 'StartingAddress': '10.42.33.5'}
+ ],
+ 'role': 'SGW_Nodal',
+ 'name': 'TestServer_2',
+ 'thread_model': 'Fireball'
+ }
+]
+
+TS1_SUTS = [
+ {'name': 'SGW - C TestNode',
+ 'role': 'SgwControlAddr',
+ 'managementIp': '12.0.1.1',
+ 'ip': '10.42.32.100',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ },
+ {'name': 'SGW - U TestNode',
+ 'role': 'SgwUserAddr',
+ 'managementIp': '12.0.1.2',
+ 'ip': '10.42.32.101',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ }
+]
+
+TS2_SUTS = [
+ {'name': 'eNodeB TestNode',
+ 'role': 'EnbUserAddr',
+ 'managementIp': '12.0.2.1',
+ 'ip': '10.42.32.2',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ },
+ {'name': 'MME TestNode',
+ 'role': 'MmeControlAddr',
+ 'managementIp': '12.0.3.1',
+ 'ip': '10.42.32.1',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ },
+ {'name': 'NetHost TestNode',
+ 'role': 'NetworkHostAddrLocal',
+ 'managementIp': '12.0.4.1',
+ 'ip': '10.42.33.1',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ },
+ {'name': 'PGW TestNode',
+ 'role': 'PgwV4Sut',
+ 'managementIp': '12.0.5.1',
+ 'ip': '10.42.32.105',
+ 'phy': 'eth5',
+ 'nextHop': '10.42.32.5'
+ },
+ {'name': 'SGW - C SUT',
+ 'role': 'SgwSut',
+ 'managementIp': '12.0.6.1',
+ 'ip': '10.42.32.100'
+ },
+ {'name': 'SGW - U SUT',
+ 'role': 'SgwUserSut',
+ 'managementIp': '12.0.6.2',
+ 'ip': '10.42.32.101'}
+]
+
+VNFD = {
+ 'vnfd:vnfd-catalog': {
+ 'vnfd': [{
+ 'short-name': 'landslide',
+ 'vdu': [{
+ 'description': 'AB client interface details',
+ 'name': 'abclient-baremetal',
+ 'id': 'abclient-baremetal',
+ 'external-interface': []}],
+ 'description': 'Spirent Landslide traffic generator',
+ 'config': [{'test_server': TEST_SERVERS[0], 'suts': TS1_SUTS},
+ {'test_server': TEST_SERVERS[1], 'suts': TS2_SUTS}],
+ 'mgmt-interface': {
+ 'vdu-id': 'landslide-tas',
+ 'user': 'user',
+ 'password': 'user',
+ 'super-user': 'super-user',
+ 'super-user-password': 'super-user-password',
+ 'cfguser_password': 'cfguser_password',
+ 'license': 48,
+ 'proto': 'http',
+ 'ip': '1.1.1.1'},
+ 'benchmark': {
+ 'kpi': [
+ 'tx_throughput_mbps',
+ 'rx_throughput_mbps',
+ 'in_packets',
+ 'out_packets',
+ 'activation_rate_sessps',
+ 'deactivation_rate_sessps']},
+ 'id': 'LandslideTrafficGen',
+ 'name': 'LandslideTrafficGen'}]}}
+
+TAS_INFO = VNFD['vnfd:vnfd-catalog']['vnfd'][0]['mgmt-interface']
+
+DMF_CFG = {
+ "dmf": {
+ "library": "test",
+ "name": "Basic UDP"
+ },
+ "clientPort": {
+ "clientPort": 2002,
+ "isClientPortRange": "false"
+ },
+ "dataProtocol": "udp",
+ "serverPort": 2003
+}
+
+SESSION_PROFILE = {
+ 'keywords': '',
+ 'duration': 60,
+ 'iterations': 1,
+ 'description': 'UE default bearer creation test case',
+ 'name': 'default_bearer_capacity',
+ 'reportOptions': {'Format': '1'},
+ "reservePorts": "true",
+ "reservations": [
+ {"tsId": 4,
+ "tsIndex": 0,
+ "tsName": TEST_SERVERS[0]['name'],
+ "phySubnets": [
+ {"base": "10.42.32.100", "mask": "/24", "name": "eth5",
+ "numIps": 20},
+ {"base": "10.42.33.100", "mask": "/24", "name": "eth6",
+ "numIps": 20}
+ ]},
+ {"tsId": 2,
+ "tsIndex": 1,
+ "tsName": "TestServer_2",
+ "phySubnets": [
+ {"base": "10.42.32.1", "mask": "/24", "name": "eth5",
+ "numIps": 100},
+ {"base": "10.42.33.1", "mask": "/24", "name": "eth6",
+ "numIps": 100}
+ ]}
+ ],
+ 'tsGroups': [
+ {
+ 'testCases': [{
+ 'type': 'SGW_Node',
+ 'name': '',
+ 'linked': False,
+ 'AssociatedPhys': '',
+ 'parameters': {
+ 'SgiPtpTunnelEn': 'false',
+ 'Gtp2Imsi': '505024101215074',
+ 'Sessions': '100000',
+ 'S5Protocol': 'GTPv2',
+ 'TrafficMtu': '1500',
+ 'Gtp2Version': '13.6.0',
+ 'BearerV4AddrPool': '1.0.0.1',
+ 'Gtp2Imei': '50502410121507',
+ 'PgwNodeEn': 'true',
+ 'DedicatedsPerDefaultBearer': '0',
+ 'DefaultBearers': '1',
+ 'SgwUserAddr': {
+ 'numLinksOrNodes': 1,
+ 'phy': 'eth1',
+ 'forcedEthInterface': '',
+ 'ip': 'SGW_USER_IP',
+ 'class': 'TestNode',
+ 'ethStatsEnabled': False,
+ 'mtu': 1500
+ },
+ 'SgwControlAddr': {
+ 'numLinksOrNodes': 1,
+ 'phy': 'eth1',
+ 'forcedEthInterface': '',
+ 'ip': 'SGW_CONTROL_IP',
+ 'class': 'TestNode',
+ 'ethStatsEnabled': False,
+ 'mtu': 1500,
+ 'nextHop': 'SGW_CONTROL_NEXT_HOP'
+ },
+ 'BearerAddrPool': '2001::1',
+ 'TestType': 'SGW-NODE'
+ }
+ }],
+ 'tsId': TEST_SERVERS[0]['name']},
+ {
+ 'testCases': [{
+ 'type': 'SGW_Nodal',
+ 'name': '',
+ 'parameters': {
+ 'DataTraffic': 'Continuous',
+ 'TrafficStartType': 'When All Sessions Established',
+ 'NetworkHost': 'Local',
+ 'Gtp2Imsi': '505024101215074',
+ 'Dmf': {
+ 'mainflows': [
+ {
+ 'name': 'Basic UDP',
+ 'library': 'test'
+ }
+ ],
+ 'class': 'Dmf',
+ 'instanceGroups': [
+ {
+ 'startPaused': False,
+ 'rate': 0,
+ 'mainflowIdx': 0,
+ 'mixType': ''
+ }
+ ]
+ },
+ 'S5Protocol': 'GTPv2',
+ 'DataUserCfgFileEn': 'false',
+ 'PgwUserSutEn': 'false',
+ 'MmeControlAddr': {
+ 'numLinksOrNodes': 1,
+ 'phy': 'eth1',
+ 'forcedEthInterface': '',
+ 'ip': 'MME_CONTROL_IP',
+ 'class': 'TestNode',
+ 'ethStatsEnabled': False,
+ 'mtu': 1500
+ },
+ 'SgwUserSut': {
+ 'class': 'Sut',
+ 'name': 'SGW_USER_NAME'
+ },
+ 'TestActivity': 'Capacity Test',
+ 'NetworkHostAddrLocal': {
+ 'numLinksOrNodes': 1,
+ 'phy': 'eth2',
+ 'forcedEthInterface': '',
+ 'ip': 'NET_HOST_IP',
+ 'class': 'TestNode',
+ 'ethStatsEnabled': False,
+ 'mtu': 1500
+ },
+ 'DedicatedsPerDefaultBearer': '0',
+ 'DisconnectRate': '1000.0',
+ 'Sessions': '100000',
+ 'SgwSut': {
+ 'class': 'Sut',
+ 'name': 'SGW_CONTROL_NAME'
+ },
+ 'TrafficMtu': '1500',
+ 'Gtp2Version': '13.6.0',
+ 'Gtp2Imei': '50502410121507',
+ 'PgwNodeEn': 'false',
+ 'StartRate': '1000.0',
+ 'PgwV4Sut': {
+ 'class': 'Sut',
+ 'name': 'PGW_SUT_NAME'
+ },
+ 'DefaultBearers': '1',
+ 'EnbUserAddr': {
+ 'numLinksOrNodes': 1,
+ 'phy': 'eth1',
+ 'forcedEthInterface': '',
+ 'ip': 'ENB_USER_IP',
+ 'class': 'TestNode',
+ 'ethStatsEnabled': False,
+ 'mtu': 1500
+ },
+ 'TestType': 'SGW-NODAL'
+ }
+ }],
+ 'tsId': TEST_SERVERS[1]['name']
+ }
+ ]
+}
+
+
+class TestLandslideResourceHelper(unittest.TestCase):
+ TEST_TERMINATED = 1
+ SUCCESS_RECORD_ID = 11
+
+ TEST_RESULTS_DATA = {
+ "interval": 0,
+ "elapsedTime": 138,
+ "actualTime": 1521548057296,
+ "iteration": 1,
+ "tabs": {
+ "Test Summary": {
+ "Start Time": "Tue Mar 20 07:11:55 CDT 2018",
+ "Attempted Dedicated Bearer Session Connects": "0",
+ "Attempted Dedicated Bearer Session Disconnects": "0",
+ "Actual Dedicated Bearer Session Connects": "0",
+ "Actual Dedicated Bearer Session Disconnects": "0",
+ "Dedicated Bearer Sessions Pending": "0",
+ "Dedicated Bearer Sessions Established": "0"
+ }}}
+
+ def setUp(self):
+ self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
+ self.mock_lsapi.start()
+
+ mock_env_helper = mock.Mock()
+ self.res_helper = tg_landslide.LandslideResourceHelper(mock_env_helper)
+ self.res_helper._url = EXAMPLE_URL
+
+ self.addCleanup(self._cleanup)
+
+ def _cleanup(self):
+ self.mock_lsapi.stop()
+ self.res_helper._url = None
+
+ def test___init__(self, *args):
+ self.assertIsInstance(self.res_helper,
+ tg_landslide.LandslideResourceHelper)
+ self.assertEqual({}, self.res_helper._result)
+
+ def test_terminate(self):
+ self.assertRaises(NotImplementedError, self.res_helper.terminate)
+
+ def test_collect_kpi_test_running(self):
+ self.assertRaises(NotImplementedError, self.res_helper.collect_kpi)
+
+
+class TestLandslideTclClient(unittest.TestCase):
+ def setUp(self):
+ self.mock_tcl_handler = mock.Mock(spec=tg_landslide.LsTclHandler)
+ self.ls_res_helper = mock.Mock(
+ spec=tg_landslide.LandslideResourceHelper)
+ self.ls_tcl_client = tg_landslide.LandslideTclClient(
+ self.mock_tcl_handler,
+ self.ls_res_helper)
+
+ def test___init__(self, *args):
+ self.ls_tcl_client = tg_landslide.LandslideTclClient(
+ self.mock_tcl_handler,
+ self.ls_res_helper)
+ self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
+ self.assertIsNone(self.ls_tcl_client._user)
+ self.assertIsNone(self.ls_tcl_client._library_id)
+ self.assertIsNone(self.ls_tcl_client._basic_library_id)
+ self.assertEqual(set(), self.ls_tcl_client.ts_ids)
+ self.assertIsInstance(self.ls_tcl_client._tc_types, set)
+ self.assertIsNotNone(self.ls_tcl_client._tc_types)
+
+ def test_connect_login_success(self, *args):
+ lib_id = '123'
+ exec_responses = ['java0x2', lib_id, lib_id]
+ auth = ('user', 'password')
+ self.mock_tcl_handler.execute.side_effect = exec_responses
+ self.ls_tcl_client.connect(TAS_INFO['ip'], *auth)
+ self.assertEqual(lib_id, self.ls_tcl_client._library_id)
+ self.assertEqual(lib_id, self.ls_tcl_client._basic_library_id)
+ self.assertEqual(TAS_INFO['ip'], self.ls_tcl_client.tcl_server_ip)
+ self.assertEqual(auth[0], self.ls_tcl_client._user)
+ self.assertEqual(len(exec_responses),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test_connect_login_failed(self, *args):
+ exec_responses = ['Login failed']
+ auth = ('user', 'password')
+ self.mock_tcl_handler.execute.side_effect = exec_responses
+ self.assertRaises(exceptions.LandslideTclException,
+ self.ls_tcl_client.connect,
+ TAS_INFO['ip'],
+ *auth)
+ self.assertIsNone(self.ls_tcl_client._library_id)
+ self.assertIsNone(self.ls_tcl_client._basic_library_id)
+ self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
+ self.assertIsNone(self.ls_tcl_client._user)
+ self.assertEqual(len(exec_responses),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test_disconnect(self, *args):
+ self.ls_tcl_client.disconnect()
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+ self.assertIsNone(self.ls_tcl_client.tcl_server_ip)
+ self.assertIsNone(self.ls_tcl_client._user)
+ self.assertIsNone(self.ls_tcl_client._library_id)
+ self.assertIsNone(self.ls_tcl_client._basic_library_id)
+
+ def test_create_test_server(self, *args):
+ return_value = '2'
+ self.ls_tcl_client._ts_context.vnfd_helper = \
+ VNFD['vnfd:vnfd-catalog']['vnfd'][0]
+ self.ls_tcl_client._ts_context.license_data = {'lic_id': return_value}
+ self.mock_tcl_handler.execute.return_value = return_value
+ self.ls_tcl_client._set_thread_model = mock.Mock()
+ res = self.ls_tcl_client.create_test_server(TEST_SERVERS[1])
+ self.assertEqual(3, self.mock_tcl_handler.execute.call_count)
+ self.ls_tcl_client._set_thread_model.assert_called_once_with(
+ TEST_SERVERS[1]['name'],
+ TEST_SERVERS[1]['thread_model'])
+ self.assertEqual(int(return_value), res)
+
+ def test_create_test_server_fail_limit_reach(self, *args):
+ self.mock_tcl_handler.execute.side_effect = ['TS not found',
+ 'Add failed']
+ self.assertRaises(RuntimeError,
+ self.ls_tcl_client.create_test_server,
+ TEST_SERVERS[0])
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test__add_test_server(self):
+ ts_id = '2'
+ self.mock_tcl_handler.execute.side_effect = ['TS not found', ts_id]
+ self.assertEqual(ts_id,
+ self.ls_tcl_client._add_test_server('name', 'ip'))
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test__add_test_server_failed(self):
+ self.mock_tcl_handler.execute.side_effect = ['TS not found',
+ 'Add failed']
+ self.assertRaises(RuntimeError, self.ls_tcl_client._add_test_server,
+ 'name', 'ip')
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test__update_license(self):
+ curr_lic_id = '111'
+ new_lic_id = '222'
+ exec_resp = ['java0x4',
+ curr_lic_id,
+ TCL_SUCCESS_RESPONSE,
+ TCL_SUCCESS_RESPONSE]
+ self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.ls_tcl_client._update_license('name')
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test__update_license_same_as_current(self):
+ curr_lic_id = '111'
+ new_lic_id = '111'
+ exec_resp = ['java0x4', curr_lic_id]
+ self.ls_tcl_client._ts_context.license_data = {'lic_id': new_lic_id}
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.ls_tcl_client._update_license('name')
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test__set_thread_model_update_needed(self):
+ self.ls_tcl_client._ts_context.vnfd_helper = {
+ 'mgmt-interface': {
+ 'cfguser_password': 'cfguser_password'
+ }
+ }
+ exec_resp = ['java0x4', 'V0', '', '']
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.ls_tcl_client._set_thread_model('name', 'Fireball')
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test__set_thread_model_no_update_needed(self):
+ self.ls_tcl_client._ts_context.vnfd_helper = {
+ 'mgmt-interface': {
+ 'cfguser_password': 'cfguser_password'
+ }
+ }
+ exec_resp = ['java0x4', 'V0']
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.ls_tcl_client._set_thread_model('name', 'Legacy')
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ @mock.patch.object(tg_landslide.LandslideTclClient,
+ 'resolve_test_server_name', return_value='2')
+ def test_create_test_session(self, *args):
+ self.ls_tcl_client._save_test_session = mock.Mock()
+ self.ls_tcl_client._configure_ts_group = mock.Mock()
+ self.ls_tcl_client.create_test_session(SESSION_PROFILE)
+ self.assertEqual(20, self.mock_tcl_handler.execute.call_count)
+
+ def test_create_dmf(self):
+ self.mock_tcl_handler.execute.return_value = '2'
+ self.ls_tcl_client._save_dmf = mock.Mock()
+ self.ls_tcl_client.create_dmf(DMF_CFG)
+ self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
+
+ def test_configure_dmf(self):
+ self.mock_tcl_handler.execute.return_value = '2'
+ self.ls_tcl_client._save_dmf = mock.Mock()
+ self.ls_tcl_client.configure_dmf(DMF_CFG)
+ self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
+
+ def test_delete_dmf(self):
+ self.assertRaises(NotImplementedError,
+ self.ls_tcl_client.delete_dmf,
+ DMF_CFG)
+
+ def test__save_dmf_valid(self):
+ exec_resp = [TCL_SUCCESS_RESPONSE, TCL_SUCCESS_RESPONSE]
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.ls_tcl_client._save_dmf()
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test__save_dmf_invalid(self):
+ exec_resp = ['Invalid', 'List of errors and warnings']
+ self.mock_tcl_handler.execute.side_effect = exec_resp
+ self.assertRaises(exceptions.LandslideTclException,
+ self.ls_tcl_client._save_dmf)
+ self.assertEqual(len(exec_resp),
+ self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_report_options(self):
+ _options = {'format': 'CSV', 'PerInterval': False}
+ self.ls_tcl_client._configure_report_options(_options)
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test___configure_ts_group(self, *args):
+ self.ls_tcl_client._configure_tc_type = mock.Mock()
+ self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
+ self.ls_tcl_client.resolve_test_server_name = mock.Mock(
+ return_value='2')
+ self.ls_tcl_client._configure_ts_group(
+ SESSION_PROFILE['tsGroups'][0], 0)
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+
+ def test___configure_ts_group_resolve_ts_fail(self, *args):
+ self.ls_tcl_client._configure_tc_type = mock.Mock()
+ self.ls_tcl_client._configure_preresolved_arp = mock.Mock()
+ self.ls_tcl_client.resolve_test_server_name = mock.Mock(
+ return_value='TS Not Found')
+ self.assertRaises(RuntimeError, self.ls_tcl_client._configure_ts_group,
+ SESSION_PROFILE['tsGroups'][0], 0)
+ self.assertEqual(0, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_tc_type(self):
+ _tc = SESSION_PROFILE['tsGroups'][0]['testCases'][0]
+ self.mock_tcl_handler.execute.return_value = TCL_SUCCESS_RESPONSE
+ self.ls_tcl_client._configure_parameters = mock.Mock()
+ self.ls_tcl_client._configure_tc_type(_tc, 0)
+ self.assertEqual(7, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_tc_type_wrong_type(self):
+ _tc = SESSION_PROFILE['tsGroups'][0]['testCases'][0]
+ _tc['type'] = 'not_supported'
+ self.ls_tcl_client._configure_parameters = mock.Mock()
+ self.assertRaises(RuntimeError,
+ self.ls_tcl_client._configure_tc_type,
+ _tc, 0)
+
+ def test__configure_tc_type_not_found_basic_lib(self):
+ _tc = SESSION_PROFILE['tsGroups'][0]['testCases'][0]
+ self.ls_tcl_client._configure_parameters = mock.Mock()
+ self.mock_tcl_handler.execute.return_value = 'Invalid'
+ self.assertRaises(RuntimeError,
+ self.ls_tcl_client._configure_tc_type,
+ _tc, 0)
+ self.assertEqual(0, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_parameters(self):
+ _params = SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters']
+ self.ls_tcl_client._configure_parameters(_params)
+ self.assertEqual(16, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_array_param(self):
+ _array = {"class": "Array",
+ "array": ["0"]}
+ self.ls_tcl_client._configure_array_param('name', _array)
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_test_node_param(self):
+ _params = SESSION_PROFILE['tsGroups'][0]['testCases'][0]['parameters']
+ self.ls_tcl_client._configure_test_node_param('SgwUserAddr',
+ _params['SgwUserAddr'])
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_sut_param(self):
+ _params = {'name': 'name'}
+ self.ls_tcl_client._configure_sut_param('name', _params)
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_dmf_param(self):
+ _params = {"mainflows": [{"library": '111',
+ "name": "Basic UDP"}],
+ "instanceGroups": [{
+ "mainflowIdx": 0,
+ "mixType": "",
+ "rate": 0.0,
+ "rows": [{
+ "clientPort": 0,
+ "context": 0,
+ "node": 0,
+ "overridePort": False,
+ "ratingGroup": 0,
+ "role": 0,
+ "serviceId": 0,
+ "transport": "Any"}]
+ }]}
+ self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
+ res = self.ls_tcl_client._configure_dmf_param('name', _params)
+ self.assertEqual(5, self.mock_tcl_handler.execute.call_count)
+ self.assertIsNone(res)
+
+ def test__configure_dmf_param_no_instance_groups(self):
+ _params = {"mainflows": [{"library": '111',
+ "name": "Basic UDP"}]}
+ self.ls_tcl_client._get_library_id = mock.Mock(return_value='111')
+ res = self.ls_tcl_client._configure_dmf_param('name', _params)
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+ self.assertIsNone(res)
+
+ def test__configure_reservation(self):
+ _reservation = SESSION_PROFILE['reservations'][0]
+ self.ls_tcl_client.resolve_test_server_name = mock.Mock(
+ return_value='2')
+ res = self.ls_tcl_client._configure_reservation(_reservation)
+ self.assertIsNone(res)
+ self.assertEqual(6, self.mock_tcl_handler.execute.call_count)
+
+ def test__configure_preresolved_arp(self):
+ _arp = [{'StartingAddress': '10.81.1.10',
+ 'NumNodes': 1}]
+ res = self.ls_tcl_client._configure_preresolved_arp(_arp)
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+ self.assertIsNone(res)
+
+ def test__configure_preresolved_arp_none(self):
+ res = self.ls_tcl_client._configure_preresolved_arp(None)
+ self.assertIsNone(res)
+
+ def test_delete_test_session(self):
+ self.assertRaises(NotImplementedError,
+ self.ls_tcl_client.delete_test_session, {})
+
+ def test__save_test_session(self):
+ self.mock_tcl_handler.execute.side_effect = [TCL_SUCCESS_RESPONSE,
+ TCL_SUCCESS_RESPONSE]
+ res = self.ls_tcl_client._save_test_session()
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+ self.assertIsNone(res)
+
+ def test__save_test_session_invalid(self):
+ self.mock_tcl_handler.execute.side_effect = ['Invalid', 'Errors']
+ self.assertRaises(exceptions.LandslideTclException,
+ self.ls_tcl_client._save_test_session)
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+ def test__get_library_id_system_lib(self):
+ self.mock_tcl_handler.execute.side_effect = ['111']
+ res = self.ls_tcl_client._get_library_id('name')
+ self.assertEqual(1, self.mock_tcl_handler.execute.call_count)
+ self.assertEqual('111', res)
+
+ def test__get_library_id_user_lib(self):
+ self.mock_tcl_handler.execute.side_effect = ['Not found', '222']
+ res = self.ls_tcl_client._get_library_id('name')
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+ self.assertEqual('222', res)
+
+ def test__get_library_id_exception(self):
+ self.mock_tcl_handler.execute.side_effect = ['Not found', 'Not found']
+ self.assertRaises(exceptions.LandslideTclException,
+ self.ls_tcl_client._get_library_id,
+ 'name')
+ self.assertEqual(2, self.mock_tcl_handler.execute.call_count)
+
+
+class TestLsTclHandler(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_lsapi = mock.patch.object(tg_landslide, 'LsApi')
+ self.mock_lsapi.start()
+
+ self.addCleanup(self._cleanup)
+
+ def _cleanup(self):
+ self.mock_lsapi.stop()
+
+ def test___init__(self, *args):
+ self.ls_tcl_handler = tg_landslide.LsTclHandler()
+ self.assertEqual({}, self.ls_tcl_handler.tcl_cmds)
+ self.ls_tcl_handler._ls.tcl.assert_called_once()
+
+ def test_execute(self, *args):
+ self.ls_tcl_handler = tg_landslide.LsTclHandler()
+ self.ls_tcl_handler.execute('command')
+ self.assertIn('command', self.ls_tcl_handler.tcl_cmds)
Code Review / yardstick.git / commitdiff