Add support for rx_disable, rx_disable_channel in CMIS_API for transceivers (sonic-net#74)

<!-- Provide a general summary of your changes in the Title above -->

#### Description
This PR introduces **RX disable support** for CMIS-compliant transceivers, aligning it with TX disable functionality.
<!--
 Describe your changes in detail
 This PR introduces **RX disable support** for CMIS-compliant transceivers, aligning it with TX disable functionality.
-->

#### Motivation and Context
- Added the following methods in `CmisApi` and `XcvrApi`:
 - `get_rx_disable()`
 - `get_rx_disable_channel()`
 - `rx_disable()`
 - `rx_disable_channel()`
- Updated `SfpOptoeBase` to expose RX disable functionality.
<!--
 Why is this change required? What problem does it solve?
 If this pull request closes/resolves an open Issue, make sure you
 include the text "fixes #xxxx", "closes #xxxx" or "resolves #xxxx" here
-->

- Defined `RX_DISABLE_SUPPORT_FIELD` in `consts.py`.
- Added `RX_DISABLE_FIELD` at **Byte 138** in `CmisMemMap`.
- Updated `CTRLS_ADVT_FIELD` to include `RX_DISABLE_SUPPORT_FIELD`.

- Integrated RX disable methods into `sfp_optoe_base.py`, mirroring TX disable operations.

#### How Has This Been Tested?
UT and deploy on testbed, please see attached testbed results
<!--
 Please describe in detail how you tested your changes.
 Include details of your testing environment, and the tests you ran to
 see how your change affects other areas of the code, etc.
-->
```
sfp1 = platform_chassis.get_sfp(1)>>> platform_chassis = sonic_platform.platform.Platform().get_chassis()
>>> sfp1 = platform_chassis.get_sfp(1)
>>> sfp1.get_rx_disable()
[True, False, False, False, False, False, False, False]
>>> sfp1.rx_disable_channel(1, 1)
True
>>> sfp1.get_rx_disable()
[True, False, False, False, False, False, False, False]
>>> sfp1.rx_disable_channel(1, 0)
True
>>> sfp1.get_rx_disable()
[False, False, False, False, False, False, False, False]
>>> sfp1.rx_disable_channel(2, 0)
True
>>> sfp1.rx_disable_channel(2, 1)
True
>>> sfp1.get_rx_disable()
[False, True, False, False, False, False, False, False]
>>> sfp1.rx_disable_channel(2, 0)
True
>>> sfp1.get_rx_disable()
[False, False, False, False, False, False, False, False]
>>>
```
#### Additional Information (Optional)

Author: mssonicbld

sonic_platform_base/sonic_xcvr/api/public/cmis.py

@@ -869,6 +869,52 @@ def tx_disable_channel(self, channel, disable):
 
         return self.xcvr_eeprom.write(consts.TX_DISABLE_FIELD, channel_state)
 
+    def get_rx_disable_support(self):
+        return not self.is_flat_memory() and self.xcvr_eeprom.read(consts.RX_DISABLE_SUPPORT_FIELD)
+
+    def get_rx_disable(self):
+        rx_disable_support = self.get_rx_disable_support()
+        if rx_disable_support is None:
+            return None
+        if not rx_disable_support:
+            return ["N/A" for _ in range(self.NUM_CHANNELS)]
+        rx_disable = self.xcvr_eeprom.read(consts.RX_DISABLE_FIELD)
+        if rx_disable is None:
+            return None
+        return [bool(rx_disable & (1 << i)) for i in range(self.NUM_CHANNELS)]
+
+    def rx_disable(self, rx_disable):
+        val = 0xFF if rx_disable else 0x0
+        return self.xcvr_eeprom.write(consts.RX_DISABLE_FIELD, val)
+
+    def get_rx_disable_channel(self):
+        rx_disable_support = self.get_rx_disable_support()
+        if rx_disable_support is None:
+            return None
+        if not rx_disable_support:
+            return 'N/A'
+        return self.xcvr_eeprom.read(consts.RX_DISABLE_FIELD)
+
+    def rx_disable_channel(self, channel, disable):
+        # Check if channel state is available
+        channel_state = self.get_rx_disable_channel()
+        if channel_state is None or channel_state == 'N/A':
+            return False
+
+        # Disable or enable the specific channel based on the input
+        if 1 <= channel <= self.NUM_CHANNELS:
+            # The channel is represented by individual RegBitField entries
+            field_name = f"{consts.RX_DISABLE_FIELD}_{channel}"
+
+            if disable:
+                # Disable the channel (set the bit to 1)
+                return self.xcvr_eeprom.write(field_name, 1)
+            else:
+                # Enable the channel (set the bit to 0)
+                return self.xcvr_eeprom.write(field_name, 0)
+
+        return False
+
     def get_laser_tuning_summary(self):
         '''
         This function returns laser tuning status summary on media lane

sonic_platform_base/sonic_xcvr/api/xcvr_api.py

@@ -28,6 +28,38 @@ def get_model(self):
         """
         raise NotImplementedError
 
+    def get_rx_disable(self):
+        """
+        Retrieves the rx_disable status of this xcvr.
+
+        Returns:
+            A list of boolean values, representing the RX disable status
+            of each available channel. The value is True if the xcvr channel
+            is RX disabled, False if not.
+
+            E.g., for a transceiver with four channels: [False, False, True, False]
+
+            If RX disable status is unsupported on the xcvr, each list element should be "N/A" instead.
+
+            If there is an issue with reading the xcvr, None should be returned.
+        """
+        raise NotImplementedError
+
+    def get_rx_disable_channel(self):
+        """
+        Retrieves the RX disabled channels in this xcvr.
+
+        Returns:
+            A hex of 4 bits (bit 0 to bit 3 as channel 0 to channel 3) to represent
+            RX channels which have been disabled in this xcvr.
+            As an example, a returned value of 0x5 indicates that channel 0
+            and channel 2 have been disabled.
+
+            If there is an issue with reading the xcvr, None should be returned.
+        """
+        raise NotImplementedError
+
+
     def get_serial(self):
         """
         Retrieves the serial number of the xcvr
@@ -397,6 +429,34 @@ def get_tx_disable_channel(self):
         """
         raise NotImplementedError
 
+    def rx_disable(self, rx_disable):
+        """
+        Disable xcvr RX for all channels
+
+        Args:
+            rx_disable : A Boolean, True to enable rx_disable mode, False to disable
+                         rx_disable mode.
+
+        Returns:
+            A boolean, True if rx_disable is set successfully, False if not
+        """
+        raise NotImplementedError
+
+    def rx_disable_channel(self, channel, disable):
+        """
+        Sets the rx_disable for specified xcvr channels
+
+        Args:
+            channel : A hex of 4 bits (bit 0 to bit 3) which represent channel 0 to 3,
+                      e.g. 0x5 for channel 0 and channel 2.
+            disable : A boolean, True to disable RX channels specified in channel,
+                      False to enable
+
+        Returns:
+            A boolean, True if successful, False if not
+        """
+        raise NotImplementedError
+
     def get_module_temperature(self):
         """
         Retrieves the temperature of this xcvr

sonic_platform_base/sonic_xcvr/fields/consts.py

@@ -32,6 +32,9 @@
 RX_POWER_HIGH_WARNING_FIELD = "RxPowerHighWarning"
 RX_POWER_LOW_WARNING_FIELD = "RxPowerLowWarning"
 
+RX_DISABLE_FIELD = "RxDisable"
+RX_DISABLE_SUPPORT_FIELD = "RxDisableSupported"
+
 TEMPERATURE_FIELD = "Temperature"
 TEMP_SUPPORT_FIELD = "Temperature Monitoring Implemented"
 TEMP_THRESHOLDS_FIELD = "TempThresholds"

sonic_platform_base/sonic_xcvr/mem_maps/public/cmis.py

@@ -215,6 +215,10 @@ def __init__(self, codes):
                 RegBitField(consts.TX_DISABLE_SUPPORT_FIELD, 1),
                 size=2, format="<H"
             ),
+            NumberRegField(consts.CTRLS_ADVT_FIELD, self.getaddr(0x1, 156),
+                RegBitField(consts.RX_DISABLE_SUPPORT_FIELD, 1),
+                size=2, format="<H"
+            ),
             NumberRegField(consts.TX_FLAGS_ADVT_FIELD, self.getaddr(0x1, 157),
                 RegBitField(consts.TX_FAULT_SUPPORT_FIELD, 0),
                 RegBitField(consts.TX_LOS_SUPPORT_FIELD, 1),
@@ -272,7 +276,11 @@ def __init__(self, codes):
 
         self.LANE_DATAPATH_CTRL = RegGroupField(consts.LANE_DATAPATH_CTRL_FIELD,
             NumberRegField(consts.DATAPATH_DEINIT_FIELD, self.getaddr(0x10, 128), ro=False),
-            NumberRegField(consts.TX_DISABLE_FIELD, self.getaddr(0x10, 130), ro=False)
+            NumberRegField(consts.TX_DISABLE_FIELD, self.getaddr(0x10, 130), ro=False),
+            NumberRegField(consts.RX_DISABLE_FIELD, self.getaddr(0x10, 138), ro=False,
+                *(RegBitField("%s_%d" % (consts.RX_DISABLE_FIELD, channel), bitpos, ro=False) 
+                  for channel, bitpos in zip(range(1, 9), range(0, 8)))  # 8 channels
+            )
         )
 
         self.LANE_DATAPATH_STATUS = RegGroupField(consts.LANE_DATAPATH_STATUS_FIELD,

sonic_platform_base/sonic_xcvr/sfp_optoe_base.py

@@ -160,6 +160,14 @@ def get_tx_fault(self):
             return tx_fault
         return None
 
+    def get_rx_disable(self):
+        api = self.get_xcvr_api()
+        return api.get_rx_disable() if api is not None else None
+
+    def get_rx_disable_channel(self):
+        api = self.get_xcvr_api()
+        return api.get_rx_disable_channel() if api is not None else None
+
     def get_tx_disable(self):
         api = self.get_xcvr_api()
         return api.get_tx_disable() if api is not None else None
@@ -224,6 +232,15 @@ def tx_disable_channel(self, channel, disable):
         api = self.get_xcvr_api()
         return api.tx_disable_channel(channel, disable) if api is not None else None
 
+    def rx_disable(self, rx_disable):
+        api = self.get_xcvr_api()
+        return api.rx_disable(rx_disable) if api is not None else None
+
+    def rx_disable_channel(self, channel, disable):
+        api = self.get_xcvr_api()
+        return api.rx_disable_channel(channel, disable) if api is not None else None
+
+
     def get_power_override(self):
         api = self.get_xcvr_api()
         return api.get_power_override() if api is not None else None

tests/sonic_xcvr/test_cmis.py

@@ -575,6 +575,81 @@ def test_tx_disable_channel(self, mock_response, input_param):
         self.api.get_tx_disable_channel.return_value = mock_response
         self.api.tx_disable_channel(*input_param)
 
+    @pytest.mark.parametrize("mock_response, expected", [
+        ([True, {'RxLOS1': 0}], [False]),
+        ([False, {'RxLOS1': 0}], ['N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A']),
+        ([None, None], None),
+        ([True, None], None)
+    ])
+    def test_get_rx_los(self, mock_response, expected):
+        self.api.get_rx_los_support = MagicMock()
+        self.api.get_rx_los_support.return_value = mock_response[0]
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response[1]
+        result = self.api.get_rx_los()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        ([False, True], True)
+    ])
+    def test_get_rx_disable_support(self, mock_response, expected):
+        self.api.is_flat_memory = MagicMock()
+        self.api.is_flat_memory.return_value = mock_response[0]
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response[1]
+        result = self.api.get_rx_disable_support()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        ([True, 0x00], [False, False, False, False, False, False, False, False]),
+        ([False, 0x00], ['N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A', 'N/A']),
+        ([None, None], None),
+        ([True, None], None)
+    ])
+    def test_get_rx_disable(self, mock_response, expected):
+        self.api.get_rx_disable_support = MagicMock()
+        self.api.get_rx_disable_support.return_value = mock_response[0]
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response[1]
+        result = self.api.get_rx_disable()
+        assert result == expected
+
+    @pytest.mark.parametrize("input_param", [
+        (True), (False)
+    ])
+    def test_rx_disable(self, input_param):
+        rc = self.api.rx_disable(input_param)
+        assert(rc != None)
+
+    @pytest.mark.parametrize("mock_response, expected", [
+        ([True, 0x00], 0),
+        ([False, 0x00], 'N/A'),
+        ([None, None], None)
+    ])
+    def test_get_rx_disable_channel(self, mock_response, expected):
+        self.api.get_rx_disable_support = MagicMock()
+        self.api.get_rx_disable_support.return_value = mock_response[0]
+        self.api.xcvr_eeprom.read = MagicMock()
+        self.api.xcvr_eeprom.read.return_value = mock_response[1]
+        result = self.api.get_rx_disable_channel()
+        assert result == expected
+
+    @pytest.mark.parametrize("mock_response, input_param", [
+        (0, (0xff, True)),
+        (0, (0, True)),
+        (0, (0, False)),
+        (0, (1, False)),
+        (0, (1, True)),
+        (None, (0, False))
+    ])
+    def test_rx_disable_channel(self, mock_response, input_param):
+        self.api.get_rx_disable_channel = MagicMock()
+        self.api.get_rx_disable_channel.return_value = mock_response
+        self.api.xcvr_eeprom.write = MagicMock()
+        self.api.xcvr_eeprom.write.return_value = mock_response
+        rc = self.api.rx_disable_channel(*input_param)
+        assert(rc != None)
+
     @pytest.mark.parametrize("mock_response, expected", [
         (1, ['TuningComplete']),
         (62, ['TargetOutputPowerOOR', 'FineTuningOutOfRange', 'TuningNotAccepted',

tests/sonic_xcvr/test_sfp_optoe_base.py

@@ -60,6 +60,22 @@ def test_unfreeze_vdm_stats(self, mock_response1, mock_response2, expected):
         result = self.sfp_optoe_api.unfreeze_vdm_stats()
         assert result == expected
 
+    @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
+        (0, cmis_api, 0),
+        (1, cmis_api, 1),
+        (None, None, None),
+        (False, cmis_api, False),
+    ])
+    def test_get_rx_disable_channel(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.cmis_api.get_rx_disable_channel = MagicMock()
+        self.cmis_api.get_rx_disable_channel.return_value = mock_response1
+
+        result = self.sfp_optoe_api.get_rx_disable_channel()
+        assert result == expected
+
+
     @pytest.mark.parametrize("mock_response1, mock_response2, expected", [
         (0, cmis_api, 0),
         (1, cmis_api, 1),