aboutsummaryrefslogtreecommitdiffstats
path: root/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py
diff options
context:
space:
mode:
Diffstat (limited to 'yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py')
-rw-r--r--yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py208
1 files changed, 188 insertions, 20 deletions
diff --git a/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py b/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py
index c1d902061..a20592dc7 100644
--- a/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py
+++ b/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2018 Intel Corporation
+# Copyright (c) 2018-2019 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -15,16 +15,54 @@
import mock
import IxNetwork
import unittest
+import re
from copy import deepcopy
+from collections import OrderedDict
from yardstick.common import exceptions
from yardstick.network_services.libs.ixia_libs.ixnet import ixnet_api
+from yardstick.network_services.traffic_profile import ixia_rfc2544
UPLINK = 'uplink'
DOWNLINK = 'downlink'
+TRAFFIC_PROFILE = {
+ 'uplink_0': {
+ 'ipv4': {
+ 'outer_l2': {
+ 'framesize': {
+ '128B': '0',
+ '1518B': '0',
+ '64B': '0',
+ '373b': '0',
+ '256B': '0',
+ '1400B': '0',
+ '570B': '0'}},
+ 'id': 1}},
+ 'description': 'Traffic profile to run RFC2544 latency',
+ 'name': 'rfc2544',
+ 'schema': 'isb:traffic_profile:0.1',
+ 'traffic_profile': {
+ 'injection_time': None,
+ 'enable_latency': True,
+ 'frame_rate': '100%',
+ 'traffic_type': 'IXIARFC2544Profile'},
+ 'downlink_0': {
+ 'ipv4': {
+ 'outer_l2': {
+ 'framesize': {
+ '128B': '0',
+ '1518B': '0',
+ '64B': '0',
+ '373b': '0',
+ '256B': '0',
+ '1400B': '0',
+ '570B': '0'}},
+ 'id': 2}}}
+
+
TRAFFIC_PARAMETERS = {
UPLINK: {
'id': 1,
@@ -103,6 +141,12 @@ class TestIxNextgen(unittest.TestCase):
self.ixnet.getRoot.return_value = 'my_root'
self.ixnet_gen = ixnet_api.IxNextgen()
self.ixnet_gen._ixnet = self.ixnet
+ self._mock_log = mock.patch.object(ixnet_api.log, 'info')
+ self.mock_log = self._mock_log.start()
+ self.addCleanup(self._stop_mocks)
+
+ def _stop_mocks(self):
+ self.mock_log.stop()
def test_get_config(self):
tg_cfg = {
@@ -373,13 +417,15 @@ class TestIxNextgen(unittest.TestCase):
self.ixnet_gen.add_static_ipv4(iface='iface',
vport='vport',
start_ip='10.0.0.0',
- count='100')
+ count='100',
+ mask='32')
self.ixnet_gen.ixnet.add.assert_called_once_with(
'vport/protocols/static', 'ip')
self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call(
'obj', '-protocolInterface', 'iface',
'-ipStart', '10.0.0.0',
'-count', '100',
+ '-mask', '32',
'-enabled', 'true')
@mock.patch.object(IxNetwork, 'IxNet')
@@ -496,12 +542,17 @@ class TestIxNextgen(unittest.TestCase):
'my_root/traffic/protocolTemplate:"my_protocol"')
def test__setup_config_elements(self):
+ # the config parsed from some_file
+ yaml_data = {'traffic_profile': {}
+ }
+ traffic_profile = ixia_rfc2544.IXIARFC2544Profile(yaml_data)
+ traffic_profile.params = TRAFFIC_PROFILE
self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'],
['cfg_element']]
with mock.patch.object(self.ixnet_gen, '_append_procotol_to_stack') as \
mock_append_proto:
- self.ixnet_gen._setup_config_elements()
- mock_append_proto.assert_has_calls([
+ self.ixnet_gen._setup_config_elements(traffic_profile=traffic_profile)
+ mock_append_proto.assert_has_calls([
mock.call(ixnet_api.PROTO_UDP, 'cfg_element/stack:"ethernet-1"'),
mock.call(ixnet_api.PROTO_IPV4, 'cfg_element/stack:"ethernet-1"')])
self.ixnet_gen.ixnet.setAttribute.assert_has_calls([
@@ -516,11 +567,15 @@ class TestIxNextgen(unittest.TestCase):
def test_create_traffic_model(self, mock__setup_config_elements,
mock__create_flow_groups,
mock__create_traffic_item):
+ # the config parsed from some_file
+ yaml_data = {'traffic_profile': {}}
+ traffic_profile = ixia_rfc2544.IXIARFC2544Profile(yaml_data)
uplink_ports = ['port1', 'port3']
downlink_ports = ['port2', 'port4']
uplink_endpoints = ['port1/protocols', 'port3/protocols']
downlink_endpoints = ['port2/protocols', 'port4/protocols']
- self.ixnet_gen.create_traffic_model(uplink_ports, downlink_ports)
+ self.ixnet_gen.create_traffic_model(uplink_ports, downlink_ports,
+ traffic_profile=traffic_profile)
mock__create_traffic_item.assert_called_once_with('raw')
mock__create_flow_groups.assert_called_once_with(uplink_endpoints,
downlink_endpoints)
@@ -534,12 +589,65 @@ class TestIxNextgen(unittest.TestCase):
mock__create_traffic_item):
uplink_topologies = ['up1', 'up3']
downlink_topologies = ['down2', 'down4']
+ traffic_profile = 'fake_profile'
self.ixnet_gen.create_ipv4_traffic_model(uplink_topologies,
- downlink_topologies)
+ downlink_topologies,
+ traffic_profile)
mock__create_traffic_item.assert_called_once_with('ipv4')
mock__create_flow_groups.assert_called_once_with(uplink_topologies,
downlink_topologies)
- mock__setup_config_elements.assert_called_once_with(False)
+ mock__setup_config_elements.assert_called_once_with(
+ traffic_profile='fake_profile', add_default_proto=False)
+
+ def test_flows_settings(self):
+ cfg = {'uplink_0': {
+ 'ipv4': {
+ 'outer_l2': {
+ 'framesize': {
+ '128B': '0',
+ '1518B': '0',
+ '64B': '0',
+ '373b': '0',
+ '256B': '0',
+ '1400B': '0',
+ '570B': '0'}},
+ 'id': 1}}}
+
+ expected = [
+ {'ipv4': {
+ 'id': 1,
+ 'outer_l2': {
+ 'framesize': {
+ '1518B': '0',
+ '1400B': '0',
+ '128B': '0',
+ '64B': '0',
+ '256B': '0',
+ '373b': '0',
+ '570B': '0'}}}}]
+
+ self.assertEqual(expected, self.ixnet_gen._flows_settings(cfg=cfg))
+
+ def test_is_qinq(self):
+ flow_data = {'ipv4': {
+ 'outer_l2': {},
+ 'id': 1}}
+ self.assertEqual(False, self.ixnet_gen.is_qinq(flow_data=flow_data))
+
+ flow_data = {'ipv4': {
+ 'outer_l2': {
+ 'QinQ': {
+ 'C-VLAN': {
+ 'priority': 0,
+ 'cfi': 0,
+ 'id': 512},
+ 'S-VLAN': {
+ 'priority': 0,
+ 'cfi': 0,
+ 'id': 128}},
+ },
+ 'id': 1}}
+ self.assertEqual(True, self.ixnet_gen.is_qinq(flow_data=flow_data))
def test__update_frame_mac(self):
with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item') as \
@@ -625,9 +733,9 @@ class TestIxNextgen(unittest.TestCase):
mock.call(self.ixnet_gen.FLOW_STATISTICS,
self.ixnet_gen.LATENCY_NAME_MAP)])
- def test__set_flow_tracking(self):
+ def test_set_flow_tracking(self):
self.ixnet_gen._ixnet.getList.return_value = ['traffic_item']
- self.ixnet_gen._set_flow_tracking(track_by=['vlanVlanId0'])
+ self.ixnet_gen.set_flow_tracking(track_by=['vlanVlanId0'])
self.ixnet_gen.ixnet.setAttribute.assert_called_once_with(
'traffic_item/tracking', '-trackBy', ['vlanVlanId0'])
self.assertEqual(self.ixnet.commit.call_count, 1)
@@ -645,17 +753,77 @@ class TestIxNextgen(unittest.TestCase):
'encapsulation', '-offset', 'IPv4 TOS Precedence')
self.assertEqual(self.ixnet.commit.call_count, 2)
- def test_get_pppoe_scenario_statistics(self):
- with mock.patch.object(self.ixnet_gen, '_build_stats_map') as \
- mock_build_stats:
- self.ixnet_gen.get_pppoe_scenario_statistics()
-
- mock_build_stats.assert_any_call(self.ixnet_gen.PORT_STATISTICS,
- self.ixnet_gen.PORT_STATS_NAME_MAP)
- mock_build_stats.assert_any_call(self.ixnet_gen.FLOW_STATISTICS,
- self.ixnet_gen.LATENCY_NAME_MAP)
- mock_build_stats.assert_any_call(self.ixnet_gen.PPPOX_CLIENT_PER_PORT,
- self.ixnet_gen.PPPOX_CLIENT_PER_PORT_NAME_MAP)
+ def test__get_view_page_stats(self):
+ expected_stats = [
+ {'header1': 'row1_1', 'header2': 'row1_2'},
+ {'header1': 'row2_1', 'header2': 'row2_2'}
+ ]
+ self.ixnet_gen._ixnet.getAttribute.side_effect = [
+ ['header1', 'header2'],
+ [
+ [['row1_1', 'row1_2']],
+ [['row2_1', 'row2_2']]
+ ]
+ ]
+ stats = self.ixnet_gen._get_view_page_stats('view_obj')
+ self.assertListEqual(stats, expected_stats)
+
+ @mock.patch.object(ixnet_api.IxNextgen, '_get_view_page_stats')
+ def test_get_pppoe_scenario_statistics(self, mock_get_view):
+
+ pattern = re.compile('Flow 2')
+
+ expected_stats = {
+ 'port_statistics': [{
+ 'port_1': 'port_stat1',
+ 'port_2': 'port_stat2'
+ }],
+ 'flow_statistic': [{
+ 'flow_1': 'flow_stat1',
+ 'flow_2': 'flow_stat2'
+ }],
+ 'pppox_client_per_port': [{
+ 'sub_1': 'sub_stat1',
+ 'sub_2': 'sub_stat2'
+ }]
+ }
+
+ pppoe_scenario_stats = OrderedDict([
+ ('port_statistics', 'view_obj'),
+ ('flow_statistic', 'view_obj'),
+ ('pppox_client_per_port', 'view_obj')
+ ])
+
+ pppoe_scenario_stats_map = {
+ 'port_statistics': {'port_1': 'Port 1',
+ 'port_2': 'Port 2'},
+ 'flow_statistic': {'flow_1': 'Flow 1',
+ 'flow_2': pattern},
+ 'pppox_client_per_port': {'sub_1': 'Sub 1',
+ 'sub_2': 'Sub 2'}
+ }
+
+ # All stats keys
+ port_stats = [{'Port 1': 'port_stat1',
+ 'Port 2': 'port_stat2',
+ 'Port 3': 'port_stat3'}]
+ flows_stats = [{'Flow 1': 'flow_stat1',
+ 'Flow 2': 'flow_stat2',
+ 'Flow 3': 'flow_stat3'}]
+ pppoe_sub_stats = [{'Sub 1': 'sub_stat1',
+ 'Sub 2': 'sub_stat2',
+ 'Sub 3': 'sub_stat3'}]
+
+ mock_get_view.side_effect = [port_stats, flows_stats, pppoe_sub_stats]
+ self.ixnet_gen._ixnet.getAttribute.return_value = '1'
+
+ with mock.patch.multiple(ixnet_api.IxNextgen,
+ PPPOE_SCENARIO_STATS=pppoe_scenario_stats,
+ PPPOE_SCENARIO_STATS_MAP=pppoe_scenario_stats_map):
+ stats = self.ixnet_gen.get_pppoe_scenario_statistics()
+ self.assertDictEqual(stats, expected_stats)
+ self.assertEqual(self.ixnet_gen.ixnet.getAttribute.call_count, 6)
+ self.ixnet_gen.ixnet.setAttribute.assert_not_called()
def test__update_ipv4_address(self):
with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',