Merge "Add IxNextgen API for checking protocols status"

2 files changed, 52 insertions, 0 deletions

diff --git a/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py b/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py
index 87e9dbf85..017b4b66d 100644
--- a/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py
+++ b/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py
@@ -44,6 +44,9 @@ ETHER_TYPE_802_1ad = '0x88a8'
 TRAFFIC_STATUS_STARTED = 'started'
 TRAFFIC_STATUS_STOPPED = 'stopped'
 
+PROTOCOL_STATUS_UP = 'up'
+PROTOCOL_STATUS_DOWN = ['down', 'notStarted']
+
 SUPPORTED_PROTO = [PROTO_UDP]
 
 
@@ -177,6 +180,15 @@ class IxNextgen(object):  # pragma: no cover
         return self.ixnet.getAttribute(self.ixnet.getRoot() + 'traffic',
                                        '-state')
 
+    def _get_protocol_status(self, proto):
+        """Get protocol status
+
+        :param proto: IxNet protocol str representation, e.g.:
+        '::ixNet::OBJ-/topology:2/deviceGroup:1/ethernet:1/ipv4:L14'
+        :return: (str) protocol status: 'up', 'down' or 'notStarted'
+        """
+        return self.ixnet.getAttribute(proto, '-sessionStatus')[0]
+
     def is_traffic_running(self):
         """Returns true if traffic state == TRAFFIC_STATUS_STARTED"""
         return self._get_traffic_state() == TRAFFIC_STATUS_STARTED
@@ -185,6 +197,28 @@ class IxNextgen(object):  # pragma: no cover
         """Returns true if traffic state == TRAFFIC_STATUS_STOPPED"""
         return self._get_traffic_state() == TRAFFIC_STATUS_STOPPED
 
+    def is_protocols_running(self, protocols):
+        """Returns true if all protocols statuses are PROTOCOL_STATUS_UP
+
+        :param protocols: list of protocols str representations, e.g.:
+        ['::ixNet::OBJ-/topology:2/deviceGroup:1/ethernet:1/ipv4:L14', ...]
+        :return: (bool) True if all protocols status is 'up', False if any
+        protocol status is 'down' or 'notStarted'
+        """
+        return all(self._get_protocol_status(proto) == PROTOCOL_STATUS_UP
+                   for proto in protocols)
+
+    def is_protocols_stopped(self, protocols):
+        """Returns true if all protocols statuses are in PROTOCOL_STATUS_DOWN
+
+        :param protocols: list of protocols str representations, e.g.:
+        ['::ixNet::OBJ-/topology:2/deviceGroup:1/ethernet:1/ipv4:L14', ...]
+        :return: (bool) True if all protocols status is 'down' or 'notStarted',
+        False if any protocol status is 'up'
+        """
+        return all(self._get_protocol_status(proto) in PROTOCOL_STATUS_DOWN
+                   for proto in protocols)
+
     @staticmethod
     def _parse_framesize(framesize):
         """Parse "framesize" config param. to return a list of weighted pairs
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..821da32ed 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
@@ -717,6 +717,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'
+        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')