# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import absolute_import
-from __future__ import print_function
-
import sys
import os
import logging
+import collections
+import subprocess
+try:
+ libs = subprocess.check_output(
+ 'python -c "import site; print(site.getsitepackages())"', shell=True)
+
+ sys.path.extend(libs[1:-1].replace("'", "").split(','))
+except subprocess.CalledProcessError:
+ pass
# ixload uses its own py2. So importing jsonutils fails. So adding below
# workaround to support call from yardstick
except ImportError:
import json as jsonutils
-from yardstick.common.utils import join_non_strings
-from yardstick.common.utils import ErrorClass
+from yardstick.common import exceptions #pylint: disable=wrong-import-position
try:
from IxLoad import IxLoad, StatCollectorUtils
except ImportError:
- IxLoad = ErrorClass
- StatCollectorUtils = ErrorClass
+ IxLoad = exceptions.ErrorClass
+ StatCollectorUtils = exceptions.ErrorClass
+
LOG = logging.getLogger(__name__)
CSV_FILEPATH_NAME = 'IxL_statResults.csv'
"""
+def validate_non_string_sequence(value, default=None, raise_exc=None):
+ if isinstance(value, collections.Sequence) and not isinstance(value, str):
+ return value
+ if raise_exc:
+ raise raise_exc # pylint: disable=raising-bad-type
+ return default
+
+
+def join_non_strings(separator, *non_strings):
+ try:
+ non_strings = validate_non_string_sequence(non_strings[0], raise_exc=RuntimeError)
+ except (IndexError, RuntimeError):
+ pass
+ return str(separator).join(str(non_string) for non_string in non_strings)
+
+
class IXLOADHttpTest(object):
def __init__(self, test_input):
- self.test_input = jsonutils.loads(test_input)
- self.parse_run_test()
self.ix_load = None
self.stat_utils = None
self.remote_server = None
self.chassis = None
self.card = None
self.ports_to_reassign = None
+ self.links_param = None
+ self.test_input = jsonutils.loads(test_input)
+ self.parse_run_test()
+ self.test = None
@staticmethod
def format_ports_for_reassignment(ports):
LOG.error('Error: IxLoad config file not found: %s', config_file)
raise
+ def update_network_address(self, net_traffic, address, gateway, prefix):
+ """Update ip address and gateway for net_traffic object
+
+ This function update field which configure source addresses for
+ traffic which is described by net_traffic object.
+ Do not return anything
+
+ :param net_traffic: (IxLoadObjectProxy) proxy obj to tcl net_traffic object
+ :param address: (str) Ipv4 range start address
+ :param gateway: (str) Ipv4 address of gateway
+ :param prefix: (int) subnet prefix
+ :return:
+ """
+ try:
+ ethernet = net_traffic.network.getL1Plugin()
+ ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
+ ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
+ ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
+
+ ix_net_ip_v4_v6_range.config(
+ prefix=prefix,
+ ipAddress=address,
+ gatewayAddress=gateway)
+ except Exception:
+ raise exceptions.InvalidRxfFile
+
+ def update_network_mac_address(self, net_traffic, mac):
+ """Update MACaddress for net_traffic object
+
+ This function update field which configure MACaddresses for
+ traffic which is described by net_traffic object.
+ If mac == "auto" then will be configured auto generated mac
+ Do not return anything.
+
+ :param net_traffic: (IxLoadObjectProxy) proxy obj to tcl net_traffic object
+ :param mac: (str) MAC
+ :return:
+ """
+ try:
+ ethernet = net_traffic.network.getL1Plugin()
+ ix_net_l2_ethernet_plugin = ethernet.childrenList[0]
+ ix_net_ip_v4_v6_plugin = ix_net_l2_ethernet_plugin.childrenList[0]
+ ix_net_ip_v4_v6_range = ix_net_ip_v4_v6_plugin.rangeList[0]
+
+ if str(mac).lower() == "auto":
+ ix_net_ip_v4_v6_range.config(autoMacGeneration=True)
+ else:
+ ix_net_ip_v4_v6_range.config(autoMacGeneration=False)
+ mac_range = ix_net_ip_v4_v6_range.getLowerRelatedRange(
+ "MacRange")
+ mac_range.config(mac=mac)
+ except Exception:
+ raise exceptions.InvalidRxfFile
+
+ def update_network_param(self, net_traffic, param):
+ """Update net_traffic by parameters specified in param"""
+
+ self.update_network_address(net_traffic, param["address"],
+ param["gateway"], param["subnet_prefix"])
+
+ self.update_network_mac_address(net_traffic, param["mac"])
+
+ def update_config(self):
+ """Update some fields by parameters from traffic profile"""
+
+ net_traffics = {}
+ # self.test.communityList is a IxLoadObjectProxy to some tcl object
+ # which contain all net_traffic objects in scenario.
+ # net_traffic item has a name in format "activity_name@item_name"
+ try:
+ for item in self.test.communityList:
+ net_traffics[item.name.split('@')[1]] = item
+ except Exception: # pylint: disable=broad-except
+ pass
+
+ for name, net_traffic in net_traffics.items():
+ try:
+ param = self.links_param[name]
+ except KeyError:
+ LOG.debug('There is no param for net_traffic %s', name)
+ continue
+
+ self.update_network_param(net_traffic, param["ip"])
+ if "uplink" in name:
+ self.update_http_client_param(net_traffic, param["http_client"])
+
+ def update_http_client_param(self, net_traffic, param):
+ """Update http client object in net_traffic
+
+ Update http client object in net_traffic by parameters
+ specified in param.
+ Do not return anything.
+
+ :param net_traffic: (IxLoadObjectProxy) proxy obj to tcl net_traffic object
+ :param param: (dict) http_client section from traffic profile
+ :return:
+ """
+ page = param.get("page_object")
+ if page:
+ self.update_page_size(net_traffic, page)
+ users = param.get("simulated_users")
+ if users:
+ self.update_user_count(net_traffic, users)
+
+ def update_page_size(self, net_traffic, page_object):
+ """Update page_object field in http client object in net_traffic
+
+ This function update field which configure page_object
+ which will be loaded from server
+ Do not return anything.
+
+ :param net_traffic: (IxLoadObjectProxy) proxy obj to tcl net_traffic object
+ :param page_object: (str) path to object on server e.g. "/4k.html"
+ :return:
+ """
+ try:
+ activity = net_traffic.activityList[0]
+ ix_http_command = activity.agent.actionList[0]
+ ix_http_command.config(pageObject=page_object)
+ except Exception:
+ raise exceptions.InvalidRxfFile
+
+ def update_user_count(self, net_traffic, user_count):
+ """Update userObjectiveValue field in activity object in net_traffic
+
+ This function update field which configure users count
+ which will be simulated by client.
+ Do not return anything.
+
+ :param net_traffic: (IxLoadObjectProxy) proxy obj to tcl net_traffic object
+ :param user_count: (int) number of simulated users
+ :return:
+ """
+ try:
+ activity = net_traffic.activityList[0]
+ activity.config(userObjectiveValue=user_count)
+ except Exception:
+ raise exceptions.InvalidRxfFile
+
def start_http_test(self):
self.ix_load = IxLoad()
# Get the first test on the testList
test_name = repository.testList[0].cget("name")
- test = repository.testList.getItem(test_name)
+ self.test = repository.testList.getItem(test_name)
self.set_results_dir(test_controller, self.results_on_windows)
- test.config(statsRequired=1, enableResetPorts=1, csvInterval=2,
- enableForceOwnership=True)
+ self.test.config(statsRequired=1, enableResetPorts=1, csvInterval=2,
+ enableForceOwnership=True)
+
+ self.update_config()
# ---- Remap ports ----
try:
- self.reassign_ports(test, repository, self.ports_to_reassign)
- except Exception:
+ self.reassign_ports(self.test, repository, self.ports_to_reassign)
+ except Exception: # pylint: disable=broad-except
LOG.exception("Exception occurred during reassign_ports")
# -----------------------------------------------------------------------
self.stat_utils.StartCollector(self.IxL_StatCollectorCommand)
- test_controller.run(test)
+ test_controller.run(self.test)
self.ix_load.waitForTestFinish()
test_controller.releaseConfigWaitFinish()
test_controller.generateReport(detailedReport=1, format="PDF;HTML")
test_controller.releaseConfigWaitFinish()
- self.ix_load.delete(test)
+ self.ix_load.delete(self.test)
self.ix_load.delete(test_controller)
self.ix_load.delete(logger)
self.ix_load.delete(log_engine)
LOG.debug("Ports to be reassigned: %s", self.ports_to_reassign)
+ self.links_param = self.test_input["links_param"]
+ LOG.debug("Links param to be applied: %s", self.links_param)
+
def main(args):
# Get the args from cmdline and parse and run the test
ixload_obj.start_http_test()
if __name__ == '__main__':
+ LOG.info("Start http_ixload test")
main(sys.argv)
Code Review / yardstick.git / blobdiff