[sfputil] add support for sfputil debug tx-output/rx-output {port} enable/disable (sonic-net#162)

<!--
 Please make sure you've read and understood our contributing guidelines:
 https://github.com/Azure/SONiC/blob/gh-pages/CONTRIBUTING.md

 failure_prs.log skip_prs.log Make sure all your commits include a signature generated with `git commit -s` **

 If this is a bug fix, make sure your description includes "closes #xxxx",
 "fixes #xxxx" or "resolves #xxxx" so that GitHub automatically closes the related
 issue when the PR is merged.

 If you are adding/modifying/removing any command or utility script, please also
 make sure to add/modify/remove any unit tests from the tests
 directory as appropriate.

 If you are modifying or removing an existing 'show', 'config' or 'sonic-clear'
 subcommand, or you are adding a new subcommand, please make sure you also
 update the Command Line Reference Guide (doc/Command-Reference.md) to reflect
 your changes.

 Please provide the following information:
-->

## **Summary of Changes**

### **What I did**
- Added a new `debug.py` module to handle **SFP diagnostics and debugging**.
- Replaced existing **utility methods** with functions from `utilities_common.platform_sfputil_helper`.
- Registered `debug` as a subcommand in `sfputil/main.py`.

### **How I did it**
- Created a new `debug.py` file under `sfputil/`.
- Implemented the `debug` command group with the following subcommands:
 - `loopback`: Configures SFP loopback modes.
 - `tx-output`: Enables or disables TX output.
 - `rx-output`: Enables or disables RX output.
- Used helper functions from `utilities_common.platform_sfputil_helper` for better modularity and maintainability.
- Updated `sfputil/main.py` to integrate the new `debug` command.

### **How to verify it**
UT and deploy changed on a testbed
1. Run the following command to check if `debug` is now available:
 ```sh
 sfputil debug --help
2. sfputil debug tx-output Ethernet0 enable
3. sfputil debug rx-output Ethernet0 disable

```
Usage: sfputil debug [OPTIONS] COMMAND [ARGS]...
Try "sfputil debug --help" for help.

Error: no such option: -h
admin@sonic:~$ sudo sfputil debug --help
Usage: sfputil debug [OPTIONS] COMMAND [ARGS]...

 Group for debugging and diagnostic control commands.

 This command group loads platform-specific utilities and prepares them for
 use in diagnostic commands.

Options:
 --help Show this message and exit.

Commands:
 loopback Set module diagnostic loopback mode.
 rx-output Enable or disable RX output on a port.
 tx-output Enable or disable TX output on a port.
admin@sonic:~$ sudo sfputil debug rx-output --help
Usage: sfputil debug rx-output [OPTIONS] PORT_NAME [enable|disable]

 Enable or disable RX output on a port.

Options:
 --help Show this message and exit.
admin@sonic:~$
```

Author: mssonicbld

doc/Command-Reference.md

@@ -50,6 +50,8 @@
   * [CMIS firmware target mode commands](#cmis-firmware-target-mode-commands)
 * [CMIS debug](#cmis-debug)
 * [CMIS debug loopback](#cmis-debug-loopback)
+* [CMIS debug rx-output](#cmis-debug-rx-output)
+* [CMIS debug tx-output](#cmis-debug-tx-output)
 * [DHCP Relay](#dhcp-relay)
   * [DHCP Relay show commands](#dhcp-relay-show-commands)
   * [DHCP Relay clear commands](#dhcp-relay-clear-commands)
@@ -3246,6 +3248,41 @@ This command is the standard CMIS diagnostic control used for troubleshooting li
   admin@sonic:~$ sfputil debug loopback Ethernet88 media-side-output disable
   ```
 
+### CMIS debug rx-output
+
+The command disables RX input by muting the optical receiver on the module, preventing it from detecting incoming signals.
+
+**sfputil debug rx-ouput**
+
+- Usage:
+  ```
+  sfputil debug rx-output PORT_NAME <enable/disable>
+
+  ```
+
+- Example:
+  ```
+  admin@sonic:~$ sfputil debug rx-output Ethernet88 enable
+  admin@sonic:~$ sfputil debug rx-output Ethernet88 disable
+  ```
+
+### CMIS debug tx-output
+
+The command disables TX output by turning off the laser on the module, effectively blocking the optical signal.
+
+**sfputil debug tx-ouput**
+
+- Usage:
+  ```
+  sfputil debug tx-output PORT_NAME <enable/disable>
+
+  ```
+
+- Example:
+  ```
+  admin@sonic:~$ sfputil debug tx-output Ethernet88 enable
+  ```
+
 ## DHCP Relay
 
 ### DHCP Relay show commands

sfputil/debug.py

@@ -0,0 +1,124 @@
+import sys
+import click
+import utilities_common.cli as clicommon
+from utilities_common import platform_sfputil_helper
+from utilities_common.platform_sfputil_helper import (
+    get_subport,
+    get_sfp_object,
+    get_subport_lane_mask,
+    get_media_lane_count,
+    get_host_lane_count,
+)
+
+EXIT_FAIL = -1
+EXIT_SUCCESS = 0
+ERROR_PERMISSIONS = 1
+ERROR_CHASSIS_LOAD = 2
+ERROR_SFPUTILHELPER_LOAD = 3
+ERROR_PORT_CONFIG_LOAD = 4
+ERROR_NOT_IMPLEMENTED = 5
+ERROR_INVALID_PORT = 6
+
+
+@click.group(cls=clicommon.AliasedGroup)
+def debug():
+    """
+    Group for debugging and diagnostic control commands.
+
+    This command group loads platform-specific utilities and prepares them for use in diagnostic commands.
+    """
+    platform_sfputil_helper.load_platform_sfputil()
+    platform_sfputil_helper.load_chassis()
+    platform_sfputil_helper.platform_sfputil_read_porttab_mappings()
+
+
+@debug.command()
+@click.argument('port_name', required=True)
+@click.argument(
+    'loopback_mode',
+    required=True,
+    type=click.Choice(["host-side-input", "host-side-output", "media-side-input", "media-side-output"])
+)
+@click.argument('enable', required=True, type=click.Choice(["enable", "disable"]))
+def loopback(port_name, loopback_mode, enable):
+    """
+    Set module diagnostic loopback mode.
+    """
+    sfp = get_sfp_object(port_name)
+
+    try:
+        api = sfp.get_xcvr_api()
+    except NotImplementedError:
+        click.echo(f"{port_name}: This functionality is not implemented")
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+
+    subport = get_subport(port_name)
+
+    host_lane_count = get_host_lane_count(port_name)
+
+    media_lane_count = get_media_lane_count(port_name)
+
+    lane_count = int(host_lane_count) if 'host-side' in loopback_mode else int(media_lane_count)
+    lane_mask = get_subport_lane_mask(int(subport), lane_count)
+
+    try:
+        status = api.set_loopback_mode(loopback_mode, lane_mask=lane_mask, enable=(enable == 'enable'))
+    except AttributeError:
+        click.echo(f"{port_name}: Set loopback mode is not applicable for this module")
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+    except TypeError:
+        click.echo(f"{port_name}: Set loopback mode failed. Parameter is not supported")
+        sys.exit(EXIT_FAIL)
+
+    if status:
+        click.echo(f"{port_name}: {enable} {loopback_mode} loopback")
+    else:
+        click.echo(f"{port_name}: {enable} {loopback_mode} loopback failed")
+        sys.exit(EXIT_FAIL)
+
+
+def set_output(port_name, enable, direction):
+    """
+    Enable or disable TX/RX output based on direction ('tx' or 'rx').
+    """
+    sfp = get_sfp_object(port_name)
+
+    subport = get_subport(port_name)
+
+    media_lane_count = get_media_lane_count(port_name)
+
+    lane_mask = get_subport_lane_mask(int(subport), int(media_lane_count))
+
+    try:
+        if direction == "tx":
+            sfp.tx_disable_channel(lane_mask, enable == "disable")
+        elif direction == "rx":
+            sfp.rx_disable_channel(lane_mask, enable == "disable")
+
+        click.echo(
+            f"{port_name}: {direction.upper()} output "
+            f"{'disabled' if enable == 'disable' else 'enabled'} on subport {subport}"
+        )
+
+    except AttributeError:
+        click.echo(f"{port_name}: {direction.upper()} disable is not applicable for this module")
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+    except Exception as e:
+        click.echo(f"{port_name}: {direction.upper()} disable failed due to {str(e)}")
+        sys.exit(EXIT_FAIL)
+
+
+@debug.command()
+@click.argument('port_name', required=True)
+@click.argument('enable', required=True, type=click.Choice(["enable", "disable"]))
+def tx_output(port_name, enable):
+    """Enable or disable TX output on a port."""
+    set_output(port_name, enable, "tx")
+
+
+@debug.command()
+@click.argument('port_name', required=True)
+@click.argument('enable', required=True, type=click.Choice(["enable", "disable"]))
+def rx_output(port_name, enable):
+    """Enable or disable RX output on a port."""
+    set_output(port_name, enable, "rx")

sfputil/main.py

@@ -17,6 +17,7 @@
 import click
 import sonic_platform
 import sonic_platform_base.sonic_sfp.sfputilhelper
+from sfputil.debug import debug
 from sonic_platform_base.sfp_base import SfpBase
 from swsscommon.swsscommon import SonicV2Connector, ConfigDBConnector
 from natsort import natsorted
@@ -605,6 +606,8 @@ def cli():
         sys.exit(ERROR_PORT_CONFIG_LOAD)
 
 
+cli.add_command(debug)
+
 # 'show' subgroup
 @cli.group()
 def show():
@@ -1965,116 +1968,5 @@ 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)
-@click.argument('loopback_mode', required=True,
-                type=click.Choice(["host-side-input", "host-side-output",
-                                   "media-side-input", "media-side-output"]))
-@click.argument('enable', required=True, type=click.Choice(["enable", "disable"]))
-def loopback(port_name, loopback_mode, enable):
-    """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)
-
-    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()
-        try:
-            subport = int(config_db.get(config_db.CONFIG_DB, f'PORT|{port_name}', 'subport'))
-        except TypeError:
-            click.echo(f"{port_name}: subport is not present in CONFIG_DB")
-            sys.exit(EXIT_FAIL)
-
-        # If subport is set to 0, assign a default value of 1 to ensure valid subport configuration
-        if subport == 0:
-            subport = 1
-    else:
-        click.echo(f"{port_name}: 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)
-        try:
-            host_lane_count = int(state_db.get(state_db.STATE_DB,
-                                               f'TRANSCEIVER_INFO|{port_name}',
-                                               'host_lane_count'))
-        except TypeError:
-            click.echo(f"{port_name}: host_lane_count is not present in STATE_DB")
-            sys.exit(EXIT_FAIL)
-
-        try:
-            media_lane_count = int(state_db.get(state_db.STATE_DB,
-                                                f'TRANSCEIVER_INFO|{port_name}',
-                                                'media_lane_count'))
-        except TypeError:
-            click.echo(f"{port_name}: media_lane_count is not present in STATE_DB")
-            sys.exit(EXIT_FAIL)
-    else:
-        click.echo(f"{port_name}: Failed to connect to STATE_DB")
-        sys.exit(EXIT_FAIL)
-
-    if 'host-side' in loopback_mode:
-        lane_mask = get_subport_lane_mask(subport, host_lane_count)
-    elif 'media-side' in loopback_mode:
-        lane_mask = get_subport_lane_mask(subport, media_lane_count)
-    else:
-        lane_mask = 0
-
-    try:
-        status = api.set_loopback_mode(loopback_mode,
-                                       lane_mask=lane_mask,
-                                       enable=enable == 'enable')
-    except AttributeError:
-        click.echo("{}: Set loopback mode is not applicable for this module".format(port_name))
-        sys.exit(ERROR_NOT_IMPLEMENTED)
-    except TypeError:
-        click.echo("{}: Set loopback mode failed. Parameter is not supported".format(port_name))
-        sys.exit(EXIT_FAIL)
-
-    if status:
-        click.echo("{}: {} {} loopback".format(port_name, enable, loopback_mode))
-    else:
-        click.echo("{}: {} {} loopback failed".format(port_name, enable, loopback_mode))
-        sys.exit(EXIT_FAIL)
-
-
-def get_subport_lane_mask(subport, lane_count):
-    """Get the lane mask for the given subport and lane count
-
-    Args:
-        subport (int): Subport number
-        lane_count (int): Lane count for the subport
-
-    Returns:
-        int: Lane mask for the given subport and lane count
-    """
-    return ((1 << lane_count) - 1) << ((subport - 1) * lane_count)
-
-
 if __name__ == '__main__':
     cli()