set duration [dict get $testSpec duration]
+ set L2CountValue 1
+ set L2Increment False
+ set L3ValueType singleValue
+ set L3CountValue 1
+ set L4ValueType singleValue
+ set L4CountValue 1
+
# RFC2544 to IXIA terminology mapping (it affects Ixia configuration inside this script):
# Test => Trial
# Trial => Iteration
}
set multipleStreams [dict get $testSpec multipleStreams]
+ set streamType [dict get $testSpec streamType]
+
if {($multipleStreams < 0)} {
- set multipleStreams 0
+ set multipleStreams 0
}
- set numflows 64000
if {$multipleStreams} {
- set numflows $multipleStreams
- set multipleStreams increment
- } else {
- set multipleStreams singleValue
+ if {($streamType == "L2")} {
+ set L2CountValue $multipleStreams
+ set L2Increment True
+ } elseif {($streamType == "L3")} {
+ set L3ValueType increment
+ set L3CountValue $multipleStreams
+ } else {
+ set L4ValueType increment
+ set L4CountValue $multipleStreams
+ }
}
set fastConvergence True
set sg_lan [ixNet add $ixNetSG_Stack(1)/protocols/static lan]
ixNet setMultiAttrs $sg_lan \
-atmEncapsulation ::ixNet::OBJ-null \
- -count 1 \
+ -count $L2CountValue \
-countPerVc 1 \
- -enableIncrementMac False \
+ -enableIncrementMac $L2Increment \
-enableIncrementVlan False \
-enableSiteId False \
-enableVlan False \
set sg_lan [ixNet add $ixNetSG_Stack(1)/protocols/static lan]
ixNet setMultiAttrs $sg_lan \
-atmEncapsulation ::ixNet::OBJ-null \
- -count 1 \
+ -count $L2CountValue \
-countPerVc 1 \
- -enableIncrementMac False \
+ -enableIncrementMac $L2Increment \
-enableIncrementVlan False \
-enableSiteId False \
-enableVlan False \
-auto False \
-randomMask {0.0.0.0} \
-trackingEnabled False \
- -valueType $multipleStreams \
+ -valueType $L3ValueType \
-activeFieldChoice False \
-startValue $dstIp \
- -countValue $numflows
+ -countValue $L3CountValue
#sg_commit
#set sg_field [lindex [ixNet remapIds $sg_field] 0]
-auto False \
-randomMask {63} \
-trackingEnabled False \
- -valueType $multipleStreams \
+ -valueType $L4ValueType \
-activeFieldChoice False \
-startValue {0} \
- -countValue $numflows
+ -countValue $L4CountValue
#
# configuring the object that corresponds to /traffic/trafficItem:1/configElement:1/stack:"udp-3"/field:"udp.header.length-3"
-# Copyright 2015 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# conventions
TEST_PARAMS = {}
+# delay enforced after every step to allow system to process changes
+TEST_STEP_DELAY = 5
+
# ############################
# Modules
# ############################
# # override any values defined by TEST_PARAMS option
# # stated in configuration files or values specified
# # on command line through --test-params parameter.
+#
+# "TestSteps": [] # Definition of detailed test steps.
+# # In case that this list is defined, then
+# # vsperf will execute defined test steps
+# # one by one. It can be used to configure
+# # vswitch, insert flows and transmit traffic.
+# # It is possible to refer to result of any
+# # previous step through #STEP[i][j] macro.
+# # Where i is a number of step (starts from 0)
+# # and j is index of result returned by step i.
# "Test Modifier": [FrameMod|Other],
# "Dependency": [Test_Case_Name |None],
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
def __init__(self):
pass
+ def _eval_param(self, param):
+ # pylint: disable=invalid-name
+ """ Helper function for expansion of references to vsperf parameters
+ """
+ if isinstance(param, str):
+ # evaluate every #PARAM reference inside parameter itself
+ macros = re.findall(r'#PARAM\((([\w\-]+)(\[[\w\[\]\-\'\"]+\])*)\)', param)
+ if macros:
+ for macro in macros:
+ # pylint: disable=eval-used
+ try:
+ tmp_val = str(eval("self.getValue('{}'){}".format(macro[1], macro[2])))
+ param = param.replace('#PARAM({})'.format(macro[0]), tmp_val)
+ # silently ignore that option required by PARAM macro can't be evaluated;
+ # It is possible, that referred parameter will be constructed during runtime
+ # and re-read later.
+ except IndexError:
+ pass
+ except AttributeError:
+ pass
+ return param
+ elif isinstance(param, list) or isinstance(param, tuple):
+ tmp_list = []
+ for item in param:
+ tmp_list.append(self._eval_param(item))
+ return tmp_list
+ elif isinstance(param, dict):
+ tmp_dict = {}
+ for (key, value) in param.items():
+ tmp_dict[key] = self._eval_param(value)
+ return tmp_dict
+ else:
+ return param
+
def getValue(self, attr):
"""Return a settings item value
"""
if attr == 'TRAFFIC':
tmp_value = copy.deepcopy(master_value)
tmp_value = merge_spec(tmp_value, cli_value)
- return tmp_value
+ return self._eval_param(tmp_value)
else:
- return cli_value
+ return self._eval_param(cli_value)
else:
- return master_value
+ return self._eval_param(master_value)
else:
raise AttributeError("%r object has no attribute %r" %
(self.__class__, attr))
for key in self.__dict__:
if key.startswith('GUEST_'):
value = self.getValue(key)
- if isinstance(value, str) and value.find('#') >= 0:
+ if isinstance(value, str) and str(value).find('#') >= 0:
self._expand_vm_settings(key, 1)
if isinstance(value, list):
def validate_getValue(self, result, attr):
"""Verifies, that correct value was returned
"""
- assert result == self.__dict__[attr]
+ # getValue must be called to expand macros and apply
+ # values from TEST_PARAM option
+ assert result == self.getValue(attr)
return True
def validate_setValue(self, dummy_result, name, value):
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
#
# bidirectional testing for OP2P is not yet supported.
# TRAFFIC['bidir'] must be set to 'False'.
-#
-# "TestSteps": [] # Definition of integration test steps.
-# # In case that this list is defined, then
-# # vsperf will execute defined test steps
-# # one by one. It can be used to configure
-# # vswitch, insert flows and transmit traffic.
-# # It is possible to refer to result of any
-# # previous step through #STEP[i][j] macro.
-# # Where i is a number of step (starts from 0)
-# # and j is index of result returned by step i.
#
# Common TestSteps parts ("macros")
"TestSteps": STEP_VSWITCH_PVVP_INIT + # STEP 0-6
[
# check that at least 2 numa slots are available
- ['tools', 'exec', 'numactl -H', 'available: ([0-9]+)'], # STEP 7
+ ['tools', 'exec_shell', 'numactl -H', 'available: ([0-9]+)'], # STEP 7
['tools', 'assert', '#STEP[-1][0]>1'], # STEP 8
# store last 2 cores from numa slot 0
- ['tools', 'exec', 'numactl -H', 'node 0 cpus:.*\s+(\\d+) (\\d+)$'], # STEP 9
+ ['tools', 'exec_shell', 'numactl -H', 'node 0 cpus:.*\s+(\\d+) (\\d+)$'], # STEP 9
# store last 2 cores from numa slot 1
- ['tools', 'exec', 'numactl -H', 'node 1 cpus:.*\s+(\\d+) (\\d+)$'], # STEP 10
+ ['tools', 'exec_shell', 'numactl -H', 'node 1 cpus:.*\s+(\\d+) (\\d+)$'], # STEP 10
# pin VNF1 to 1st NUMA slot and VNF2 to 2nd NUMA slot
['settings', 'setValue', 'GUEST_CORE_BINDING', # STEP 11
[("#STEP[-2][0][0]", "#STEP[-2][0][1]"),
['settings', 'getValue', 'TOOLS'], # STEP 14
# check that PMD thread serving VNF1 runs at NUMA slot 0
## i.e. get numa slot ID serving dpdhvhostuser0...
- ['tools', 'exec', "sudo #STEP[-1]['ovs-appctl'] " # STEP 15
+ ['tools', 'exec_shell', "sudo #STEP[-1]['ovs-appctl'] " # STEP 15
"dpif-netdev/pmd-rxq-show | "
"sed -e '/dpdkvhostuser0/,$d' | tac",
'pmd thread numa_id ([0-9])+'
['tools', 'assert', '#STEP[-1][0]==0'], # STEP 16
# check that PMD thread serving VNF2 runs at NUMA slot 1
## i.e. get numa slot ID serving dpdhvhostuser2...
- ['tools', 'exec', "sudo #STEP[-3]['ovs-appctl'] " # STEP 17
+ ['tools', 'exec_shell', "sudo #STEP[-3]['ovs-appctl'] " # STEP 17
"dpif-netdev/pmd-rxq-show | "
"sed -e '/dpdkvhostuser2/,$d' | tac",
'pmd thread numa_id ([0-9])+'
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
self._lossrate = float(settings.getValue('TRAFFICGEN_LOSSRATE'))
self._packet_sizes = settings.getValue('TRAFFICGEN_PKT_SIZES')
- self._mode = settings.getValue('mode').lower()
+ self._mode = str(settings.getValue('mode')).lower()
self._results = []
def __enter__(self):
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
self._logger.debug('send_traffic with ' +
str(self._traffic_gen_class))
+ # update type with detailed traffic value
+ self._type = traffic['traffic_type']
+
for packet_size in self._packet_sizes:
# Merge framesize with the default traffic definition
if 'l2' in traffic:
result = self._traffic_gen_class.send_rfc2544_throughput(
traffic, tests=self._tests, duration=self._duration, lossrate=self._lossrate)
else:
- raise RuntimeError("Unsupported traffic type {} was \
- detected".format(traffic['traffic_type']))
+ raise RuntimeError("Unsupported traffic type {} was "
+ "detected".format(traffic['traffic_type']))
result = self._append_results(result, packet_size)
self._results.append(result)
self._logger.debug('send_traffic_async with ' +
str(self._traffic_gen_class))
+ # update type with detailed traffic value
+ self._type = traffic['traffic_type']
+
for packet_size in self._packet_sizes:
traffic['l2'] = {'framesize': packet_size}
self._traffic_gen_class.start_rfc2544_throughput(
-# Copyright 2016 Spirent Communications, Intel Corporation.
+# Copyright 2016-2017 Spirent Communications, Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
self._logger.debug('send_traffic with ' +
str(self._traffic_gen_class))
+ # update type with detailed traffic value
+ self._type = traffic['traffic_type']
+
for packet_size in self._packet_sizes:
# Merge framesize with the default traffic definition
if 'l2' in traffic:
self._logger.debug('send_traffic_async with ' +
str(self._traffic_gen_class))
+ # update type with detailed traffic value
+ self._type = traffic['traffic_type']
+
for packet_size in self._packet_sizes:
traffic['l2'] = {'framesize': packet_size}
self._traffic_gen_class.start_rfc2889_forwarding(
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
from core.vswitch_controller import IVswitchController
from vswitches.utils import add_ports_to_flow
-from conf import settings
+from conf import settings as S
from tools import tasks
_FLOW_TEMPLATE = {
if self._tunnel_operation == "encapsulation":
self._setup_encap()
else:
- if settings.getValue('VSWITCH').endswith('Vanilla'):
+ if str(S.getValue('VSWITCH')).endswith('Vanilla'):
self._setup_decap_vanilla()
else:
self._setup_decap()
try:
self._vswitch.start()
- bridge = settings.getValue('TUNNEL_INTEGRATION_BRIDGE')
- bridge_ext = settings.getValue('TUNNEL_EXTERNAL_BRIDGE')
- bridge_ext_ip = settings.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
- tg_port2_mac = settings.getValue('TRAFFICGEN_PORT2_MAC')
- vtep_ip2 = settings.getValue('VTEP_IP2')
+ bridge = S.getValue('TUNNEL_INTEGRATION_BRIDGE')
+ bridge_ext = S.getValue('TUNNEL_EXTERNAL_BRIDGE')
+ bridge_ext_ip = S.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
+ tg_port2_mac = S.getValue('TRAFFICGEN_PORT2_MAC')
+ vtep_ip2 = S.getValue('VTEP_IP2')
self._vswitch.add_switch(bridge)
tasks.run_task(['sudo', 'ip', 'addr', 'add',
- settings.getValue('VTEP_IP1'), 'dev', bridge],
+ S.getValue('VTEP_IP1'), 'dev', bridge],
self._logger, 'Assign ' +
- settings.getValue('VTEP_IP1') + ' to ' + bridge,
+ S.getValue('VTEP_IP1') + ' to ' + bridge,
False)
tasks.run_task(['sudo', 'ip', 'link', 'set', 'dev', bridge, 'up'],
self._logger, 'Bring up ' + bridge, False)
'Set ' + bridge_ext + 'status to up')
self._vswitch.add_route(bridge,
- settings.getValue('VTEP_IP2_SUBNET'),
+ S.getValue('VTEP_IP2_SUBNET'),
bridge_ext)
- if settings.getValue('VSWITCH').endswith('Vanilla'):
+ if str(S.getValue('VSWITCH')).endswith('Vanilla'):
tasks.run_task(['sudo', 'arp', '-s', vtep_ip2, tg_port2_mac],
self._logger,
'Set ' + bridge_ext + ' status to up')
try:
self._vswitch.start()
- bridge = settings.getValue('TUNNEL_INTEGRATION_BRIDGE')
- bridge_ext = settings.getValue('TUNNEL_EXTERNAL_BRIDGE')
- bridge_ext_ip = settings.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
- tgen_ip1 = settings.getValue('TRAFFICGEN_PORT1_IP')
+ bridge = S.getValue('TUNNEL_INTEGRATION_BRIDGE')
+ bridge_ext = S.getValue('TUNNEL_EXTERNAL_BRIDGE')
+ bridge_ext_ip = S.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
+ tgen_ip1 = S.getValue('TRAFFICGEN_PORT1_IP')
self._vswitch.add_switch(bridge)
tasks.run_task(['sudo', 'ip', 'addr', 'add',
- settings.getValue('VTEP_IP1'), 'dev', bridge],
+ S.getValue('VTEP_IP1'), 'dev', bridge],
self._logger, 'Assign ' +
- settings.getValue('VTEP_IP1') + ' to ' + bridge, False)
+ S.getValue('VTEP_IP1') + ' to ' + bridge, False)
tasks.run_task(['sudo', 'ip', 'link', 'set', 'dev', bridge, 'up'],
self._logger, 'Bring up ' + bridge, False)
self._vswitch.add_phy_port(bridge)
(_, phy2_number) = self._vswitch.add_phy_port(bridge_ext)
if tunnel_type == "vxlan":
- vxlan_vni = 'options:key=' + settings.getValue('VXLAN_VNI')
+ vxlan_vni = 'options:key=' + S.getValue('VXLAN_VNI')
(_, phy3_number) = self._vswitch.add_tunnel_port(bridge_ext,
tgen_ip1,
tunnel_type,
'Set ' + bridge_ext + ' status to up')
self._vswitch.set_tunnel_arp(tgen_ip1,
- settings.getValue('TRAFFICGEN_PORT1_MAC'),
+ S.getValue('TRAFFICGEN_PORT1_MAC'),
bridge)
# Test is unidirectional for now
self._vswitch.del_flow(bridge_ext)
try:
self._vswitch.start()
- bridge = settings.getValue('TUNNEL_INTEGRATION_BRIDGE')
- bridge_ext = settings.getValue('TUNNEL_EXTERNAL_BRIDGE')
- bridge_ext_ip = settings.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
- tgen_ip1 = settings.getValue('TRAFFICGEN_PORT1_IP')
+ bridge = S.getValue('TUNNEL_INTEGRATION_BRIDGE')
+ bridge_ext = S.getValue('TUNNEL_EXTERNAL_BRIDGE')
+ bridge_ext_ip = S.getValue('TUNNEL_EXTERNAL_BRIDGE_IP')
+ tgen_ip1 = S.getValue('TRAFFICGEN_PORT1_IP')
self._vswitch.add_switch(bridge)
tasks.run_task(['sudo', 'ip', 'addr', 'add',
- settings.getValue('TUNNEL_INT_BRIDGE_IP'), 'dev', bridge],
+ S.getValue('TUNNEL_INT_BRIDGE_IP'), 'dev', bridge],
self._logger, 'Assign ' +
- settings.getValue('TUNNEL_INT_BRIDGE_IP') + ' to ' + bridge, False)
+ S.getValue('TUNNEL_INT_BRIDGE_IP') + ' to ' + bridge, False)
tasks.run_task(['sudo', 'ip', 'link', 'set', 'dev', bridge, 'up'],
self._logger, 'Bring up ' + bridge, False)
(_, phy2_number) = self._vswitch.add_phy_port(bridge)
if tunnel_type == "vxlan":
- vxlan_vni = 'options:key=' + settings.getValue('VXLAN_VNI')
+ vxlan_vni = 'options:key=' + S.getValue('VXLAN_VNI')
self._vswitch.add_tunnel_port(bridge, tgen_ip1, tunnel_type,
params=[vxlan_vni])
else:
self._logger,
'Set ' + bridge_ext + ' status to up')
- tg_port2_mac = settings.getValue('TRAFFICGEN_PORT2_MAC')
- vtep_ip2 = settings.getValue('TRAFFICGEN_PORT2_IP')
+ tg_port2_mac = S.getValue('TRAFFICGEN_PORT2_MAC')
+ vtep_ip2 = S.getValue('TRAFFICGEN_PORT2_IP')
self._vswitch.set_tunnel_arp(vtep_ip2,
tg_port2_mac,
bridge_ext)
self._vswitch.add_route(bridge,
- settings.getValue('VTEP_IP2_SUBNET'),
+ S.getValue('VTEP_IP2_SUBNET'),
bridge)
def get_ports_info(self):
"""See IVswitchController for description
"""
- self._logger.debug('get_ports_info using ' + str(self._vswitch_class))
- return self._vswitch.get_ports(settings.getValue('VSWITCH_BRIDGE_NAME'))
+ self._logger.debug('get_ports_info for bridges: %s, %s',
+ S.getValue('TUNNEL_INTEGRATION_BRIDGE'),
+ S.getValue('TUNNEL_EXTERNAL_BRIDGE'))
+ return self._vswitch.get_ports(
+ S.getValue('TUNNEL_INTEGRATION_BRIDGE')) +\
+ self._vswitch.get_ports(
+ S.getValue('TUNNEL_EXTERNAL_BRIDGE'))
def dump_vswitch_flows(self):
"""See IVswitchController for description
"""
- self._vswitch.dump_flows(settings.getValue('VSWITCH_BRIDGE_NAME'))
+ self._vswitch.dump_flows(S.getValue('TUNNEL_INTEGRATION_BRIDGE'))
+ self._vswitch.dump_flows(S.getValue('TUNNEL_EXTERNAL_BRIDGE'))
in case that condition is not ``True``
* ``Eval expression`` - evaluates given expression as a python code and returns
its result
- * ``Exec command [regex]`` - executes a shell command and filters its output by
+ * ``Exec_Shell command [regex]`` - executes a shell command and filters its output by
(optional) regular expression
+ * ``Exec_Python code`` - executes a python code
+
Examples:
.. code-block:: python
- ['tools', 'exec', 'numactl -H', 'available: ([0-9]+)']
+ ['tools', 'exec_shell', 'numactl -H', 'available: ([0-9]+)']
['tools', 'assert', '#STEP[-1][0]>1']
* ``wait`` - is used for test case interruption. This object doesn't have
['wait']
+ * ``sleep`` - is used to pause testcase execution for defined number of seconds.
+
+ Examples:
+
+ .. code-block:: python
+
+ ['sleep', '60']
+
Test Macros
-----------
.. _overriding-parameters-documentation:
+Referencing parameter values
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+It is possible to use a special macro ``#PARAM()`` to refer to the value of
+another configuration parameter. This reference is evaluated during
+access of the parameter value (by ``settings.getValue()`` call), so it
+can refer to parameters created during VSPERF runtime, e.g. NICS dictionary.
+It can be used to reflect DUT HW details in the testcase definition.
+
+Example:
+
+.. code:: python
+
+ {
+ ...
+ "Name": "testcase",
+ "Parameters" : {
+ "TRAFFIC" : {
+ 'l2': {
+ # set destination MAC to the MAC of the first
+ # interface from WHITELIST_NICS list
+ 'dstmac' : '#PARAM(NICS[0]["mac"])',
+ },
+ },
+ ...
+
Overriding values defined in configuration files
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
import logging
import string
import re
+import netaddr
from tools import tasks
from conf import settings
_CACHE_FILE_NAME = '/tmp/vsperf_flows_cache'
+# only simple regex is used; validity of IPv4 is not checked by regex
+_IPV4_REGEX = r"([0-9]{1,3}(\.[0-9]{1,3}){3}(\/[0-9]{1,2})?)"
+
class OFBase(object):
"""Add/remove/show datapaths using ``ovs-ofctl``.
"""
flow_src = flow_src.replace('udp_dst', 'tp_dst')
flow_src = flow_src.replace('tcp_src', 'tp_src')
flow_src = flow_src.replace('tcp_dst', 'tp_dst')
+ flow_src = flow_src.replace('0x800', '0x0800')
+
+ # modify IPv4 CIDR to real network addresses
+ for ipv4_cidr in re.findall(_IPV4_REGEX, flow_src):
+ if ipv4_cidr[2]:
+ tmp_cidr = str(netaddr.IPNetwork(ipv4_cidr[0]).cidr)
+ flow_src = flow_src.replace(ipv4_cidr[0], tmp_cidr)
# split flow strings into lists of comparable elements
- flow_dump_list = re.findall(r"[\w.:=()]+", flow_dump)
- flow_src_list = re.findall(r"[\w.:=()]+", flow_src)
+ flow_dump_list = re.findall(r"[\w.:=()/]+", flow_dump)
+ flow_src_list = re.findall(r"[\w.:=()/]+", flow_src)
# check if all items from source flow are present in dump flow
flow_src_ctrl = list(flow_src_list)
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
""" Report test results
"""
if self.test:
- results = OrderedDict()
- results['status'] = 'OK' if self._step_status['status'] else 'FAILED'
- results['details'] = self._step_status['details']
- TestCase.write_result_to_file([results], self._output_file)
+ tmp_results = OrderedDict()
+ tmp_results['status'] = 'OK' if self._step_status['status'] else 'FAILED'
+ tmp_results['details'] = self._step_status['details']
+ self._tc_results = [tmp_results]
+
+ super(IntegrationTestCase, self).run_report()
+
self.step_report_status("Test '{}'".format(self.name), self._step_status['status'])
# inform vsperf about testcase failure
if not self._step_status['status']:
self._traffic = functions.check_traffic(self._traffic)
# Packet Forwarding mode
- self._vswitch_none = S.getValue('VSWITCH').strip().lower() == 'none'
+ self._vswitch_none = str(S.getValue('VSWITCH')).strip().lower() == 'none'
# trafficgen configuration required for tests of tunneling protocols
if self.deployment == "op2p":
self._logger.debug("Traffic Results:")
self._traffic_ctl.print_results()
+ if self._tc_results is None:
self._tc_results = self._append_results(results)
- TestCase.write_result_to_file(self._tc_results, self._output_file)
+ else:
+ # integration step driven tests have their status and possible
+ # failure details stored inside self._tc_results
+ results = self._append_results(results)
+ if len(self._tc_results) < len(results):
+ if len(self._tc_results) > 1:
+ raise RuntimeError('Testcase results do not match:'
+ 'results: {}\n'
+ 'trafficgen results: {}\n',
+ self._tc_results,
+ results)
+ else:
+ tmp_results = copy.deepcopy(self._tc_results[0])
+ self._tc_results = []
+ for res in results:
+ tmp_res = copy.deepcopy(tmp_results)
+ tmp_res.update(res)
+ self._tc_results.append(tmp_res)
+ else:
+ for i, result in enumerate(results):
+ self._tc_results[i].update(result)
+
+ TestCase.write_result_to_file(self._tc_results, self._output_file)
def run(self):
"""Run the test
# hugepages are needed by DPDK and Qemu
if not self._hugepages_mounted and \
(self.deployment.count('v') or \
- S.getValue('VSWITCH').lower().count('dpdk') or \
+ str(S.getValue('VSWITCH')).lower().count('dpdk') or \
self._vswitch_none or \
self.test and 'vnf' in [step[0][0:3] for step in self.test]):
hugepages.mount_hugepages()
# get hugepage amounts for each socket on dpdk
sock0_mem, sock1_mem = 0, 0
- if S.getValue('VSWITCH').lower().count('dpdk'):
+ if str(S.getValue('VSWITCH')).lower().count('dpdk'):
sock_mem = S.getValue('DPDK_SOCKET_MEM')
sock0_mem, sock1_mem = (int(sock_mem[0]) * 1024 / hugepage_size,
int(sock_mem[1]) * 1024 / hugepage_size)
input(os.linesep + "Step {}: Press Enter to continue with "
"the next step...".format(i) + os.linesep + os.linesep)
continue
+ elif step[0] == 'sleep':
+ self._logger.debug("Sleep %s seconds", step[1])
+ time.sleep(int(step[1]))
+ continue
else:
self._logger.error("Unsupported test object %s", step[0])
self._step_status = {'status' : False, 'details' : ' '.join(step)}
self._step_result[i] = test_method(*step_params)
self._logger.debug("Step %s '%s' results '%s'", i,
step_log, self._step_result[i])
- time.sleep(5)
+ time.sleep(S.getValue('TEST_STEP_DELAY'))
if self._step_check:
step_ok = test_method_check(self._step_result[i], *step_params)
except (AssertionError, AttributeError, IndexError) as ex:
'vswitch': _get_version(S.getValue('VSWITCH'), versions),
}
- if S.getValue('VSWITCH').lower().count('dpdk'):
+ if str(S.getValue('VSWITCH')).lower().count('dpdk'):
env.update({'dpdk': _get_version('dpdk', versions)})
if result[ResultsConstants.DEPLOYMENT].count('v'):
-# Copyright 2016 Intel Corporation.
+# Copyright 2016-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
import subprocess
import locale
+_LOGGER = logging.getLogger(__name__)
+
class TestStepsTools(object):
""" Various tools and functions used by step driven testcases
"""
# Functions use nonstandard names to avoid conflicts with
# standard python keywords.
# pylint: disable=invalid-name
- def __init__(self):
- """ TestStepsTools initialization
- """
- self._logger = logging.getLogger(__name__)
-
- def Assert(self, condition):
+ @staticmethod
+ def Assert(condition):
""" Evaluate given `condition' and raise AssertionError
in case, that evaluation fails
"""
try:
- assert self.Eval(condition)
+ assert TestStepsTools.Eval(condition)
except AssertionError:
- self._logger.error('Condition %s is not True', condition)
+ _LOGGER.error('Condition %s is not True', condition)
raise
return True
return result is not None
@staticmethod
- def Exec(command, regex=None):
+ def Exec_Python(code):
+ """ Execute a python `code' and return True on success
+ """
+ # pylint: disable=exec-used
+ try:
+ exec(code, globals())
+ # pylint: disable=broad-except
+ # pylint: disable=bare-except
+ except:
+ _LOGGER.error('Execution of following code has failed %s', code)
+ return False
+ return True
+
+ @staticmethod
+ def validate_Exec_Python(result, dummy_code):
+ """ Validate result of python `code' execution
+ """
+ return result
+
+ @staticmethod
+ def Exec_Shell(command, regex=None):
""" Execute a shell `command' and return its output filtered
out by optional `regex' expression.
"""
return output
@staticmethod
- def validate_Exec(result, dummy_command, dummy_regex=None):
+ def validate_Exec_Shell(result, dummy_command, dummy_regex=None):
""" validate result of shell `command' execution
"""
return result is not None
-# Copyright 2015-2016 Intel Corporation.
+# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
self._testpmd_fwd_mode = S.getValue('GUEST_TESTPMD_FWD_MODE')[self._number]
# in case of SRIOV we must ensure, that MAC addresses are not swapped
if S.getValue('SRIOV_ENABLED') and self._testpmd_fwd_mode.startswith('mac') and \
- not S.getValue('VNF').endswith('PciPassthrough'):
+ not str(S.getValue('VNF')).endswith('PciPassthrough'):
self._logger.info("SRIOV detected, forwarding mode of testpmd was changed from '%s' to '%s'",
self._testpmd_fwd_mode, 'io')
'installer': installer_name,
'pkg_list': pkg_list,
'db_url': opnfv_url}
- if settings.getValue('VSWITCH').endswith('Vanilla'):
+ if str(settings.getValue('VSWITCH')).endswith('Vanilla'):
int_data['vanilla'] = True
opnfvdashboard.results2opnfv_dashboard(results_path, int_data)
Code Review / vswitchperf.git / commitdiff