Added PSU support

src/sonic_ax_impl/main.py

@@ -33,6 +33,7 @@ class SonicMIB(
     dell.force10.SSeriesMIB,
     cisco.mgmt.CiscoSystemExtMIB,
     cisco.bgp4.CiscoBgp4MIB,
+    cisco.ciscoEntityFruControlMIB.cefcFruPowerStatusTable,
 ):
     """
     If SONiC was to create custom MIBEntries, they may be specified here.

src/sonic_ax_impl/mibs/vendor/cisco/__init__.py

@@ -1 +1,2 @@
 from . import mgmt, bgp4
+from . import ciscoEntityFruControlMIB

src/sonic_ax_impl/mibs/vendor/cisco/ciscoEntityFruControlMIB.py

@@ -0,0 +1,102 @@
+import imp
+import re
+import sys
+
+from sonic_ax_impl import mibs
+from ax_interface import MIBMeta, ValueType, MIBUpdater, MIBEntry, SubtreeMIBEntry
+from ax_interface.encodings import ObjectIdentifier
+
+PSU_PLUGIN_MODULE_NAME = 'psuutil'
+PSU_PLUGIN_MODULE_PATH = "/usr/share/sonic/platform/plugins/{}.py".format(PSU_PLUGIN_MODULE_NAME)
+PSU_PLUGIN_CLASS_NAME = 'PsuUtil'
+
+class PowerStatusHandler:
+    """
+    Class to handle the SNMP request
+    """
+    def __init__(self):
+        """
+        init the handler
+        """
+        self.psuutil = None
+
+        try:
+            module = imp.load_source(PSU_PLUGIN_MODULE_NAME, PSU_PLUGIN_MODULE_PATH)
+        except ImportError as e:
+            mibs.logger.error("Failed to load PSU module '%s': %s" % (PSU_PLUGIN_MODULE_NAME, str(e)), True)
+            return
+        except FileNotFoundError as e:
+            mibs.logger.error("Failed to get platform specific PSU module '%s': %s" % (PSU_PLUGIN_MODULE_NAME, str(e)), True)
+            return
+
+        try:
+            platform_psuutil_class = getattr(module, PSU_PLUGIN_CLASS_NAME)
+            self.psuutil = platform_psuutil_class()
+        except AttributeError as e:
+            mibs.logger.error("Failed to instantiate '%s' class: %s" % (PLATFORM_SPECIFIC_CLASS_NAME, str(e)), True)
+
+    def _getPsuIndex(self, sub_id):
+        """
+        Get the PSU index from sub_id
+        :return: the index of supported PSU
+        """
+        if not self.psuutil or not sub_id or len(sub_id) > 1:
+            return None
+
+        psu_index = int(sub_id[0])
+
+        if psu_index < 1 or psu_index > self.psuutil.get_num_psus():
+            return None
+
+        return psu_index
+
+    def get_next(self, sub_id):
+        """
+        :param sub_id: The 1-based snmp sub-identifier query.
+        :return: the next sub id.
+        """
+        if not self.psuutil:
+            return None
+
+        if not sub_id:
+            return (1,)
+
+        psu_index = self._getPsuIndex(sub_id)
+
+        if psu_index and psu_index + 1 <= self.psuutil.get_num_psus():
+            return (psu_index + 1,)
+
+        return None
+
+    def getPsuStatus(self, sub_id):
+        """
+        :param sub_id: The 1-based sub-identifier query.
+        :return: the status of requested PSU according to cefcModuleOperStatus ModuleOperType
+                 2 - PSU has correct functionalling - ok
+                 7 - PSU has a problem with functionalling - failed
+        :ref: https://www.cisco.com/c/en/us/td/docs/switches/wan/mgx/mgx_8850/software/mgx_r2-0-10/pxm/reference/guide/pxm/cscoent.html
+        """
+        psu_index = self._getPsuIndex(sub_id)
+
+        if not psu_index:
+            return None
+
+        psu_status = self.psuutil.get_psu_status(psu_index)
+
+        if psu_status:
+            return 2
+
+        return 7
+
+
+class cefcFruPowerStatusTable(metaclass=MIBMeta, prefix='.1.3.6.1.4.1.9.9.117.1.1.2'):
+    """
+    'cefcFruPowerStatusTable' http://oidref.com/1.3.6.1.4.1.9.9.117.1.1.2
+    """
+
+    power_status_handler = PowerStatusHandler()
+
+    # cefcFruPowerStatusTable = '1.3.6.1.4.1.9.9.117.1.1.2'
+    # csqIfQosGroupStatsEntry = '1.3.6.1.4.1.9.9.117.1.1.2.1'
+
+    psu_status = SubtreeMIBEntry('1.2', power_status_handler, ValueType.INTEGER, power_status_handler.getPsuStatus)

tests/__init__.py

@@ -1 +1,2 @@
 import tests.mock_tables.socket
+import tests.mock_tables.imp

tests/mock_tables/imp.py

@@ -0,0 +1,15 @@
+import imp
+import os
+
+TEST_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# Backup original function
+_load_source = getattr(imp, 'load_source')
+
+# Monkey patch
+def load_source(name, pathname):
+    if name == 'psuutil':
+        return _load_source(name, TEST_DIR + '/plugins/psuutil.py')
+
+# Replace the function with mocked one
+imp.load_source = load_source

tests/plugins/psuutil.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python
+
+#############################################################################
+# Mellanox
+#
+# Module contains an implementation of SONiC PSU Base API and
+# provides the PSUs status for SNMP testing
+#
+#############################################################################
+
+class PsuUtil():
+    """PSUutil class for SNMP testing"""
+
+    def __init__(self):
+        """ For testing purpose only """
+        self.num_of_psus = 2
+        self.psu_status = { 1 : True, 2 : False }
+
+    def get_num_psus(self):
+        """
+        Retrieves the number of PSUs available on the device
+
+        :return: An integer, the number of PSUs available on the device
+        """
+        """ For testing purpose only """
+        return self.num_of_psus
+
+    def get_psu_status(self, index):
+        """
+        Retrieves the oprational status of power supply unit (PSU) defined
+                by 1-based index <index>
+
+        :param index: An integer, 1-based index of the PSU of which to query status
+        :return: Boolean, True if PSU is operating properly, False if PSU is faulty
+        """
+        """ For testing purpose only """
+        if not isinstance(index, int):
+            return False
+        elif index > 0 and index <= self.num_of_psus:
+            return self.psu_status[index]
+        else:
+            return False
+
+    def get_psu_presence(self, index):
+        """
+        Retrieves the presence status of power supply unit (PSU) defined
+                by 1-based index <index>
+
+        :param index: An integer, 1-based index of the PSU of which to query status
+        :return: Boolean, True if PSU is plugged, False if not
+        """
+        """ For testing purpose only """
+        if not isinstance(index, int):
+            return False
+        elif index > 0 and index <= self.num_of_psus:
+            return self.psu_status[index]
+        else:
+            return False

tests/test_psu.py

@@ -0,0 +1,101 @@
+import os
+import sys
+
+# noinspection PyUnresolvedReferences
+import tests.mock_tables.dbconnector
+
+modules_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(modules_path, 'src'))
+
+from unittest import TestCase
+
+from ax_interface import ValueType
+from ax_interface.pdu_implementations import GetPDU, GetNextPDU
+from ax_interface.encodings import ObjectIdentifier
+from ax_interface.constants import PduTypes
+from ax_interface.pdu import PDU, PDUHeader
+from ax_interface.mib import MIBTable
+from sonic_ax_impl.mibs.vendor.cisco import ciscoEntityFruControlMIB
+
+class TestPsuStatus(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.lut = MIBTable(ciscoEntityFruControlMIB.cefcFruPowerStatusTable)
+
+    def test_getPsu1Status(self):
+        oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 1))
+        get_pdu = GetPDU(
+            header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0),
+            oids=[oid]
+        )
+
+        encoded = get_pdu.encode()
+        response = get_pdu.make_response(self.lut)
+
+        value0 = response.values[0]
+        self.assertEqual(value0.type_, ValueType.INTEGER)
+        self.assertEqual(str(value0.name), str(oid))
+        self.assertEqual(value0.data, 2)
+
+    def test_getNextPsu1(self):
+        oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 1))
+        expected_oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 2))
+        get_pdu = GetNextPDU(
+            header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0),
+            oids=[oid]
+        )
+
+        encoded = get_pdu.encode()
+        response = get_pdu.make_response(self.lut)
+
+        value0 = response.values[0]
+        self.assertEqual(value0.type_, ValueType.INTEGER)
+        self.assertEqual(str(value0.name), str(expected_oid))
+        self.assertEqual(value0.data, 7)
+
+    def test_getPsu2Status(self):
+        oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 2))
+        get_pdu = GetPDU(
+            header=PDUHeader(1, PduTypes.GET, 16, 0, 42, 0, 0, 0),
+            oids=[oid]
+        )
+
+        encoded = get_pdu.encode()
+        response = get_pdu.make_response(self.lut)
+
+        value0 = response.values[0]
+        self.assertEqual(value0.type_, ValueType.INTEGER)
+        self.assertEqual(str(value0.name), str(oid))
+        self.assertEqual(value0.data, 7)
+
+    def test_getNextPsu2(self):
+        oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 2))
+        expected_oid = None
+        get_pdu = GetNextPDU(
+            header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0),
+            oids=[oid]
+        )
+
+        encoded = get_pdu.encode()
+        response = get_pdu.make_response(self.lut)
+
+        value0 = response.values[0]
+        self.assertEqual(value0.type_, ValueType.END_OF_MIB_VIEW)
+        self.assertEqual(str(value0.name), str(oid))
+        self.assertEqual(value0.data, None)
+
+    def test_getMissedPsu(self):
+        oid = ObjectIdentifier(2, 0, 0, 0, (1, 3, 6, 1, 4, 1, 9, 9, 117, 1, 1, 2, 1, 2, 5, 1))
+        expected_oid = None
+        get_pdu = GetNextPDU(
+            header=PDUHeader(1, PduTypes.GET_NEXT, 16, 0, 42, 0, 0, 0),
+            oids=[oid]
+        )
+
+        encoded = get_pdu.encode()
+        response = get_pdu.make_response(self.lut)
+
+        value0 = response.values[0]
+        self.assertEqual(value0.type_, ValueType.END_OF_MIB_VIEW)
+        self.assertEqual(str(value0.name), str(oid))
+        self.assertEqual(value0.data, None)