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 | 172 |
1 files changed, 165 insertions, 7 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 66fed81f1..c1d902061 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 @@ -46,7 +46,8 @@ TRAFFIC_PARAMETERS = { 'dstip': '152.16.40.20', 'srcip': '152.16.100.20', 'dstmask': 24, - 'srcmask': 24 + 'srcmask': 24, + 'priority': {'raw': '0x01'} }, 'outer_l4': { 'seed': 1, @@ -78,7 +79,8 @@ TRAFFIC_PARAMETERS = { 'dstip': '2001::10', 'srcip': '2021::10', 'dstmask': 64, - 'srcmask': 64 + 'srcmask': 64, + 'priority': {'raw': '0x01'} }, 'outer_l4': { 'seed': 1, @@ -349,6 +351,37 @@ class TestIxNextgen(unittest.TestCase): 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') + 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', + '-enabled', 'true') + @mock.patch.object(IxNetwork, 'IxNet') def test_connect(self, mock_ixnet): mock_ixnet.return_value = self.ixnet @@ -582,15 +615,47 @@ class TestIxNextgen(unittest.TestCase): self.ixnet_gen.update_frame(TRAFFIC_PARAMETERS, 40) def test_get_statistics(self): - port_statistics = '::ixNet::OBJ-/statistics/view:"Port Statistics"' - flow_statistics = '::ixNet::OBJ-/statistics/view:"Flow Statistics"' 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(port_statistics, self.ixnet_gen.PORT_STATS_NAME_MAP), - mock.call(flow_statistics, self.ixnet_gen.LATENCY_NAME_MAP)]) + 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_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__update_ipv4_address(self): with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item', @@ -602,6 +667,78 @@ class TestIxNextgen(unittest.TestCase): '-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'): @@ -622,10 +759,13 @@ class TestIxNextgen(unittest.TestCase): 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'): + '_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, @@ -717,6 +857,24 @@ class TestIxNextgen(unittest.TestCase): 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') |