[sfputil] Add loopback sub-command for debugging and module diagnostics for branch 202311

doc/Command-Reference.md

@@ -43,6 +43,8 @@
   * [CMIS firmware version show commands](#cmis-firmware-version-show-commands)
   * [CMIS firmware upgrade commands](#cmis-firmware-upgrade-commands)
   * [CMIS firmware target mode commands](#cmis-firmware-target-mode-commands)
+* [CMIS debug](#cmis-debug)
+* [CMIS debug loopback](#cmis-debug-loopback)
 * [DHCP Relay](#dhcp-relay)
   * [DHCP Relay show commands](#dhcp-relay-show-commands)
   * [DHCP Relay clear commands](#dhcp-relay-clear-commands)
@@ -2891,6 +2893,31 @@ Example of the module supporting target mode
   Target Mode set to 1
   ```
 
+## CMIS debug
+
+### CMIS debug loopback
+
+This command is the standard CMIS diagnostic control used for troubleshooting link and performance issues between the host switch and transceiver module.
+
+**sfputil debug loopback**
+
+- Usage:
+  ```
+  sfputil debug loopback PORT_NAME LOOPBACK_MODE
+
+  Set the loopback mode
+  host-side-input: host side input loopback mode
+  host-side-output: host side output loopback mode
+  media-side-input: media side input loopback mode
+  media-side-output: media side output loopback mode
+  none: disable loopback mode
+  ```
+
+- Example:
+  ```
+  admin@sonic:~$ sfputil debug loopback Ethernet88 host-side-input
+  ```
+
 ## DHCP Relay
 
 ### DHCP Relay show commands

sfputil/main.py

@@ -1886,5 +1886,50 @@ def get_overall_offset_sff8472(api, page, offset, size, wire_addr):
         return page * PAGE_SIZE + offset + PAGE_SIZE_FOR_A0H
 
 
+# 'debug' subgroup
+@cli.group()
+def debug():
+    """Module debug and diagnostic control"""
+    pass
+
+
+# 'loopback' subcommand
+@debug.command()
+@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"]))
+def loopback(port_name, loopback_mode):
+    """Set module diagnostic loopback mode
+    """
+    physical_port = logical_port_to_physical_port_index(port_name)
+    sfp = platform_chassis.get_sfp(physical_port)
+
+    if is_port_type_rj45(port_name):
+        click.echo("{}: This functionality is not applicable for RJ45 port".format(port_name))
+        sys.exit(EXIT_FAIL)
+
+    if not is_sfp_present(port_name):
+        click.echo("{}: SFP EEPROM not detected".format(port_name))
+        sys.exit(EXIT_FAIL)
+
+    try:
+        api = sfp.get_xcvr_api()
+    except NotImplementedError:
+        click.echo("{}: This functionality is not implemented".format(port_name))
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+
+    try:
+        status = api.set_loopback_mode(loopback_mode)
+    except AttributeError:
+        click.echo("{}: Set loopback mode is not applicable for this module".format(port_name))
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+
+    if status:
+        click.echo("{}: Set {} loopback".format(port_name, loopback_mode))
+    else:
+        click.echo("{}: Set {} loopback failed".format(port_name, loopback_mode))
+        sys.exit(EXIT_FAIL)
+
 if __name__ == '__main__':
     cli()

tests/sfputil_test.py

@@ -1498,3 +1498,46 @@ def test_target_firmware(self, mock_chassis):
         result = runner.invoke(sfputil.cli.commands['firmware'].commands['target'], ["Ethernet0", "1"])
         assert result.output == 'Target Mode set failed!\n'
         assert result.exit_code == EXIT_FAIL
+
+    @patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=False))
+    @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))
+    def test_debug_loopback(self, mock_chassis):
+        mock_sfp = MagicMock()
+        mock_api = MagicMock()
+        mock_chassis.get_sfp = MagicMock(return_value=mock_sfp)
+        mock_sfp.get_presence.return_value = True
+        mock_sfp.get_xcvr_api = MagicMock(return_value=mock_api)
+
+        runner = CliRunner()
+        mock_sfp.get_presence.return_value = False
+        result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+                               ["Ethernet0", "host-side-input"])
+        assert result.output == 'Ethernet0: SFP EEPROM not detected\n'
+        mock_sfp.get_presence.return_value = True
+
+        mock_sfp.get_xcvr_api = MagicMock(side_effect=NotImplementedError)
+        result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+                               ["Ethernet0", "host-side-input"])
+        assert result.output == 'Ethernet0: This functionality is not implemented\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", "host-side-input"])
+        assert result.output == 'Ethernet0: Set host-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"])
+        assert result.output == 'Ethernet0: Set none loopback failed\n'
+        assert result.exit_code == EXIT_FAIL
+
+        mock_api.set_loopback_mode.return_value = True
+        mock_api.set_loopback_mode.side_effect = AttributeError
+        result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+                               ["Ethernet0", "none"])
+        assert result.output == 'Ethernet0: Set loopback mode is not applicable for this module\n'
+        assert result.exit_code == ERROR_NOT_IMPLEMENTED