Add IxNextgen API for checking protocols status

Added IxNextgen API methods which allows to check status of created
protocols

JIRA: YARDSTICK-1468

Change-Id: Id40b59ac258bcce6501862520bd042d31d926783
Signed-off-by: Oleksandr Naumets <oleksandrx.naumets@intel.com>

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 d4f75ba..0ddab2a 100644 (file)
--- a/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py
+++ b/yardstick/network_services/libs/ixia_libs/ixnet/ixnet_api.py
@@ -47,6 +47,9 @@ IP_VERSION_6_MASK = 64
 TRAFFIC_STATUS_STARTED = 'started'
 TRAFFIC_STATUS_STOPPED = 'stopped'
 
+PROTOCOL_STATUS_UP = 'up'
+PROTOCOL_STATUS_DOWN = ['down', 'notStarted']
+
 SUPPORTED_PROTO = [PROTO_UDP]
 
 
@@ -180,6 +183,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
@@ -188,6 +200,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 66fed81..821da32 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
@@ -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')