[sfputil] Add loopback sub-command for debugging and module diagnosti… (

sonic-net#3369) * [sfputil] Add loopback sub-command for debugging and module diagnostic control Signed-off-by: xinyu <[email protected]> * [sfputil] Correct and update the reference of sfputil debug loopback command Signed-off-by: xinyu <[email protected]> --------- Signed-off-by: xinyu <[email protected]>
ayurkiv-nvda · Jul 12, 2024 · b6f7c2b · b6f7c2b
1 parent 1f94444
commit b6f7c2b
Show file tree

Hide file tree

Showing 3 changed files with 115 additions and 0 deletions.
diff --git a/doc/Command-Reference.md b/doc/Command-Reference.md
@@ -47,6 +47,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)
@@ -3094,6 +3096,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

diff --git a/sfputil/main.py b/sfputil/main.py
@@ -1888,5 +1888,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()
diff --git a/tests/sfputil_test.py b/tests/sfputil_test.py
@@ -1510,3 +1510,46 @@ def test_load_port_config(self, mock_is_multi_asic):
 
         mock_is_multi_asic.return_value = False
         assert sfputil.load_port_config() == True
+
+    @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