Add sfputil power enable/disable command

config/main.py

@@ -5537,6 +5537,32 @@ def dom(ctx, interface_name, desired_config):
     else:
         config_db.mod_entry("PORT", interface_name, {"dom_polling": "disabled" if desired_config == "disable" else "enabled"})
 
+#
+# 'power' subcommand ('config interface power ... ')
+#
+
+
+@transceiver.command()
+@click.argument('interface_name', metavar='<interface_name>', required=True)
+@click.argument('mode', metavar='<mode>', required=True, type=click.Choice(["enable", "disable"]))
+@click.pass_context
+def power(ctx, interface_name, mode):
+    """enable/disable power for SFP transceiver module"""
+    log.log_info("interface transceiver power {} {} executing...".format(interface_name, mode))
+    # Get the config_db connector
+    config_db = ctx.obj['config_db']
+
+    if clicommon.get_interface_naming_mode() == "alias":
+        interface_name = interface_alias_to_name(config_db, interface_name)
+        if interface_name is None:
+            ctx.fail("'interface_name' is None!")
+
+    if interface_name_is_valid(config_db, interface_name) is False:
+        ctx.fail("Interface name is invalid. Please enter a valid interface name!!")
+
+    cmd = ['sudo', 'sfputil', 'power', str(mode), str(interface_name)]
+    clicommon.run_command(cmd)
+
 #
 # 'mpls' subgroup ('config interface mpls ...')
 #

doc/Command-Reference.md

@@ -5552,6 +5552,22 @@ This command is used to configure the Digital Optical Monitoring (DOM) for an in
   user@sonic~$ sudo config interface transceiver dom Ethernet0 disable
   ```
 
+**config interface transceiver power**
+
+This command is used to enable or disable power to the module of an interface.
+
+- Usage:
+  ```
+  config interface transceiver power <interface_name> (enable | disable)
+  ```
+
+- Examples:
+  ```
+  user@sonic~$ sudo config interface transceiver power Ethernet0 enable
+
+  user@sonic~$ sudo config interface transceiver power Ethernet0 disable
+  ```
+
 **config interface mtu <interface_name> (Versions >= 201904)**
 
 This command is used to configure the mtu for the Physical interface. Use the value 1500 for setting max transfer unit size to 1500 bytes.

sfputil/main.py

@@ -1306,6 +1306,61 @@ def reset(port_name):
 
         i += 1
 
+
+# 'power' subgroup
+@cli.group()
+def power():
+    """Enable or disable power of SFP transceiver"""
+    pass
+
+
+# Helper method for setting low-power mode
+def set_power(port_name, enable):
+    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("Power disable/enable is not available for RJ45 port {}.".format(port_name))
+        sys.exit(EXIT_FAIL)
+
+    try:
+        presence = sfp.get_presence()
+    except NotImplementedError:
+        click.echo("sfp get_presence() NOT implemented!")
+        sys.exit(EXIT_FAIL)
+
+    if not presence:
+        click.echo("{}: SFP EEPROM not detected\n".format(port_name))
+        sys.exit(EXIT_FAIL)
+
+    try:
+        result = platform_chassis.get_sfp(physical_port).set_power(enable)
+    except (NotImplementedError, AttributeError):
+        click.echo("This functionality is currently not implemented for this platform")
+        sys.exit(ERROR_NOT_IMPLEMENTED)
+
+    if result:
+        click.echo("OK")
+    else:
+        click.echo("Failed")
+
+
+# 'disable' subcommand
+@power.command()
+@click.argument('port_name', metavar='<port_name>')
+def disable(port_name):
+    """Disable power of SFP transceiver"""
+    set_power(port_name, False)
+
+
+# 'enable' subcommand
+@power.command()
+@click.argument('port_name', metavar='<port_name>')
+def enable(port_name):
+    """Enable power of SFP transceiver"""
+    set_power(port_name, True)
+
+
 def update_firmware_info_to_state_db(port_name):
     physical_port = logical_port_to_physical_port_index(port_name)
 

tests/config_xcvr_test.py

@@ -67,6 +67,16 @@ def test_dom(self, mock_get_entry, ctx):
         mock_get_entry.return_value = {'subport': '1'}
         result = self.basic_check("dom", [interface_name, desired_config], ctx)
 
+    def test_power(self, ctx):
+        interface_name = 'Ethernet0'
+        desired_config = 'enable'
+
+        result = self.basic_check("power", ["", desired_config], ctx, operator.ne)
+        assert "Interface name is invalid. Please enter a valid interface name!!" in result.output
+
+        desired_config = 'disable'
+        result = self.basic_check("power", [interface_name, desired_config], ctx, operator.ne)
+
     def basic_check(self, command_name, para_list, ctx, op=operator.eq, expect_result=0):
         runner = CliRunner()
         result = runner.invoke(config.config.commands["interface"].commands["transceiver"].commands[command_name], para_list, obj = ctx)

tests/sfputil_test.py

@@ -570,6 +570,50 @@ def test_show_lpmode(self, mock_chassis):
 """
         assert result.output == expected_output
 
+    @patch('sfputil.main.platform_chassis')
+    @patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
+    @patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=True))
+    def test_power_RJ45(self, mock_chassis):
+        mock_sfp = MagicMock()
+        mock_api = MagicMock()
+        mock_sfp.get_xcvr_api = MagicMock(return_value=mock_api)
+        mock_sfp.get_presence.return_value = True
+        mock_chassis.get_sfp = MagicMock(return_value=mock_sfp)
+        runner = CliRunner()
+        result = runner.invoke(sfputil.cli.commands['power'].commands['enable'], ["Ethernet0"])
+        assert result.output == 'Power disable/enable is not available for RJ45 port Ethernet0.\n'
+        assert result.exit_code == EXIT_FAIL
+
+    @patch('sfputil.main.platform_chassis')
+    @patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
+    @patch('sfputil.main.platform_sfputil', MagicMock(is_logical_port=MagicMock(return_value=1)))
+    @patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=False))
+    def test_power(self, mock_chassis):
+        mock_sfp = MagicMock()
+        mock_api = MagicMock()
+        mock_sfp.get_xcvr_api = MagicMock(return_value=mock_api)
+        mock_chassis.get_sfp = MagicMock(return_value=mock_sfp)
+        mock_sfp.get_presence.return_value = True
+        runner = CliRunner()
+        result = runner.invoke(sfputil.cli.commands['power'].commands['enable'], ["Ethernet0"])
+        assert result.exit_code == 0
+
+        mock_sfp.get_presence.return_value = False
+        result = runner.invoke(sfputil.cli.commands['power'].commands['enable'], ["Ethernet0"])
+        assert result.output == 'Ethernet0: SFP EEPROM not detected\n\n'
+
+        mock_sfp.get_presence.return_value = True
+        mock_sfp.set_power = MagicMock(side_effect=NotImplementedError)
+        runner = CliRunner()
+        result = runner.invoke(sfputil.cli.commands['power'].commands['enable'], ["Ethernet0"])
+        assert result.output == 'This functionality is currently not implemented for this platform\n'
+        assert result.exit_code == ERROR_NOT_IMPLEMENTED
+
+        mock_sfp.set_power = MagicMock(return_value=False)
+        runner = CliRunner()
+        result = runner.invoke(sfputil.cli.commands['power'].commands['enable'], ["Ethernet0"])
+        assert result.output == 'Failed\n'
+
     @patch('sfputil.main.platform_chassis')
     @patch('sfputil.main.logical_port_to_physical_port_index', MagicMock(return_value=1))
     @patch('sfputil.main.logical_port_name_to_physical_port_list', MagicMock(return_value=[1]))