[Mellanox] Add bitmap support for SFP error event

platform/mellanox/mlnx-platform-api/sonic_platform/sfp_event.py

@@ -8,7 +8,14 @@
 import os
 import time
 import select
-from python_sdk_api.sx_api import *
+if 'MLNX_PLATFORM_API_UNIT_TESTING' not in os.environ:
+    from python_sdk_api.sx_api import *
+else:
+    from mock import MagicMock
+    class MockSxFd(object):
+        fd = 99
+    new_sx_fd_t_p = MagicMock(return_value=MockSxFd())
+    new_sx_user_channel_t_p = MagicMock()
 from sonic_py_common.logger import Logger
 
 # SFP status from PMAOS register
@@ -24,13 +31,14 @@
 SDK_SFP_STATE_DIS = 0x4
 
 # SFP status that will be handled by XCVRD 
-STATUS_PLUGIN = '1'
-STATUS_PLUGOUT = '0'
-STATUS_ERR_I2C_STUCK = '2'
-STATUS_ERR_BAD_EEPROM = '3'
-STATUS_ERR_UNSUPPORTED_CABLE = '4'
-STATUS_ERR_HIGH_TEMP = '5'
-STATUS_ERR_BAD_CABLE = '6'
+STATUS_PLUGIN = '1' # 00000001
+STATUS_PLUGOUT = '0' # 00000000
+# SFP error status always come with STATUS_PLUGIN, so the last bit is always 1
+STATUS_ERR_I2C_STUCK = '3'  # 00000011
+STATUS_ERR_BAD_EEPROM = '5' # 00000101
+STATUS_ERR_UNSUPPORTED_CABLE = '9' # 00001001
+STATUS_ERR_HIGH_TEMP = '17' # 00010001
+STATUS_ERR_BAD_CABLE = '33' # 00100001
 
 # SFP status used in this file only, will not expose to XCVRD
 # STATUS_ERROR will be mapped to different status according to the error code

platform/mellanox/mlnx-platform-api/tests/test_sfp_event.py

@@ -0,0 +1,36 @@
+import os
+import select
+import sys
+
+from mock import MagicMock
+
+test_path = os.path.dirname(os.path.abspath(__file__))
+modules_path = os.path.dirname(test_path)
+sys.path.insert(0, modules_path)
+
+class TestSfpEvent(object):
+    @classmethod
+    def setup_class(cls):
+        os.environ["MLNX_PLATFORM_API_UNIT_TESTING"] = "1"
+        select.select = MagicMock(return_value=([99], None, None))
+
+    def test_check_sfp_status(self):
+        from sonic_platform.sfp_event import SDK_SFP_STATE_IN, SDK_SFP_STATE_OUT, SDK_SFP_STATE_ERR
+        from sonic_platform.sfp_event import STATUS_PLUGIN, STATUS_PLUGOUT
+        from sonic_platform.sfp_event import sdk_sfp_err_type_dict
+
+        self.executor(SDK_SFP_STATE_IN, None, STATUS_PLUGIN)
+        self.executor(SDK_SFP_STATE_OUT, None, STATUS_PLUGOUT)
+        for error_type, error_status in sdk_sfp_err_type_dict.items():
+            self.executor(SDK_SFP_STATE_ERR, error_type, error_status)
+
+    def executor(self, mock_module_state, mock_error_type, expect_status):
+        from sonic_platform.sfp_event import sfp_event
+
+        event = sfp_event()
+        event.on_pmpe = MagicMock(return_value=(True, [0,1], mock_module_state, mock_error_type))
+        port_change = {}
+        found = event.check_sfp_status(port_change, 0)
+        assert found
+        assert 1 in port_change and port_change[1] == expect_status
+        assert 2 in port_change and port_change[2] == expect_status