[sfputil] Configure the debug loopback mode only on the relevant lanes of the logical port

sfputil/main.py

@@ -18,7 +18,7 @@
 import sonic_platform
 import sonic_platform_base.sonic_sfp.sfputilhelper
 from sonic_platform_base.sfp_base import SfpBase
-from swsscommon.swsscommon import SonicV2Connector
+from swsscommon.swsscommon import SonicV2Connector, ConfigDBConnector
 from natsort import natsorted
 from sonic_py_common import device_info, logger, multi_asic
 from utilities_common.sfp_helper import covert_application_advertisement_to_output_string
@@ -1469,10 +1469,10 @@ def is_fw_switch_done(port_name):
                 status = -1 # Abnormal status.
             elif (ImageARunning == 1) and (ImageACommitted == 0):   # ImageA is running, but not committed.
                 click.echo("FW images switch successful : ImageA is running")
-                status = 1  # run_firmware is done. 
+                status = 1  # run_firmware is done.
             elif (ImageBRunning == 1) and (ImageBCommitted == 0):   # ImageB is running, but not committed.
                 click.echo("FW images switch successful : ImageB is running")
-                status = 1  # run_firmware is done. 
+                status = 1  # run_firmware is done.
             else:                                                   # No image is running, or running and committed image is same.
                 click.echo("FW info error : Failed to switch into uncommitted image!")
                 status = -1 # Failure for Switching images.
@@ -1970,7 +1970,9 @@ def debug():
 @click.argument('port_name', required=True, default=None)
 @click.argument('loopback_mode', required=True, default="none",
                 type=click.Choice(["none", "host-side-input", "host-side-output",
-                                   "media-side-input", "media-side-output"]))
+                                   "media-side-input", "media-side-output",
+                                   "host-side-input-none", "host-side-output-none",
+                                   "media-side-input-none", "media-side-output-none"]))
 def loopback(port_name, loopback_mode):
     """Set module diagnostic loopback mode
     """
@@ -1991,11 +1993,50 @@ def loopback(port_name, loopback_mode):
         click.echo("{}: This functionality is not implemented".format(port_name))
         sys.exit(ERROR_NOT_IMPLEMENTED)
 
+    namespace = multi_asic.get_namespace_for_port(port_name)
+    config_db = ConfigDBConnector(use_unix_socket_path=True, namespace=namespace)
+    if config_db is not None:
+        config_db.connect()
+        subport = int(config_db.get(config_db.CONFIG_DB,
+                                    'PORT|{}'.format(port_name), 'subport'))
+
+        # If subport is not defined (None) or set to 0, assign a default value of 1
+        # to ensure valid subport configuration.
+        if subport is None or subport == 0:
+            subport = 1
+    else:
+        click.echo("Failed to connect to CONFIG_DB")
+        sys.exit(EXIT_FAIL)
+
+    state_db = SonicV2Connector(use_unix_socket_path=False, namespace=namespace)
+    if state_db is not None:
+        state_db.connect(state_db.STATE_DB)
+        host_lane_count = int(state_db.get(state_db.STATE_DB,
+                                           'TRANSCEIVER_INFO|{}'.format(port_name),
+                                           'host_lane_count'))
+        media_lane_count = int(state_db.get(state_db.STATE_DB,
+                                            'TRANSCEIVER_INFO|{}'.format(port_name),
+                                            'media_lane_count'))
+    else:
+        click.echo("Failed to connect to STATE_DB")
+        sys.exit(EXIT_FAIL)
+
+    if loopback_mode in ("host-side-input", "host-side-output",
+                         "host-side-input-none", "host-side-output-none"):
+        lane_mask = ((1 << host_lane_count) - 1) << ((subport - 1) * host_lane_count)
+    elif loopback_mode in ("media-side-input", "media-side-output",
+                           "media-side-input-none", "media-side-output-none"):
+        lane_mask = ((1 << media_lane_count) - 1) << ((subport - 1) * media_lane_count)
+    else:
+        lane_mask = 0
+
     try:
-        status = api.set_loopback_mode(loopback_mode)
+        status = api.set_loopback_mode(loopback_mode, lane_mask=lane_mask)
     except AttributeError:
         click.echo("{}: Set loopback mode is not applicable for this module".format(port_name))
         sys.exit(ERROR_NOT_IMPLEMENTED)
+    except TypeError:
+        status = api.set_loopback_mode(loopback_mode)
 
     if status:
         click.echo("{}: Set {} loopback".format(port_name, loopback_mode))

tests/sfputil_test.py

@@ -1633,6 +1633,9 @@ def test_load_port_config(self, mock_is_multi_asic):
     @patch('sfputil.main.platform_chassis')
     @patch('sfputil.main.platform_sfputil', MagicMock(is_logical_port=MagicMock(return_value=1)))
     @patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
+    @patch('sonic_py_common.multi_asic.get_front_end_namespaces', MagicMock(return_value=['']))
+    @patch('sfputil.main.ConfigDBConnector', MagicMock())
+    @patch('sfputil.main.SonicV2Connector', MagicMock())
     def test_debug_loopback(self, mock_chassis):
         mock_sfp = MagicMock()
         mock_api = MagicMock()
@@ -1659,6 +1662,12 @@ def test_debug_loopback(self, mock_chassis):
         assert result.output == 'Ethernet0: Set host-side-input loopback\n'
         assert result.exit_code != ERROR_NOT_IMPLEMENTED
 
+        mock_sfp.get_xcvr_api = MagicMock(return_value=mock_api)
+        result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+                               ["Ethernet0", "media-side-input"])
+        assert result.output == 'Ethernet0: Set media-side-input loopback\n'
+        assert result.exit_code != ERROR_NOT_IMPLEMENTED
+
         mock_api.set_loopback_mode.return_value = False
         result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
                                ["Ethernet0", "none"])
@@ -1671,3 +1680,8 @@ def test_debug_loopback(self, mock_chassis):
                                ["Ethernet0", "none"])
         assert result.output == 'Ethernet0: Set loopback mode is not applicable for this module\n'
         assert result.exit_code == ERROR_NOT_IMPLEMENTED
+
+        mock_api.set_loopback_mode.side_effect = [TypeError, True]
+        result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+                               ["Ethernet0", "none"])
+        assert result.output == 'Ethernet0: Set none loopback\n'