g[sfputil debug] Fix issue: do not check output status when CMIS version is lower than 5.0

Author: Junchao-Mellanox

sfputil/debug.py

@@ -97,11 +97,13 @@ def set_output(port_name, enable, direction):
         if direction == "tx":
             lane_count = get_media_lane_count(port_name)
             disable_func = sfp.tx_disable_channel
+            get_disable_status_func = api.get_tx_disable
             get_status_func = api.get_tx_output_status
             status_key = "TxOutputStatus"
         elif direction == "rx":
             lane_count = get_host_lane_count(port_name)
             disable_func = sfp.rx_disable_channel
+            get_disable_status_func = api.get_rx_disable
             get_status_func = api.get_rx_output_status
             status_key = "RxOutputStatus"
 
@@ -112,6 +114,39 @@ def set_output(port_name, enable, direction):
 
         time.sleep(TX_RX_OUTPUT_UPDATE_WAIT_TIME)
 
+        disable_status_list = get_disable_status_func()
+        if not disable_status_list:
+            click.echo(f"{port_name}: {direction.upper()} disable status not available for subport {subport}")
+            sys.exit(EXIT_FAIL)
+
+        for lane in range(1, CMIS_MAX_CHANNELS + 1):
+            if lane_mask & (1 << (lane - 1)):
+                disable_status = disable_status_list[lane - 1]
+                if enable == "disable":
+                    if not disable_status or disable_status == "N/A":
+                        click.echo(f"{port_name}: {direction.upper()} output on lane {lane} is still "
+                                   f"enabled on subport {subport}. Restoring state.")
+                        sys.exit(EXIT_FAIL)
+                else:
+                    if disable_status or disable_status == "N/A":
+                        click.echo(f"{port_name}: {direction.upper()} output on lane {lane} is still "
+                                   f"disabled on subport {subport}. Restoring state.")
+                        sys.exit(EXIT_FAIL)
+
+        if hasattr(api, 'get_cmis_rev'):
+            cmis_rev = api.get_cmis_rev()
+            if cmis_rev is None:
+                click.echo(f"{port_name}: CMIS revision not available for subport {subport}")
+                sys.exit(EXIT_FAIL)
+
+            # OutputStatusRx and OutputStatusTx are supported from CMIS 5.0
+            if float(cmis_rev) < 5.0:
+                click.echo(
+                    f"{port_name}: {direction.upper()} output "
+                    f"{'disabled' if enable == 'disable' else 'enabled'} on subport {subport}"
+                )
+                return
+
         output_dict = get_status_func()
         if output_dict is None:
             click.echo(f"{port_name}: {direction.upper()} output status not available for subport {subport}")

tests/sfputil_test.py

@@ -1822,23 +1822,29 @@ def test_debug_loopback(self, mock_sonic_v2_connector, mock_config_db_connector,
         assert result.exit_code == EXIT_FAIL
 
     @pytest.mark.parametrize(
-        "direction, lane_count, enable, disable_func_result, output_dict, expected_echo, expected_exit",
+        "direction, lane_count, enable, disable_func_result, disable_status, output_dict, expected_echo, expected_exit",
         [
             # TX disable success
             (
-                "tx", 2, "disable", True, {"TxOutputStatus1": False, "TxOutputStatus2": False},
+                "tx", 2, "disable", True, [True, True], {"TxOutputStatus1": False, "TxOutputStatus2": False},
                 "TX output disabled", None
             ),
             # RX enable success
-            ("rx", 1, "enable", True, {"RxOutputStatus1": True}, "RX output enabled", None),
+            ("rx", 1, "enable", True, [False], {"RxOutputStatus1": True}, "RX output enabled", None),
             # TX disable fails to disable
-            ("tx", 1, "disable", True, {"TxOutputStatus1": True}, "TX output on lane 1 is still enabled", SystemExit),
+            (
+                "tx", 1, "disable", True, [True], {"TxOutputStatus1": True},
+                "TX output on lane 1 is still enabled", SystemExit
+            ),
             # RX enable fails to enable
-            ("rx", 1, "enable", True, {"RxOutputStatus1": False}, "RX output on lane 1 is still disabled", SystemExit),
+            (
+                "rx", 1, "enable", True, [False], {"RxOutputStatus1": False},
+                "RX output on lane 1 is still disabled", SystemExit
+            ),
             # TX disable_func returns False
-            ("tx", 1, "disable", False, {}, "TX disable failed", SystemExit),
+            ("tx", 1, "disable", False, [True], {}, "TX disable failed", SystemExit),
             # RX output_dict is None
-            ("rx", 1, "disable", True, None, "RX output status not available", SystemExit),
+            ("rx", 1, "disable", True, [True], None, "RX output status not available", SystemExit),
         ]
     )
     @patch("sfputil.debug.get_sfp_object")
@@ -1857,6 +1863,7 @@ def test_set_output_cli(
         lane_count,
         enable,
         disable_func_result,
+        disable_status,
         output_dict,
         expected_echo,
         expected_exit
@@ -1875,12 +1882,15 @@ def test_set_output_cli(
         # Mock SFP and API
         mock_sfp = MagicMock()
         mock_api = MagicMock()
+        mock_api.get_cmis_rev.return_value = "5.0"
         if direction == "tx":
             mock_sfp.tx_disable_channel.return_value = disable_func_result
             mock_api.get_tx_output_status.return_value = output_dict
+            mock_api.get_tx_disable.return_value = disable_status
         elif direction == "rx":
             mock_sfp.rx_disable_channel.return_value = disable_func_result
             mock_api.get_rx_output_status.return_value = output_dict
+            mock_api.get_rx_disable.return_value = disable_status
         mock_sfp.get_xcvr_api.return_value = mock_api
         mock_get_sfp_object.return_value = mock_sfp