diff --git a/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py b/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py index bb7e3a703..92a4a50fb 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py +++ b/sonic_platform_base/sonic_xcvr/api/public/c_cmis.py @@ -147,6 +147,43 @@ def set_tx_power(self, tx_power): time.sleep(1) return status + def freeze_vdm_stats(self): + ''' + This function freeze all the vdm statistics reporting registers. + When raised by the host, causes the module to freeze and hold all + reported statistics reporting registers (minimum, maximum and + average values)in Pages 24h-27h. + + Returns True if the provision succeeds and False incase of failure. + ''' + return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE) + + def get_vdm_freeze_status(self): + ''' + This function reads and returns the vdm Freeze done status. + + Returns True if the vdm stats freeze is successful and False if not freeze. + ''' + return self.xcvr_eeprom.read(consts.VDM_FREEZE_DONE) + + def unfreeze_vdm_stats(self): + ''' + This function unfreeze all the vdm statistics reporting registers. + When freeze is ceased by the host, releases the freeze request, allowing the + reported minimum, maximum and average values to update again. + + Returns True if the provision succeeds and False incase of failure. + ''' + return self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE) + + def get_vdm_unfreeze_status(self): + ''' + This function reads and returns the vdm unfreeze status. + + Returns True if the vdm stats unfreeze is successful and False if not unfreeze. + ''' + return self.xcvr_eeprom.read(consts.VDM_UNFREEZE_DONE) + def get_pm_all(self): ''' This function returns the PMs reported in Page 34h and 35h in OIF C-CMIS document @@ -163,14 +200,6 @@ def get_pm_all(self): SOPROC: unit in krad/s MER: unit in dB ''' - # When raised by the host, causes the module to freeze and hold all - # reported statistics reporting registers (minimum, maximum and - # average values)in Pages 24h-27h. - # When ceased by the host, releases the freeze request, allowing the - # reported minimum, maximum and average values to update again. - self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE) - time.sleep(1) - self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE) PM_dict = dict() rx_bits_pm = self.xcvr_eeprom.read(consts.RX_BITS_PM) @@ -256,7 +285,6 @@ def get_pm_all(self): PM_dict['rx_mer_max'] = self.xcvr_eeprom.read(consts.RX_MAX_MER_PM) return PM_dict - def get_transceiver_info(self): """ Retrieves transceiver info of this SFP diff --git a/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py b/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py index a76c7a39e..dba2a7350 100644 --- a/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py +++ b/sonic_platform_base/sonic_xcvr/api/public/cmisVDM.py @@ -170,14 +170,6 @@ def get_vdm_allpage(self): return None VDM_START_PAGE = 0x20 vdm = dict() - # When raised by the host, causes the module to freeze and hold all - # reported statistics reporting registers (minimum, maximum and - # average values)in Pages 24h-27h. - # When ceased by the host, releases the freeze request, allowing the - # reported minimum, maximum and average values to update again. - self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_FREEZE) - time.sleep(1) - self.xcvr_eeprom.write(consts.VDM_CONTROL, VDM_UNFREEZE) vdm_flag_page = self.xcvr_eeprom.read_raw(VDM_FLAG_PAGE * PAGE_SIZE + PAGE_OFFSET, PAGE_SIZE) for page in range(VDM_START_PAGE, VDM_START_PAGE + vdm_page_supported_raw + 1): vdm_current_page = self.get_vdm_page(page, vdm_flag_page) diff --git a/sonic_platform_base/sonic_xcvr/fields/consts.py b/sonic_platform_base/sonic_xcvr/fields/consts.py index c36e46601..68ff4b2da 100644 --- a/sonic_platform_base/sonic_xcvr/fields/consts.py +++ b/sonic_platform_base/sonic_xcvr/fields/consts.py @@ -413,6 +413,10 @@ VDM_SUPPORTED = "VdmSupported" VDM_SUPPORTED_PAGE = "VdmSupportedPage" VDM_CONTROL = "VdmControl" +VDM_STATUS = "VdmStatus" +VDM_FREEZE_DONE = "VdmFreezeDone" +VDM_UNFREEZE_DONE = "VdmUnfreezeDone" + MEDIA_LANE_FEC_PM = "Media Lane FEC Performance Monitoring" MEDIA_LANE_LINK_PM = "Media Lane Link Performance Monitoring" RX_BITS_PM = "rxBitsPm" diff --git a/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py b/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py index 140faeafd..8702e76f3 100644 --- a/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py +++ b/sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py @@ -449,6 +449,10 @@ def __init__(self, codes): *(RegBitField("Bit%d" % (bit), bit) for bit in range (0, 2)) ), NumberRegField(consts.VDM_CONTROL, self.getaddr(0x2f, 144), size=1, ro=False), + NumberRegField(consts.VDM_STATUS, self.getaddr(0x2f, 145), + RegBitField(consts.VDM_UNFREEZE_DONE, 6), + RegBitField(consts.VDM_FREEZE_DONE, 7), + ), ) self.TRANS_CONFIG = RegGroupField(consts.TRANS_CONFIG_FIELD, diff --git a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py index 668389e14..54e5ebf44 100644 --- a/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py +++ b/sonic_platform_base/sonic_xcvr/sfp_optoe_base.py @@ -51,6 +51,61 @@ def get_transceiver_pm(self): api = self.get_xcvr_api() return api.get_transceiver_pm() if api is not None else None + def freeze_vdm_stats(self): + ''' + This function freeze all the vdm statistics reporting registers. + When raised by the host, causes the module to freeze and hold all + reported statistics reporting registers (minimum, maximum and + average values)in Pages 24h-27h. + + Returns True if the provision succeeds and False incase of failure. + ''' + api = self.get_xcvr_api() + try: + return api.freeze_vdm_stats() if api is not None else False + except (NotImplementedError, AttributeError): + return False + + def unfreeze_vdm_stats(self): + ''' + This function unfreeze all the vdm statistics reporting registers. + When freeze is ceased by the host, releases the freeze request, allowing the + reported minimum, maximum and average values to update again. + + Returns True if the provision succeeds and False incase of failure. + ''' + api = self.get_xcvr_api() + try: + return api.unfreeze_vdm_stats() if api is not None else False + except (NotImplementedError, AttributeError): + return False + + + def get_vdm_freeze_status(self): + ''' + This function reads and returns the vdm Freeze done status. + + Returns True if the vdm stats freeze is successful and False if not freeze. + ''' + api = self.get_xcvr_api() + try: + return api.get_vdm_freeze_status() if api is not None else False + except (NotImplementedError, AttributeError): + return False + + def get_vdm_unfreeze_status(self): + ''' + This function reads and returns the vdm unfreeze status. + + Returns True if the vdm stats unfreeze is successful and False if not unfreeze. + ''' + api = self.get_xcvr_api() + try: + return api.get_vdm_unfreeze_status() if api is not None else False + except (NotImplementedError, AttributeError): + return False + + def get_rx_los(self): api = self.get_xcvr_api() if api is not None: diff --git a/tests/sonic_xcvr/test_ccmis.py b/tests/sonic_xcvr/test_ccmis.py index 8b2eb7b7e..75fa2c146 100644 --- a/tests/sonic_xcvr/test_ccmis.py +++ b/tests/sonic_xcvr/test_ccmis.py @@ -624,3 +624,44 @@ def test_get_transceiver_pm(self, mock_response, expected): self.api.get_pm_all.return_value = mock_response result = self.api.get_transceiver_pm() assert result == expected + + @pytest.mark.parametrize("mock_response, expected", [ + (0, 0), + (1, 1), + ]) + def test_get_vdm_freeze_status(self, mock_response, expected): + self.api.xcvr_eeprom.read = MagicMock() + self.api.xcvr_eeprom.read.return_value = mock_response + result = self.api.get_vdm_freeze_status() + assert result == expected + + @pytest.mark.parametrize("mock_response, expected", [ + (0, 0), + (1, 1), + ]) + def test_get_vdm_unfreeze_status(self, mock_response, expected): + self.api.xcvr_eeprom.read = MagicMock() + self.api.xcvr_eeprom.read.return_value = mock_response + result = self.api.get_vdm_unfreeze_status() + assert result == expected + + @pytest.mark.parametrize("mock_response, expected", [ + (0, 0), + (1, 1), + ]) + def test_freeze_vdm_stats(self, mock_response, expected): + self.api.xcvr_eeprom.write = MagicMock() + self.api.xcvr_eeprom.write.return_value = mock_response + result = self.api.freeze_vdm_stats() + assert result == expected + + @pytest.mark.parametrize("mock_response, expected", [ + (0, 0), + (1, 1), + ]) + def test_unfreeze_vdm_stats(self, mock_response, expected): + self.api.xcvr_eeprom.write = MagicMock() + self.api.xcvr_eeprom.write.return_value = mock_response + result = self.api.unfreeze_vdm_stats() + assert result == expected + diff --git a/tests/sonic_xcvr/test_sfp_optoe_base.py b/tests/sonic_xcvr/test_sfp_optoe_base.py new file mode 100644 index 000000000..5b18beb0e --- /dev/null +++ b/tests/sonic_xcvr/test_sfp_optoe_base.py @@ -0,0 +1,81 @@ +from mock import MagicMock +from mock import patch +import pytest +from sonic_platform_base.sonic_xcvr.sfp_optoe_base import SfpOptoeBase +from sonic_platform_base.sonic_xcvr.api.public.c_cmis import CCmisApi +from sonic_platform_base.sonic_xcvr.api.public.cmis import CmisApi +from sonic_platform_base.sonic_xcvr.mem_maps.public.c_cmis import CCmisMemMap +from sonic_platform_base.sonic_xcvr.xcvr_eeprom import XcvrEeprom +from sonic_platform_base.sonic_xcvr.codes.public.cmis import CmisCodes + +class TestSfpOptoeBase(object): + + codes = CmisCodes + mem_map = CCmisMemMap(codes) + reader = MagicMock(return_value=None) + writer = MagicMock() + eeprom = XcvrEeprom(reader, writer, mem_map) + sfp_optoe_api = SfpOptoeBase() + ccmis_api = CCmisApi(eeprom) + cmis_api = CmisApi(eeprom) + + @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ + (0, ccmis_api, 0), + (1, ccmis_api, 1), + (None, None, False), + (None, cmis_api, False), + ]) + def test_freeze_vdm_stats(self, mock_response1, mock_response2, expected): + self.sfp_optoe_api.get_xcvr_api = MagicMock() + self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 + self.ccmis_api.freeze_vdm_stats = MagicMock() + self.ccmis_api.freeze_vdm_stats.return_value = mock_response1 + + result = self.sfp_optoe_api.freeze_vdm_stats() + assert result == expected + + @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ + (0, ccmis_api, 0), + (1, ccmis_api, 1), + (None, None, False), + (None, cmis_api, False), + ]) + def test_unfreeze_vdm_stats(self, mock_response1, mock_response2, expected): + self.sfp_optoe_api.get_xcvr_api = MagicMock() + self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 + self.ccmis_api.unfreeze_vdm_stats = MagicMock() + self.ccmis_api.unfreeze_vdm_stats.return_value = mock_response1 + + result = self.sfp_optoe_api.unfreeze_vdm_stats() + assert result == expected + + @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ + (0, ccmis_api, 0), + (1, ccmis_api, 1), + (None, None, False), + (None, cmis_api, False), + ]) + def test_get_vdm_freeze_status(self, mock_response1, mock_response2, expected): + self.sfp_optoe_api.get_xcvr_api = MagicMock() + self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 + self.ccmis_api.get_vdm_freeze_status = MagicMock() + self.ccmis_api.get_vdm_freeze_status.return_value = mock_response1 + + result = self.sfp_optoe_api.get_vdm_freeze_status() + assert result == expected + + @pytest.mark.parametrize("mock_response1, mock_response2, expected", [ + (0, ccmis_api, 0), + (1, ccmis_api, 1), + (None, None, False), + (None, cmis_api, False), + ]) + def test_get_vdm_unfreeze_status(self, mock_response1, mock_response2, expected): + self.sfp_optoe_api.get_xcvr_api = MagicMock() + self.sfp_optoe_api.get_xcvr_api.return_value = mock_response2 + self.ccmis_api.get_vdm_unfreeze_status = MagicMock() + self.ccmis_api.get_vdm_unfreeze_status.return_value = mock_response1 + + result = self.sfp_optoe_api.get_vdm_unfreeze_status() + assert result == expected +