# 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 os
import six
import unittest
-from tests.unit import STL_MOCKS
-
-STLClient = mock.MagicMock()
-stl_patch = mock.patch.dict("sys.modules", STL_MOCKS)
-stl_patch.start()
+from yardstick.network_services.vnf_generic.vnf import tg_rfc2544_ixia
+from yardstick.network_services.traffic_profile import base as tp_base
-if stl_patch:
- from yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia import IxiaTrafficGen
- from yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia import IxiaRfc2544Helper
- from yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia import IxiaResourceHelper
- from yardstick.network_services.traffic_profile.base import TrafficProfile
TEST_FILE_YAML = 'nsb_test_case.yaml'
NAME = "tg__1"
-@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.IxNextgen")
class TestIxiaResourceHelper(unittest.TestCase):
- def test___init___with_custom_rfc_helper(self, *args):
- class MyRfcHelper(IxiaRfc2544Helper):
+
+ def setUp(self):
+ self._mock_IxNextgen = mock.patch.object(tg_rfc2544_ixia, 'IxNextgen')
+ self.mock_IxNextgen = self._mock_IxNextgen.start()
+ self.addCleanup(self._stop_mocks)
+
+ def _stop_mocks(self):
+ self._mock_IxNextgen.stop()
+
+ def test___init___with_custom_rfc_helper(self):
+ class MyRfcHelper(tg_rfc2544_ixia.IxiaRfc2544Helper):
pass
- ixia_resource_helper = IxiaResourceHelper(mock.Mock(), MyRfcHelper)
+ ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(
+ mock.Mock(), MyRfcHelper)
self.assertIsInstance(ixia_resource_helper.rfc_helper, MyRfcHelper)
- def test_stop_collect_with_client(self, *args):
+ def test_stop_collect_with_client(self):
mock_client = mock.Mock()
- ixia_resource_helper = IxiaResourceHelper(mock.Mock())
+ ixia_resource_helper = tg_rfc2544_ixia.IxiaResourceHelper(mock.Mock())
ixia_resource_helper.client = mock_client
ixia_resource_helper.stop_collect()
self.assertEqual(mock_client.ix_stop_traffic.call_count, 1)
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test__initialise_client(self):
+ pass
+
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_run_traffic(self):
+ pass
+
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.IxNextgen")
class TestIXIATrafficGen(unittest.TestCase):
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
# NOTE(ralonsoh): check the object returned.
- IxiaTrafficGen(NAME, vnfd)
+ tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
def test_listen_traffic(self, *args):
with mock.patch("yardstick.ssh.SSH") as ssh:
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
self.assertIsNone(ixnet_traffic_gen.listen_traffic({}))
def test_instantiate(self, *args):
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
scenario_cfg = {'tc': "nsb_test_case", "topology": "",
'ixia_profile': "ixload.cfg"}
scenario_cfg.update({'options': {'packetsize': 64,
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- ixnet_traffic_gen = IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
ixnet_traffic_gen.data = {}
restult = ixnet_traffic_gen.collect_kpi()
self.assertEqual({}, restult)
ssh_mock.execute = \
mock.Mock(return_value=(0, "", ""))
ssh.from_node.return_value = ssh_mock
- ixnet_traffic_gen = IxiaTrafficGen(NAME, vnfd)
+ ixnet_traffic_gen = tg_rfc2544_ixia.IxiaTrafficGen(NAME, vnfd)
ixnet_traffic_gen._terminated = mock.MagicMock()
ixnet_traffic_gen._terminated.value = 0
ixnet_traffic_gen._ixia_traffic_gen = mock.MagicMock()
def test__check_status(self, *args):
vnfd = self.VNFD['vnfd:vnfd-catalog']['vnfd'][0]
- sut = IxiaTrafficGen('vnf1', vnfd)
+ sut = tg_rfc2544_ixia.IxiaTrafficGen('vnf1', vnfd)
sut._check_status()
@mock.patch("yardstick.network_services.vnf_generic.vnf.tg_rfc2544_ixia.time")
@mock.patch("yardstick.ssh.SSH")
def test_traffic_runner(self, mock_ssh, *args):
- mock_traffic_profile = mock.Mock(autospec=TrafficProfile)
+ mock_traffic_profile = mock.Mock(autospec=tp_base.TrafficProfile)
mock_traffic_profile.get_traffic_definition.return_value = "64"
mock_traffic_profile.params = self.TRAFFIC_PROFILE
# traffic_profile.ports is standardized on port_num
mock_traffic_profile.execute_traffic.return_value = ['Completed', samples]
mock_traffic_profile.get_drop_percentage.return_value = ['Completed', samples]
- sut = IxiaTrafficGen(name, vnfd)
+ sut = tg_rfc2544_ixia.IxiaTrafficGen(name, vnfd)
sut.vnf_port_pairs = [[[0], [1]]]
sut.tc_file_name = self._get_file_abspath(TEST_FILE_YAML)
sut.topology = ""
IP_VERSION_4 = 4
IP_VERSION_6 = 6
+PROTO_IPV4 = 'ipv4'
+PROTO_IPV6 = 'ipv6'
+PROTO_UDP = 'udp'
+PROTO_TCP = 'tcp'
+PROTO_VLAN = 'vlan'
class TrafficStreamHelper(object):
class IxNextgen(object):
- STATS_NAME_MAP = {
- "traffic_item": 'Traffic Item',
- "Tx_Frames": 'Tx Frames',
- "Rx_Frames": 'Rx Frames',
- "Tx_Frame_Rate": 'Tx Frame Rate',
- "Rx_Frame_Rate": 'Tx Frame Rate',
- "Store-Forward_Avg_latency_ns": 'Store-Forward Avg Latency (ns)',
- "Store-Forward_Min_latency_ns": 'Store-Forward Min Latency (ns)',
- "Store-Forward_Max_latency_ns": 'Store-Forward Max Latency (ns)',
- }
-
PORT_STATS_NAME_MAP = {
"stat_name": 'Stat Name',
"Frames_Tx": 'Frames Tx.',
MODE_SEEDS_DEFAULT = 'downlink', ['2048', '256']
- @staticmethod
- def find_view_obj(view_name, views):
- edited_view_name = '::ixNet::OBJ-/statistics/view:"{}"'.format(view_name)
- return next((view for view in views if edited_view_name == view), '')
-
@staticmethod
def get_config(tg_cfg):
card = []
self.ixnet = ixnet
self._objRefs = dict()
self._cfg = None
- self._logger = logging.getLogger(__name__)
self._params = None
self._bidir = None
self.ixnet.commit()
- def _connect(self, tg_cfg):
+ def connect(self, tg_cfg):
self._cfg = self.get_config(tg_cfg)
self.ixnet = IxNetwork.IxNet()
machine = self._cfg['machine']
port = str(self._cfg['port'])
version = str(self._cfg['version'])
- result = self.ixnet.connect(machine, '-port', port, '-version', version)
- return result
+ return self.ixnet.connect(machine, '-port', port,
+ '-version', version)
- def clear_ixia_config(self):
+ def clear_config(self):
+ """Wipe out any possible configuration present in the client"""
self.ixnet.execute('newConfig')
- def load_ixia_profile(self, profile):
- self.ixnet.execute('loadConfig', self.ixnet.readFrom(profile))
-
- def ix_load_config(self, profile):
- self.clear_ixia_config()
- self.load_ixia_profile(profile)
-
- def ix_assign_ports(self):
- vports = self.ixnet.getList(self.ixnet.getRoot(), 'vport')
- ports = []
-
- chassis = self._cfg['chassis']
- ports = [(chassis, card, port) for card, port in
+ def assign_ports(self):
+ """Create and assign vports for each physical port defined in config
+
+ This configuration is present in the IXIA profile file. E.g.:
+ name: trafficgen_1
+ role: IxNet
+ interfaces:
+ xe0:
+ vpci: "2:15" # Card:port
+ driver: "none"
+ dpdk_port_num: 0
+ local_ip: "152.16.100.20"
+ netmask: "255.255.0.0"
+ local_mac: "00:98:10:64:14:00"
+ xe1:
+ ...
+ """
+ chassis_ip = self._cfg['chassis']
+ ports = [(chassis_ip, card, port) for card, port in
zip(self._cfg['cards'], self._cfg['ports'])]
- vport_list = self.ixnet.getList("/", "vport")
- self.ixnet.execute('assignPorts', ports, [], vport_list, True)
+ log.info('Create and assign vports: %s', ports)
+ for port in ports:
+ vport = self.ixnet.add(self.ixnet.getRoot(), 'vport')
+ self.ixnet.commit()
+ self.ixnet.execute('assignPorts', [port], [], [vport], True)
+ self.ixnet.commit()
+ if self.ixnet.getAttribute(vport, '-state') != 'up':
+ log.warning('Port %s is down', vport)
+
+ def _create_traffic_item(self):
+ """Create the traffic item to hold the flow groups
+
+ The traffic item tracking by "Traffic Item" is enabled to retrieve the
+ latency statistics.
+ """
+ log.info('Create the traffic item "RFC2544"')
+ traffic_item = self.ixnet.add(self.ixnet.getRoot() + '/traffic',
+ 'trafficItem')
+ self.ixnet.setMultiAttribute(traffic_item, '-name', 'RFC2544',
+ '-trafficType', 'raw')
self.ixnet.commit()
- for vport in vports:
- if self.ixnet.getAttribute(vport, '-state') != 'up':
- log.error("Both thr ports are down...")
+ traffic_item_id = self.ixnet.remapIds(traffic_item)[0]
+ self.ixnet.setAttribute(traffic_item_id + '/tracking',
+ '-trackBy', 'trafficGroupId0')
+ self.ixnet.commit()
+
+ def _create_flow_groups(self):
+ """Create the flow groups between the assigned ports"""
+ traffic_item_id = self.ixnet.getList(self.ixnet.getRoot() + 'traffic',
+ 'trafficItem')[0]
+ log.info('Create the flow groups')
+ vports = self.ixnet.getList(self.ixnet.getRoot(), 'vport')
+ uplink_ports = vports[::2]
+ downlink_ports = vports[1::2]
+ for up, down in zip(uplink_ports, downlink_ports):
+ log.info('FGs: %s <--> %s', up, down)
+ endpoint_set_1 = self.ixnet.add(traffic_item_id, 'endpointSet')
+ endpoint_set_2 = self.ixnet.add(traffic_item_id, 'endpointSet')
+ self.ixnet.setMultiAttribute(
+ endpoint_set_1,
+ '-sources', [up + ' /protocols'],
+ '-destinations', [down + '/protocols'])
+ self.ixnet.setMultiAttribute(
+ endpoint_set_2,
+ '-sources', [down + ' /protocols'],
+ '-destinations', [up + '/protocols'])
+ self.ixnet.commit()
+
+ def _append_procotol_to_stack(self, protocol_name, previous_element):
+ """Append a new element in the packet definition stack"""
+ protocol = (self.ixnet.getRoot() +
+ '/traffic/protocolTemplate:"{}"'.format(protocol_name))
+ self.ixnet.execute('append', previous_element, protocol)
+
+ def _setup_config_elements(self):
+ """Setup the config elements
+
+ The traffic item is configured to allow individual configurations per
+ config element. The default frame configuration is applied:
+ Ethernet II: added by default
+ IPv4: element to add
+ UDP: element to add
+ Payload: added by default
+ Ethernet II (Trailer): added by default
+ :return:
+ """
+ traffic_item_id = self.ixnet.getList(self.ixnet.getRoot() + 'traffic',
+ 'trafficItem')[0]
+ log.info('Split the frame rate distribution per config element')
+ config_elements = self.ixnet.getList(traffic_item_id, 'configElement')
+ for config_element in config_elements:
+ self.ixnet.setAttribute(config_element + '/frameRateDistribution',
+ '-portDistribution', 'splitRateEvenly')
+ self.ixnet.setAttribute(config_element + '/frameRateDistribution',
+ '-streamDistribution', 'splitRateEvenly')
+ self.ixnet.commit()
+ self._append_procotol_to_stack(
+ PROTO_UDP, config_element + '/stack:"ethernet-1"')
+ self._append_procotol_to_stack(
+ PROTO_IPV4, config_element + '/stack:"ethernet-1"')
+
+ def create_traffic_model(self):
+ """Create a traffic item and the needed flow groups
+
+ Each flow group inside the traffic item (only one is present)
+ represents the traffic between two ports:
+ (uplink) (downlink)
+ FlowGroup1: port1 -> port2
+ FlowGroup2: port1 <- port2
+ FlowGroup3: port3 -> port4
+ FlowGroup4: port3 <- port4
+ """
+ self._create_traffic_item()
+ self._create_flow_groups()
+ self._setup_config_elements()
def ix_update_frame(self, params):
streams = ["configElement"]
self.ixnet.setMultiAttribute(helper.transmissionControl,
'-type', '{0}'.format(param.get('traffic_type',
- 'continuous')),
+ 'fixedDuration')),
'-duration', '{0}'.format(param.get('duration',
"30")))
self.ixnet.execute('stop', '/traffic')
def build_stats_map(self, view_obj, name_map):
- return {kl: self.execute_get_column_values(view_obj, kr) for kl, kr in name_map.items()}
-
- def execute_get_column_values(self, view_obj, name):
- return self.ixnet.execute('getColumnValues', view_obj, name)
-
- def ix_get_statistics(self):
- views = self.ixnet.getList('/statistics', 'view')
- stats = {}
- view_obj = self.find_view_obj("Traffic Item Statistics", views)
- stats = self.build_stats_map(view_obj, self.STATS_NAME_MAP)
-
- view_obj = self.find_view_obj("Port Statistics", views)
- ports_stats = self.build_stats_map(view_obj, self.PORT_STATS_NAME_MAP)
-
- view_obj = self.find_view_obj("Flow Statistics", views)
- stats["latency"] = self.build_stats_map(view_obj, self.LATENCY_NAME_MAP)
-
- return stats, ports_stats
+ return {data_yardstick: self.ixnet.execute(
+ 'getColumnValues', view_obj, data_ixia)
+ for data_yardstick, data_ixia in name_map.items()}
+
+ def get_statistics(self):
+ """Retrieve port and flow statistics
+
+ "Port Statistics" parameters are stored in self.PORT_STATS_NAME_MAP.
+ "Flow Statistics" parameters are stored in self.LATENCY_NAME_MAP.
+
+ :return: dictionary with the statistics; the keys of this dictionary
+ are PORT_STATS_NAME_MAP and LATENCY_NAME_MAP keys.
+ """
+ port_statistics = '::ixNet::OBJ-/statistics/view:"Port Statistics"'
+ flow_statistics = '::ixNet::OBJ-/statistics/view:"Flow Statistics"'
+ stats = self.build_stats_map(port_statistics, self.PORT_STATS_NAME_MAP)
+ stats.update(self.build_stats_map(flow_statistics,
+ self.LATENCY_NAME_MAP))
+ return stats
# 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 IxNetwork
import unittest
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IxNextgen
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IP_VERSION_4
-from yardstick.network_services.libs.ixia_libs.IxNet.IxNet import IP_VERSION_6
+from yardstick.network_services.libs.ixia_libs.IxNet import IxNet
+
-UPLINK = "uplink"
-DOWNLINK = "downlink"
+UPLINK = 'uplink'
+DOWNLINK = 'downlink'
class TestIxNextgen(unittest.TestCase):
+ def setUp(self):
+ self.ixnet = mock.Mock()
+ self.ixnet.execute = mock.Mock()
+
def test___init__(self):
- ixnet_gen = IxNextgen()
+ ixnet_gen = IxNet.IxNextgen()
+ self.assertIsNone(ixnet_gen.ixnet)
+ self.assertTrue(isinstance(ixnet_gen._objRefs, dict))
+ self.assertIsNone(ixnet_gen._cfg)
+ self.assertIsNone(ixnet_gen._params)
self.assertIsNone(ixnet_gen._bidir)
+ def test___init__ixnet(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ self.assertIsNotNone(ixnet_gen.ixnet)
+
@mock.patch.object(IxNetwork, 'IxNet')
def test_connect(self, mock_ixnet):
- ixnet_instance = mock.Mock()
- mock_ixnet.return_value = ixnet_instance
- ixnet_gen = IxNextgen()
+ mock_ixnet.return_value = self.ixnet
+ ixnet_gen = IxNet.IxNextgen()
with mock.patch.object(ixnet_gen, 'get_config') as mock_config:
mock_config.return_value = {'machine': 'machine_fake',
'port': 'port_fake',
'version': 12345}
- ixnet_gen._connect(mock.ANY)
+ ixnet_gen.connect(mock.ANY)
- ixnet_instance.connect.assert_called_once_with(
+ self.ixnet.connect.assert_called_once_with(
'machine_fake', '-port', 'port_fake', '-version', '12345')
mock_config.assert_called_once()
- def test_clear_ixia_config(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
-
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.clear_ixia_config()
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
-
- def test_load_ixia_profile(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
+ def test_connect_invalid_config_no_machine(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.get_config = mock.Mock(return_value={
+ 'port': 'port_fake',
+ 'version': '12345'})
+ self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
+ self.ixnet.connect.assert_not_called()
+
+ def test_connect_invalid_config_no_port(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.get_config = mock.Mock(return_value={
+ 'machine': 'machine_fake',
+ 'version': '12345'})
+ self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
+ self.ixnet.connect.assert_not_called()
+
+ def test_connect_invalid_config_no_version(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.get_config = mock.Mock(return_value={
+ 'machine': 'machine_fake',
+ 'port': 'port_fake'})
+ self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
+ self.ixnet.connect.assert_not_called()
+
+ def test_connect_no_config(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.get_config = mock.Mock(return_value={})
+ self.assertRaises(KeyError, ixnet_gen.connect, mock.ANY)
+ self.ixnet.connect.assert_not_called()
+
+ def test_clear_config(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.clear_config()
+ self.ixnet.execute.assert_called_once_with('newConfig')
+
+ def test_assign_ports_2_ports(self):
+ self.ixnet.getAttribute.side_effect = ['up', 'down']
- ixnet_gen = IxNextgen(ixnet)
-
- result = ixnet_gen.load_ixia_profile({})
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
+ config = {
+ 'chassis': '1.1.1.1',
+ 'cards': ['1', '2'],
+ 'ports': ['2', '2'],
+ }
- def test_load_ixia_config(self):
- ixnet = mock.MagicMock()
- ixnet.execute = mock.Mock()
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen._cfg = config
- ixnet_gen = IxNextgen(ixnet)
+ self.assertIsNone(ixnet_gen.assign_ports())
- result = ixnet_gen.ix_load_config({})
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 2)
+ self.assertEqual(self.ixnet.execute.call_count, 2)
+ self.assertEqual(self.ixnet.commit.call_count, 4)
+ self.assertEqual(self.ixnet.getAttribute.call_count, 2)
- @mock.patch('yardstick.network_services.libs.ixia_libs.IxNet.IxNet.log')
- def test_ix_assign_ports(self, mock_logger):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0, 1]
- ixnet.getAttribute.side_effect = ['up', 'down']
+ @mock.patch.object(IxNet, 'log')
+ def test_assign_ports_port_down(self, mock_log):
+ self.ixnet.getAttribute.return_value = 'down'
config = {
'chassis': '1.1.1.1',
'cards': ['1', '2'],
- 'ports': ['2', '2'],
+ 'ports': ['3', '4'],
}
-
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
ixnet_gen._cfg = config
-
- result = ixnet_gen.ix_assign_ports()
- self.assertIsNone(result)
- self.assertEqual(ixnet.execute.call_count, 1)
- self.assertEqual(ixnet.commit.call_count, 1)
- self.assertEqual(ixnet.getAttribute.call_count, 2)
- self.assertEqual(mock_logger.error.call_count, 1)
+ ixnet_gen.assign_ports()
+ mock_log.warning.assert_called()
+
+ def test_assign_ports_no_config(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen._cfg = {}
+
+ self.assertRaises(KeyError, ixnet_gen.assign_ports)
+
+ def test__create_traffic_item(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ self.ixnet.getRoot.return_value = 'rootyrootroot'
+ self.ixnet.add.return_value = 'my_new_traffic_item'
+ self.ixnet.remapIds.return_value = ['my_traffic_item_id']
+
+ ixnet_gen._create_traffic_item()
+
+ self.ixnet.add.assert_called_once_with(
+ 'rootyrootroot/traffic', 'trafficItem')
+ self.ixnet.setMultiAttribute.assert_called_once_with(
+ 'my_new_traffic_item', '-name', 'RFC2544', '-trafficType', 'raw')
+ self.assertEqual(2, self.ixnet.commit.call_count)
+ self.ixnet.remapIds.assert_called_once_with('my_new_traffic_item')
+ self.ixnet.setAttribute('my_traffic_item_id/tracking',
+ '-trackBy', 'trafficGroupId0')
+
+ def test__create_flow_groups(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ self.ixnet.getRoot.return_value = 'rootyrootroot'
+ ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']]
+ ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2']
+ ixnet_gen._create_flow_groups()
+ ixnet_gen.ixnet.add.assert_has_calls([
+ mock.call('traffic_item', 'endpointSet'),
+ mock.call('traffic_item', 'endpointSet')])
+ ixnet_gen.ixnet.setMultiAttribute.asser_has_calls([
+ mock.call('endp1', '-sources', ['1/protocols'], '-destinations',
+ ['2/protocols']),
+ mock.call('endp2', '-sources', ['2/protocols'], '-destinations',
+ ['1/protocols'])])
+
+ def test__append_protocol_to_stack(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ self.ixnet.getRoot.return_value = 'my_root'
+
+ ixnet_gen._append_procotol_to_stack('my_protocol', 'prev_element')
+ self.ixnet.execute.assert_called_with(
+ 'append', 'prev_element',
+ 'my_root/traffic/protocolTemplate:"my_protocol"')
+
+ def test__setup_config_elements(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+ ixnet_gen.ixnet.getRoot.return_value = 'root_element'
+ ixnet_gen.ixnet.getList.side_effect = [['traffic_item'],
+ ['cfg_element']]
+ with mock.patch.object(ixnet_gen, '_append_procotol_to_stack') as \
+ mock_append_proto:
+ ixnet_gen._setup_config_elements()
+ mock_append_proto.assert_has_calls([
+ mock.call(IxNet.PROTO_UDP, 'cfg_element/stack:"ethernet-1"'),
+ mock.call(IxNet.PROTO_IPV4, 'cfg_element/stack:"ethernet-1"')])
+ ixnet_gen.ixnet.setAttribute.assert_has_calls([
+ mock.call('cfg_element/frameRateDistribution', '-portDistribution',
+ 'splitRateEvenly'),
+ mock.call('cfg_element/frameRateDistribution',
+ '-streamDistribution', 'splitRateEvenly')])
+
+ @mock.patch.object(IxNet.IxNextgen, '_create_traffic_item')
+ @mock.patch.object(IxNet.IxNextgen, '_create_flow_groups')
+ @mock.patch.object(IxNet.IxNextgen, '_setup_config_elements')
+ def test_create_traffic_model(self, mock__setup_config_elements,
+ mock__create_flow_groups,
+ mock__create_traffic_item):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
+
+ ixnet_gen.create_traffic_model()
+ mock__create_traffic_item.assert_called_once()
+ mock__create_flow_groups.assert_called_once()
+ mock__setup_config_elements.assert_called_once()
def test_ix_update_frame(self):
static_traffic_params = {
}
}
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
+ self.ixnet.remapIds.return_value = ["0"]
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [
[1],
[1],
[1],
],
]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_update_frame(static_traffic_params)
self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 7)
- self.assertEqual(ixnet.commit.call_count, 2)
+ self.assertEqual(self.ixnet.setMultiAttribute.call_count, 7)
+ self.assertEqual(self.ixnet.commit.call_count, 2)
- def test_ix_update_udp(self):
- ixnet = mock.MagicMock()
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_update_ether_multi_attribute(self):
+ pass
+
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_update_ether_multi_attributes(self):
+ pass
+
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_update_ether(self):
+ pass
- ixnet_gen = IxNextgen(ixnet)
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_ix_update_udp(self):
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_update_udp({})
self.assertIsNone(result)
+ # NOTE(ralonsoh): to be updated in next patchset
def test_ix_update_tcp(self):
- ixnet = mock.MagicMock()
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_update_tcp({})
self.assertIsNone(result)
+ # NOTE(ralonsoh): to be updated in next patchset
def test_ix_start_traffic(self):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0]
- ixnet.getAttribute.return_value = 'down'
+ self.ixnet.getList.return_value = [0]
+ self.ixnet.getAttribute.return_value = 'down'
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_start_traffic()
self.assertIsNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 3)
+ self.ixnet.getList.assert_called_once()
+ self.assertEqual(self.ixnet.execute.call_count, 3)
+ # NOTE(ralonsoh): to be updated in next patchset
def test_ix_stop_traffic(self):
- ixnet = mock.MagicMock()
- ixnet.getList.return_value = [0]
+ self.ixnet.getList.return_value = [0]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_stop_traffic()
self.assertIsNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 1)
+ self.ixnet.getList.assert_called_once()
+ self.ixnet.execute.assert_called_once()
- def test_ix_get_statistics(self):
- ixnet = mock.MagicMock()
- ixnet.execute.return_value = ""
- ixnet.getList.side_effect = [
- [
- '::ixNet::OBJ-/statistics/view:"Traffic Item Statistics"',
- '::ixNet::OBJ-/statistics/view:"Port Statistics"',
- ],
- [
- '::ixNet::OBJ-/statistics/view:"Flow Statistics"',
- ],
- ]
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_build_stats_map(self):
+ pass
- ixnet_gen = IxNextgen(ixnet)
+ # NOTE(ralonsoh): to be updated in next patchset
+ def test_get_statistics(self):
+ self.ixnet.execute.return_value = ""
- result = ixnet_gen.ix_get_statistics()
- self.assertIsNotNone(result)
- self.assertEqual(ixnet.getList.call_count, 1)
- self.assertEqual(ixnet.execute.call_count, 20)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
- def test_find_view_obj_no_where(self):
- views = ['here', 'there', 'everywhere']
- result = IxNextgen.find_view_obj('no_where', views)
- self.assertEqual(result, '')
+ result = ixnet_gen.get_statistics()
+ self.assertIsNotNone(result)
+ self.assertEqual(self.ixnet.execute.call_count, 12)
+ # NOTE(ralonsoh): to be updated in next patchset
def test_add_ip_header_v4(self):
static_traffic_params = {
"uplink_0": {
}
}
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [[1], [0], [0], ["srcIp", "dstIp"]]
+ self.ixnet.remapIds.return_value = ["0"]
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [[1], [0], [0], ["srcIp", "dstIp"]]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_4)
+ result = ixnet_gen.add_ip_header(static_traffic_params,
+ IxNet.IP_VERSION_4)
self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
+ self.ixnet.setMultiAttribute.assert_called()
+ self.ixnet.commit.assert_called_once()
def test_add_ip_header_v4_nothing_to_do(self):
static_traffic_params = {
}
}
- ixnet = mock.MagicMock()
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [[1], [0, 1], [0], ["srcIp", "dstIp"]]
+ self.ixnet.remapIds.return_value = ["0"]
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [[1], [0, 1], [0], ["srcIp", "dstIp"]]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_4)
+ result = ixnet_gen.add_ip_header(static_traffic_params,
+ IxNet.IP_VERSION_4)
self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
+ self.ixnet.setMultiAttribute.assert_called()
+ self.ixnet.commit.assert_called_once()
def test_add_ip_header_v6(self):
static_traffic_profile = {
}
}
- ixnet = mock.MagicMock()
- ixnet.getList.side_effect = [[1], [1], [1], ["srcIp", "dstIp"]]
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [[1], [1], [1], ["srcIp", "dstIp"]]
+ self.ixnet.remapIds.return_value = ["0"]
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
- result = ixnet_gen.add_ip_header(static_traffic_profile, IP_VERSION_6)
+ result = ixnet_gen.add_ip_header(static_traffic_profile,
+ IxNet.IP_VERSION_6)
self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
- self.assertEqual(ixnet.commit.call_count, 1)
+ self.ixnet.setMultiAttribute.assert_called()
+ self.ixnet.commit.assert_called_once()
def test_add_ip_header_v6_nothing_to_do(self):
static_traffic_params = {
}
}
- ixnet = mock.MagicMock()
- ixnet.getList.side_effect = [[1], [0, 1], [1], ["srcIP", "dstIP"]]
- ixnet.remapIds.return_value = ["0"]
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [[1], [0, 1], [1], ["srcIP", "dstIP"]]
+ self.ixnet.remapIds.return_value = ["0"]
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
- result = ixnet_gen.add_ip_header(static_traffic_params, IP_VERSION_6)
+ result = ixnet_gen.add_ip_header(static_traffic_params,
+ IxNet.IP_VERSION_6)
self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 0)
+ self.ixnet.setMultiAttribute.assert_not_called()
def test_set_random_ip_multi_attributes_bad_ip_version(self):
bad_ip_version = object()
- ixnet_gen = IxNextgen(mock.Mock())
+ ixnet_gen = IxNet.IxNextgen(mock.Mock())
with self.assertRaises(ValueError):
ixnet_gen.set_random_ip_multi_attributes(
mock.Mock(), bad_ip_version, mock.Mock(), mock.Mock())
'bidir': True,
}
- result = IxNextgen.get_config(tg_cfg)
- self.assertDictEqual(result, expected)
+ result = IxNet.IxNextgen.get_config(tg_cfg)
+ self.assertEqual(result, expected)
def test_ix_update_ether(self):
static_traffic_params = {
}
}
- ixnet = mock.MagicMock()
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [
[1],
[1],
[1],
],
]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_update_ether(static_traffic_params)
self.assertIsNone(result)
- self.assertGreater(ixnet.setMultiAttribute.call_count, 0)
+ self.ixnet.setMultiAttribute.assert_called()
def test_ix_update_ether_nothing_to_do(self):
static_traffic_params = {
}
}
- ixnet = mock.MagicMock()
- ixnet.setMultiAttribute.return_value = [1]
- ixnet.commit.return_value = [1]
- ixnet.getList.side_effect = [
+ self.ixnet.setMultiAttribute.return_value = [1]
+ self.ixnet.commit.return_value = [1]
+ self.ixnet.getList.side_effect = [
[1],
[1],
[1],
],
]
- ixnet_gen = IxNextgen(ixnet)
+ ixnet_gen = IxNet.IxNextgen(self.ixnet)
result = ixnet_gen.ix_update_ether(static_traffic_params)
self.assertIsNone(result)
- self.assertEqual(ixnet.setMultiAttribute.call_count, 0)
+ self.ixnet.setMultiAttribute.assert_not_called()
Code Review / yardstick.git / commitdiff