Merge "[docs] Rename image tc014_pod_frase{e,}r.png"

[yardstick.git] / yardstick / tests / unit / network_services / libs / ixia_libs / test_ixnet_api.py

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 4e3199a..c80cbbe 100644 (file)
--- 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
@@ -334,6 +334,21 @@ class TestIxNextgen(unittest.TestCase):
 
         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')
+
     @mock.patch.object(IxNetwork, 'IxNet')
     def test_connect(self, mock_ixnet):
         mock_ixnet.return_value = self.ixnet
@@ -421,17 +436,24 @@ class TestIxNextgen(unittest.TestCase):
                                 '-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']
-        self.ixnet_gen._create_flow_groups()
+        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', ['1/protocols'],
-                      '-destinations', ['2/protocols']),
-            mock.call('endp2', '-name', '2', '-sources', ['2/protocols'],
-                      '-destinations', ['1/protocols'])])
+            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):
 
@@ -461,12 +483,31 @@ class TestIxNextgen(unittest.TestCase):
     def test_create_traffic_model(self, mock__setup_config_elements,
                                   mock__create_flow_groups,
                                   mock__create_traffic_item):
-
-        self.ixnet_gen.create_traffic_model()
-        mock__create_traffic_item.assert_called_once()
-        mock__create_flow_groups.assert_called_once()
+        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)
+        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']
+        self.ixnet_gen.create_ipv4_traffic_model(uplink_topologies,
+                                                 downlink_topologies)
+        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)
+
     def test__update_frame_mac(self):
         with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item') as \
                 mock_get_field:
@@ -551,6 +592,26 @@ class TestIxNextgen(unittest.TestCase):
             mock.call(port_statistics, self.ixnet_gen.PORT_STATS_NAME_MAP),
             mock.call(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__update_ipv4_address(self):
         with mock.patch.object(self.ixnet_gen, '_get_field_in_stack_item',
                                return_value='field_desc'):
@@ -617,6 +678,9 @@ class TestIxNextgen(unittest.TestCase):
                 'outer_l3': {
                     'proto': 'unsupported',
                 },
+                'outer_l4': {
+                    'seed': 1
+                }
             },
         }
         with mock.patch.object(self.ixnet_gen,
@@ -672,3 +736,34 @@ class TestIxNextgen(unittest.TestCase):
         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'
+        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'
+        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')