diff options
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.py | 1057 |
1 files changed, 1057 insertions, 0 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 new file mode 100644 index 000000000..a20592dc7 --- /dev/null +++ b/yardstick/tests/unit/network_services/libs/ixia_libs/test_ixnet_api.py @@ -0,0 +1,1057 @@ +# 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. +# 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 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, + 'bidir': 'False', + 'duration': 60, + 'rate': 10000.5, + 'rate_unit': 'fps', + 'outer_l2': { + 'framesize': {'64B': '25', '256B': '75'}, + 'QinQ': None + }, + 'outer_l3': { + 'count': 512, + 'srcseed': 10, + 'dstseed': 20, + 'dscp': 0, + 'proto': 'udp', + 'ttl': 32, + 'dstip': '152.16.40.20', + 'srcip': '152.16.100.20', + 'dstmask': 24, + 'srcmask': 24, + 'priority': {'raw': '0x01'} + }, + 'outer_l4': { + 'seed': 1, + 'count': 1, + 'dstport': 2001, + 'srcport': 1234, + 'srcportmask': 0, + 'dstportmask': 0 + }, + 'traffic_type': 'continuous' + }, + DOWNLINK: { + 'id': 2, + 'bidir': 'False', + 'duration': 60, + 'rate': 75.2, + 'rate_unit': '%', + 'outer_l2': { + 'framesize': {'128B': '35', '1024B': '65'}, + 'QinQ': None + }, + 'outer_l3': { + 'count': 1024, + 'srcseed': 30, + 'dstseed': 40, + 'dscp': 0, + 'proto': 'udp', + 'ttl': 32, + 'dstip': '2001::10', + 'srcip': '2021::10', + 'dstmask': 64, + 'srcmask': 64, + 'priority': {'raw': '0x01'} + }, + 'outer_l4': { + 'seed': 1, + 'count': 1, + 'dstport': 1234, + 'srcport': 2001, + 'srcportmask': 0, + 'dstportmask': 0 + }, + 'traffic_type': 'continuous' + } +} + + +class TestIxNextgen(unittest.TestCase): + + def setUp(self): + self.ixnet = mock.Mock() + self.ixnet.execute = mock.Mock() + 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 = { + 'vdu': [ + { + 'external-interface': [ + {'virtual-interface': {'vpci': '0000:07:00.1'}}, + {'virtual-interface': {'vpci': '0001:08:01.2'}} + ] + }, + ], + 'mgmt-interface': { + 'ip': 'test1', + 'tg-config': { + 'dut_result_dir': 'test2', + 'version': 'test3', + 'ixchassis': 'test4', + 'tcl_port': 'test5', + }, + } + } + + expected = { + 'machine': 'test1', + 'port': 'test5', + 'chassis': 'test4', + 'cards': ['0000', '0001'], + 'ports': ['07', '08'], + 'output_dir': 'test2', + 'version': 'test3', + 'bidir': True, + } + + result = ixnet_api.IxNextgen.get_config(tg_cfg) + self.assertEqual(result, expected) + + def test__get_config_element_by_flow_group_name(self): + self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'], + ['fg_01']] + self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_01' + output = self.ixnet_gen._get_config_element_by_flow_group_name( + 'flow_group_01') + self.assertEqual('traffic_item/configElement:flow_group_01', output) + + def test__get_config_element_by_flow_group_name_no_match(self): + self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'], + ['fg_01']] + self.ixnet_gen._ixnet.getAttribute.return_value = 'flow_group_02' + output = self.ixnet_gen._get_config_element_by_flow_group_name( + 'flow_group_01') + self.assertIsNone(output) + + def test__get_stack_item(self): + self.ixnet_gen._ixnet.getList.return_value = ['tcp1', 'tcp2', 'udp'] + with mock.patch.object( + self.ixnet_gen, '_get_config_element_by_flow_group_name') as \ + mock_get_cfg_element: + mock_get_cfg_element.return_value = 'cfg_element' + output = self.ixnet_gen._get_stack_item(mock.ANY, ixnet_api.PROTO_TCP) + self.assertEqual(['tcp1', 'tcp2'], output) + + def test__get_stack_item_no_config_element(self): + with mock.patch.object( + self.ixnet_gen, '_get_config_element_by_flow_group_name', + return_value=None): + with self.assertRaises(exceptions.IxNetworkFlowNotPresent): + self.ixnet_gen._get_stack_item(mock.ANY, mock.ANY) + + def test__get_field_in_stack_item(self): + self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2'] + output = self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field2') + self.assertEqual('field2', output) + + def test__get_field_in_stack_item_no_field_present(self): + self.ixnet_gen._ixnet.getList.return_value = ['field1', 'field2'] + with self.assertRaises(exceptions.IxNetworkFieldNotPresentInStackItem): + self.ixnet_gen._get_field_in_stack_item(mock.ANY, 'field3') + + def test__parse_framesize(self): + framesize = {'64B': '75', '512b': '25'} + output = self.ixnet_gen._parse_framesize(framesize) + self.assertEqual(2, len(output)) + self.assertIn([64, 64, 75], output) + self.assertIn([512, 512, 25], output) + + def test_add_topology(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_topology('topology 1', 'vports') + self.ixnet_gen.ixnet.add.assert_called_once_with('my_root', 'topology') + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'obj', '-name', 'topology 1', '-vports', 'vports') + self.ixnet_gen.ixnet.commit.assert_called_once() + + def test_add_device_group(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_device_group('topology', 'device group 1', '1') + self.ixnet_gen.ixnet.add.assert_called_once_with('topology', + 'deviceGroup') + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'obj', '-name', 'device group 1', '-multiplier', '1') + self.ixnet_gen.ixnet.commit.assert_called_once() + + def test_add_ethernet(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_ethernet('device_group', 'ethernet 1') + self.ixnet_gen.ixnet.add.assert_called_once_with('device_group', + 'ethernet') + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'obj', '-name', 'ethernet 1') + self.ixnet_gen.ixnet.commit.assert_called_once() + + def test_add_vlans_single(self): + obj = 'ethernet' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.ixnet.getList.return_value = ['vlan1', 'vlan2'] + vlan1 = ixnet_api.Vlan(vlan_id=100, tp_id='ethertype88a8', prio=2) + vlan2 = ixnet_api.Vlan(vlan_id=101, tp_id='ethertype88a8', prio=3) + self.ixnet_gen.add_vlans(obj, [vlan1, vlan2]) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('ethernet', + '-vlanCount', 2) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue', + '-value', 100) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue', + '-value', 101) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue', + '-value', 2) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('attr/singleValue', + '-value', 3) + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', 'ethertype88a8') + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_vlans_increment(self): + obj = 'ethernet' + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.ixnet.getList.return_value = ['vlan1'] + vlan = ixnet_api.Vlan(vlan_id=100, vlan_id_step=1, prio=3, prio_step=2) + self.ixnet_gen.add_vlans(obj, [vlan]) + self.ixnet.setMultiAttribute.assert_any_call('obj', '-start', 100, + '-step', 1, + '-direction', 'increment') + self.ixnet.setMultiAttribute.assert_any_call('obj', '-start', 3, + '-step', 2, + '-direction', 'increment') + + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_vlans_invalid(self): + vlans = [] + self.assertRaises(RuntimeError, self.ixnet_gen.add_vlans, 'obj', vlans) + + def test_add_ipv4(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_ipv4('ethernet 1', name='ipv4 1') + self.ixnet_gen.ixnet.add.assert_called_once_with('ethernet 1', 'ipv4') + self.ixnet_gen.ixnet.setAttribute.assert_called_once_with('obj', + '-name', + 'ipv4 1') + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_ipv4_single(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.add_ipv4('ethernet 1', name='ipv4 1', addr='100.1.1.100', + prefix='24', gateway='100.1.1.200') + self.ixnet_gen.ixnet.add.assert_called_once_with('ethernet 1', 'ipv4') + self.ixnet_gen.ixnet.setAttribute.assert_called_once_with('obj', + '-name', + 'ipv4 1') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', '100.1.1.100') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', '24') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', '100.1.1.200') + + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_ipv4_counter(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.add_ipv4('ethernet 1', name='ipv4 1', + addr='100.1.1.100', + addr_step='1', + addr_direction='increment', + prefix='24', + gateway='100.1.1.200', + gw_step='1', + gw_direction='increment') + self.ixnet_gen.ixnet.add.assert_any_call('ethernet 1', 'ipv4') + self.ixnet_gen.ixnet.setAttribute.assert_called_once_with('obj', + '-name', + 'ipv4 1') + self.ixnet_gen.ixnet.add.assert_any_call('attr', 'counter') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('obj', '-start', + '100.1.1.100', + '-step', '1', + '-direction', + 'increment') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', '24') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call('obj', '-start', + '100.1.1.200', + '-step', '1', + '-direction', + 'increment') + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_pppox_client(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.add_pppox_client('ethernet 1', 'pap', 'user', 'pwd') + self.ixnet_gen.ixnet.add.assert_called_once_with('ethernet 1', + 'pppoxclient') + + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', 'pap') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', 'user') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'attr/singleValue', '-value', 'pwd') + + self.assertEqual(self.ixnet.commit.call_count, 2) + + def test_add_pppox_client_invalid_auth(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.assertRaises(NotImplementedError, self.ixnet_gen.add_pppox_client, + 'ethernet 1', 'invalid_auth', 'user', 'pwd') + + self.ixnet_gen.ixnet.setMultiAttribute.assert_not_called() + + def test_add_bgp(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.ixnet.getAttribute.return_value = 'attr' + self.ixnet_gen.add_bgp(ipv4='ipv4 1', + dut_ip='10.0.0.1', + local_as=65000, + bgp_type='external') + self.ixnet_gen.ixnet.add.assert_called_once_with('ipv4 1', 'bgpIpv4Peer') + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'attr/singleValue', '-value', '10.0.0.1') + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'attr/singleValue', '-value', 65000) + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'attr/singleValue', '-value', 'external') + + def test_add_interface(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_interface(vport='vport', + ip='10.0.0.2', + mac='00:00:00:00:00:00', + gateway='10.0.0.1') + self.ixnet_gen.ixnet.add.assert_any_call('vport', 'interface') + self.ixnet_gen.ixnet.add.assert_any_call('obj', 'ipv4') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'obj/ethernet', '-macAddress', '00:00:00:00:00:00') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'obj', '-ip', '10.0.0.2') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'obj', '-gateway', '10.0.0.1') + self.ixnet_gen.ixnet.setMultiAttribute.assert_any_call( + 'obj', '-enabled', 'true') + + def test_add_static_ipv4(self): + self.ixnet_gen.ixnet.add.return_value = 'obj' + self.ixnet_gen.add_static_ipv4(iface='iface', + vport='vport', + start_ip='10.0.0.0', + 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') + def test_connect(self, mock_ixnet): + mock_ixnet.return_value = self.ixnet + with mock.patch.object(self.ixnet_gen, 'get_config') as mock_config: + mock_config.return_value = {'machine': 'machine_fake', + 'port': 'port_fake', + 'version': 12345} + self.ixnet_gen.connect(mock.ANY) + + self.ixnet.connect.assert_called_once_with( + 'machine_fake', '-port', 'port_fake', '-version', '12345') + mock_config.assert_called_once() + + def test_connect_invalid_config_no_machine(self): + self.ixnet_gen.get_config = mock.Mock(return_value={ + 'port': 'port_fake', + 'version': '12345'}) + self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY) + self.ixnet.connect.assert_not_called() + + def test_connect_invalid_config_no_port(self): + self.ixnet_gen.get_config = mock.Mock(return_value={ + 'machine': 'machine_fake', + 'version': '12345'}) + self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY) + self.ixnet.connect.assert_not_called() + + def test_connect_invalid_config_no_version(self): + self.ixnet_gen.get_config = mock.Mock(return_value={ + 'machine': 'machine_fake', + 'port': 'port_fake'}) + self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY) + self.ixnet.connect.assert_not_called() + + def test_connect_no_config(self): + self.ixnet_gen.get_config = mock.Mock(return_value={}) + self.assertRaises(KeyError, self.ixnet_gen.connect, mock.ANY) + self.ixnet.connect.assert_not_called() + + def test_clear_config(self): + self.ixnet_gen.clear_config() + self.ixnet.execute.assert_called_once_with('newConfig') + + @mock.patch.object(ixnet_api, 'log') + def test_assign_ports_2_ports(self, *args): + self.ixnet.getAttribute.side_effect = ['up', 'down'] + config = { + 'chassis': '1.1.1.1', + 'cards': ['1', '2'], + 'ports': ['2', '2']} + self.ixnet_gen._cfg = config + + self.assertIsNone(self.ixnet_gen.assign_ports()) + self.assertEqual(self.ixnet.execute.call_count, 1) + self.assertEqual(self.ixnet.commit.call_count, 3) + self.assertEqual(self.ixnet.getAttribute.call_count, 2) + + @mock.patch.object(ixnet_api, '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': ['3', '4']} + self.ixnet_gen._cfg = config + self.ixnet_gen.assign_ports() + mock_log.warning.assert_called() + + def test_assign_ports_no_config(self): + self.ixnet_gen._cfg = {} + self.assertRaises(KeyError, self.ixnet_gen.assign_ports) + + def test__create_traffic_item(self): + self.ixnet.add.return_value = 'my_new_traffic_item' + self.ixnet.remapIds.return_value = ['my_traffic_item_id'] + + self.ixnet_gen._create_traffic_item() + self.ixnet.add.assert_called_once_with( + 'my_root/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): + uplink_endpoints = ['up_endp1', 'up_endp2'] + downlink_endpoints = ['down_endp1', 'down_endp2'] + self.ixnet_gen.ixnet.getList.side_effect = [['traffic_item'], ['1', '2']] + self.ixnet_gen.ixnet.add.side_effect = ['endp1', 'endp2', 'endp3', + 'endp4'] + self.ixnet_gen._create_flow_groups(uplink_endpoints, downlink_endpoints) + self.ixnet_gen.ixnet.add.assert_has_calls([ + mock.call('traffic_item', 'endpointSet'), + mock.call('traffic_item', 'endpointSet')]) + self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([ + mock.call('endp1', '-name', '1', '-sources', ['up_endp1'], + '-destinations', ['down_endp1']), + mock.call('endp2', '-name', '2', '-sources', ['down_endp1'], + '-destinations', ['up_endp1']), + mock.call('endp3', '-name', '3', '-sources', ['up_endp2'], + '-destinations', ['down_endp2']), + mock.call('endp4', '-name', '4', '-sources', ['down_endp2'], + '-destinations', ['up_endp2'])]) + + def test__append_protocol_to_stack(self): + + self.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): + # 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(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([ + mock.call('cfg_element/frameRateDistribution', '-portDistribution', + 'splitRateEvenly'), + mock.call('cfg_element/frameRateDistribution', + '-streamDistribution', 'splitRateEvenly')]) + + @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item') + @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups') + @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements') + 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, + 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) + mock__setup_config_elements.assert_called_once() + + @mock.patch.object(ixnet_api.IxNextgen, '_create_traffic_item') + @mock.patch.object(ixnet_api.IxNextgen, '_create_flow_groups') + @mock.patch.object(ixnet_api.IxNextgen, '_setup_config_elements') + def test_create_ipv4_traffic_model(self, mock__setup_config_elements, + mock__create_flow_groups, + 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, + 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( + 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 \ + mock_get_field: + mock_get_field.return_value = 'field_descriptor' + self.ixnet_gen._update_frame_mac('ethernet_descriptor', 'field', 'mac') + mock_get_field.assert_called_once_with('ethernet_descriptor', 'field') + self.ixnet_gen.ixnet.setMultiAttribute( + 'field_descriptor', '-singleValue', 'mac', '-fieldValue', 'mac', + '-valueType', 'singleValue') + self.ixnet_gen.ixnet.commit.assert_called_once() + + def test_update_frame(self): + with mock.patch.object( + self.ixnet_gen, '_get_config_element_by_flow_group_name', + return_value='cfg_element'), \ + mock.patch.object(self.ixnet_gen, '_update_frame_mac') as \ + mock_update_frame, \ + mock.patch.object(self.ixnet_gen, '_get_stack_item') as \ + mock_get_stack_item: + mock_get_stack_item.side_effect = [['item1'], ['item2'], + ['item3'], ['item4']] + self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 50) + + self.assertEqual(6, len(self.ixnet_gen.ixnet.setMultiAttribute.mock_calls)) + self.assertEqual(4, len(mock_update_frame.mock_calls)) + + self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([ + mock.call('cfg_element/transmissionControl', + '-type', 'continuous', '-duration', 50) + ]) + + def test_update_frame_qinq(self): + with mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', + return_value='cfg_element'), \ + mock.patch.object(self.ixnet_gen, '_update_frame_mac'),\ + mock.patch.object(self.ixnet_gen, '_get_stack_item', + return_value='item'), \ + mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field'): + + traffic_parameters = deepcopy(TRAFFIC_PARAMETERS) + traffic_parameters[UPLINK]['outer_l2']['QinQ'] = { + 'S-VLAN': {'id': 128, + 'priority': 1, + 'cfi': 0}, + 'C-VLAN': {'id': 512, + 'priority': 0, + 'cfi': 2} + } + + self.ixnet_gen.update_frame(traffic_parameters, 50) + + self.ixnet_gen.ixnet.setMultiAttribute.assert_has_calls([ + mock.call('field', '-auto', 'false', '-singleValue', '0x88a8', + '-fieldValue', '0x88a8', '-valueType', 'singleValue'), + mock.call('field', '-auto', 'false', '-singleValue', 1, + '-fieldValue', 1, '-valueType', 'singleValue'), + mock.call('field', '-auto', 'false', '-singleValue', 128, + '-fieldValue', 128, '-valueType', 'singleValue'), + mock.call('field', '-auto', 'false', '-singleValue', 512, + '-fieldValue', 512, '-valueType', 'singleValue'), + mock.call('field', '-auto', 'false', '-singleValue', 2, + '-fieldValue', 2, '-valueType', 'singleValue') + ], any_order=True) + + def test_update_frame_flow_not_present(self): + with mock.patch.object( + self.ixnet_gen, '_get_config_element_by_flow_group_name', + return_value=None): + with self.assertRaises(exceptions.IxNetworkFlowNotPresent): + self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 40) + + def test_get_statistics(self): + with mock.patch.object(self.ixnet_gen, '_build_stats_map') as \ + mock_build_stats: + self.ixnet_gen.get_statistics() + + mock_build_stats.assert_has_calls([ + mock.call(self.ixnet_gen.PORT_STATISTICS, + self.ixnet_gen.PORT_STATS_NAME_MAP), + mock.call(self.ixnet_gen.FLOW_STATISTICS, + self.ixnet_gen.LATENCY_NAME_MAP)]) + + 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.ixnet.setAttribute.assert_called_once_with( + 'traffic_item/tracking', '-trackBy', ['vlanVlanId0']) + self.assertEqual(self.ixnet.commit.call_count, 1) + + def test__set_egress_flow_tracking(self): + self.ixnet_gen._ixnet.getList.side_effect = [['traffic_item'], + ['encapsulation']] + self.ixnet_gen._set_egress_flow_tracking(encapsulation='Ethernet', + offset='IPv4 TOS Precedence') + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'traffic_item', '-egressEnabled', True) + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'encapsulation', '-encapsulation', 'Ethernet') + self.ixnet_gen.ixnet.setAttribute.assert_any_call( + 'encapsulation', '-offset', 'IPv4 TOS Precedence') + self.assertEqual(self.ixnet.commit.call_count, 2) + + 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', + return_value='field_desc'): + self.ixnet_gen._update_ipv4_address(mock.ANY, mock.ANY, '192.168.1.1', + 100, 26, 25) + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'field_desc', '-seed', 100, '-fixedBits', '192.168.1.1', + '-randomMask', '0.0.0.63', '-valueType', 'random', + '-countValue', 25) + + def test__update_ipv4_priority_raw(self): + priority = {'raw': '0x01'} + self.ixnet_gen._set_priority_field = mock.Mock() + with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field_desc'): + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + + self.ixnet_gen._set_priority_field.assert_called_once_with( + 'field_desc', priority['raw']) + + def test__update_ipv4_priority_dscp(self): + priority = {'dscp': {'defaultPHB': [0, 1, 2, 3]}} + self.ixnet_gen._set_priority_field = mock.Mock() + with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field_desc'): + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + + self.ixnet_gen._set_priority_field.assert_called_once_with( + 'field_desc', priority['dscp']['defaultPHB']) + + def test__update_ipv4_priority_tos(self): + priority = {'tos': {'precedence': [0, 4, 7]}} + self.ixnet_gen._set_priority_field = mock.Mock() + with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field_desc'): + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + + self.ixnet_gen._set_priority_field.assert_called_once_with( + 'field_desc', priority['tos']['precedence']) + + def test__update_ipv4_priority_wrong_priority_type(self): + priority = {'test': [0, 4, 7]} + self.ixnet_gen._set_priority_field = mock.Mock() + with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field_desc'): + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + + self.ixnet_gen._set_priority_field.assert_not_called() + + def test__update_ipv4_priority_not_supported_dscp_class(self): + priority = {'dscp': {'testPHB': [0, 4, 7]}} + self.ixnet_gen._set_priority_field = mock.Mock() + self.ixnet_gen._get_field_in_stack_item = mock.Mock() + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + self.ixnet_gen._set_priority_field.assert_not_called() + self.ixnet_gen._get_field_in_stack_item.assert_not_called() + + def test__update_ipv4_priority_not_supported_tos_field(self): + priority = {'tos': {'test': [0, 4, 7]}} + self.ixnet_gen._set_priority_field = mock.Mock() + self.ixnet_gen._get_field_in_stack_item = mock.Mock() + self.ixnet_gen._update_ipv4_priority('field_desc', priority) + self.ixnet_gen._set_priority_field.assert_not_called() + self.ixnet_gen._get_field_in_stack_item.assert_not_called() + + def test__set_priority_field_list_value(self): + value = [1, 4, 7] + self.ixnet_gen._set_priority_field('field_desc', value) + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'field_desc', + '-valueList', [1, 4, 7], + '-activeFieldChoice', 'true', + '-valueType', 'valueList') + + def test__set_priority_field_single_value(self): + value = 7 + self.ixnet_gen._set_priority_field('field_desc', value) + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'field_desc', + '-activeFieldChoice', 'true', + '-singleValue', '7') + + def test__update_udp_port(self): + with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', + return_value='field_desc'): + self.ixnet_gen._update_udp_port(mock.ANY, mock.ANY, 1234, + 2, 0, 2) + + self.ixnet_gen.ixnet.setMultiAttribute.assert_called_once_with( + 'field_desc', + '-auto', 'false', + '-seed', 1, + '-fixedBits', 1234, + '-randomMask', 0, + '-valueType', 'random', + '-countValue', 1) + + def test_update_ip_packet(self): + with mock.patch.object(self.ixnet_gen, '_update_ipv4_address') as \ + mock_update_add, \ + mock.patch.object(self.ixnet_gen, '_get_stack_item'), \ + mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', return_value='celm'), \ + mock.patch.object(self.ixnet_gen, '_update_ipv4_priority') as \ + mock_update_priority: + self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS) + + self.assertEqual(4, len(mock_update_add.mock_calls)) + self.assertEqual(2, len(mock_update_priority.mock_calls)) + + def test_update_ip_packet_exception_no_config_element(self): + with mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', + return_value=None): + with self.assertRaises(exceptions.IxNetworkFlowNotPresent): + self.ixnet_gen.update_ip_packet(TRAFFIC_PARAMETERS) + + def test_update_l4(self): + with mock.patch.object(self.ixnet_gen, '_update_udp_port') as \ + mock_update_udp, \ + mock.patch.object(self.ixnet_gen, '_get_stack_item'), \ + mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', return_value='celm'): + self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS) + + self.assertEqual(4, len(mock_update_udp.mock_calls)) + + def test_update_l4_exception_no_config_element(self): + with mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', + return_value=None): + with self.assertRaises(exceptions.IxNetworkFlowNotPresent): + self.ixnet_gen.update_l4(TRAFFIC_PARAMETERS) + + def test_update_l4_exception_no_supported_proto(self): + traffic_parameters = { + UPLINK: { + 'id': 1, + 'outer_l3': { + 'proto': 'unsupported', + }, + 'outer_l4': { + 'seed': 1 + } + }, + } + with mock.patch.object(self.ixnet_gen, + '_get_config_element_by_flow_group_name', + return_value='celm'): + with self.assertRaises(exceptions.IXIAUnsupportedProtocol): + self.ixnet_gen.update_l4(traffic_parameters) + + @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state') + def test_start_traffic(self, mock_ixnextgen_get_traffic_state): + self.ixnet_gen._ixnet.getList.return_value = [0] + + mock_ixnextgen_get_traffic_state.side_effect = [ + 'stopped', 'started', 'started', 'started'] + + result = self.ixnet_gen.start_traffic() + self.assertIsNone(result) + self.ixnet.getList.assert_called_once() + self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count) + + @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state') + def test_start_traffic_traffic_running( + self, mock_ixnextgen_get_traffic_state): + self.ixnet_gen._ixnet.getList.return_value = [0] + mock_ixnextgen_get_traffic_state.side_effect = [ + 'started', 'stopped', 'started'] + + result = self.ixnet_gen.start_traffic() + self.assertIsNone(result) + self.ixnet.getList.assert_called_once() + self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count) + + @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state') + def test_start_traffic_wait_for_traffic_to_stop( + self, mock_ixnextgen_get_traffic_state): + self.ixnet_gen._ixnet.getList.return_value = [0] + mock_ixnextgen_get_traffic_state.side_effect = [ + 'started', 'started', 'started', 'stopped', 'started'] + + result = self.ixnet_gen.start_traffic() + self.assertIsNone(result) + self.ixnet.getList.assert_called_once() + self.assertEqual(4, self.ixnet_gen._ixnet.execute.call_count) + + @mock.patch.object(ixnet_api.IxNextgen, '_get_traffic_state') + def test_start_traffic_wait_for_traffic_start( + self, mock_ixnextgen_get_traffic_state): + self.ixnet_gen._ixnet.getList.return_value = [0] + mock_ixnextgen_get_traffic_state.side_effect = [ + 'stopped', 'stopped', 'stopped', 'started'] + + result = self.ixnet_gen.start_traffic() + self.assertIsNone(result) + self.ixnet.getList.assert_called_once() + self.assertEqual(3, self.ixnet_gen._ixnet.execute.call_count) + + def test__get_protocol_status(self): + self.ixnet.getAttribute.return_value = ['up'] + self.ixnet_gen._get_protocol_status('ipv4') + self.ixnet.getAttribute.assert_called_once_with('ipv4', + '-sessionStatus') + + @mock.patch.object(ixnet_api.IxNextgen, '_get_protocol_status') + def test_is_protocols_running(self, mock_ixnextgen_get_protocol_status): + mock_ixnextgen_get_protocol_status.return_value = ['up', 'up'] + result = self.ixnet_gen.is_protocols_running(['ethernet', 'ipv4']) + self.assertTrue(result) + + @mock.patch.object(ixnet_api.IxNextgen, '_get_protocol_status') + def test_is_protocols_stopped(self, mock_ixnextgen_get_protocol_status): + mock_ixnextgen_get_protocol_status.return_value = ['down', 'down'] + result = self.ixnet_gen.is_protocols_running(['ethernet', 'ipv4']) + self.assertFalse(result) + + def test_start_protocols(self): + self.ixnet_gen.start_protocols() + self.ixnet.execute.assert_called_once_with('startAllProtocols') + + def test_stop_protocols(self): + self.ixnet_gen.stop_protocols() + self.ixnet.execute.assert_called_once_with('stopAllProtocols') + + def test_get_vports(self): + self.ixnet_gen._ixnet.getRoot.return_value = 'root' + self.ixnet_gen.get_vports() + self.ixnet.getList.assert_called_once_with('root', 'vport') |