[sfputil] Reject the loopback command and report an error if subport,…

… host_lane_count, or media_lane_count is not present

Signed-off-by: xinyu <xinyu0123@gmail.com>

doc/Command-Reference.md

@@ -3109,11 +3109,15 @@ This command is the standard CMIS diagnostic control used for troubleshooting li
  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
+ host-side-input: enable host side input loopback mode
+ host-side-output: enable host side output loopback mode
+ media-side-input: enable media side input loopback mode
+ media-side-output: enable media side output loopback mode
+ host-side-input-none: disable host side input loopback mode
+ host-side-output-none: disable host side output loopback mode
+ media-side-input-none: disable media side input loopback mode
+ media-side-output-none: disable media side output loopback mode
+ none: disable all loopback mode
  ```
 
 - Example:

sfputil/main.py

@@ -1997,28 +1997,39 @@ def loopback(port_name, loopback_mode):
  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'))
+ 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 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:
+ # 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("Failed to connect to CONFIG_DB")
+ 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)
- 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'))
+ 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("Failed to connect to STATE_DB")
+ click.echo(f"{port_name}: Failed to connect to STATE_DB")
  sys.exit(EXIT_FAIL)
 
  if 'host-side' in loopback_mode:

tests/sfputil_test.py

@@ -1631,14 +1631,16 @@ def test_load_port_config(self, mock_is_multi_asic):
 
  @patch('sfputil.main.is_port_type_rj45', MagicMock(return_value=False))
  @patch('sfputil.main.platform_chassis')
+ @patch('sfputil.main.ConfigDBConnector')
+ @patch('sfputil.main.SonicV2Connector')
  @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):
+ def test_debug_loopback(self, mock_sonic_v2_connector, mock_config_db_connector, mock_chassis):
  mock_sfp = MagicMock()
  mock_api = MagicMock()
+ mock_config_db_connector.return_value = MagicMock()
+ mock_sonic_v2_connector.return_value = 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)
@@ -1685,3 +1687,24 @@ def test_debug_loopback(self, mock_chassis):
  result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
  ["Ethernet0", "none"])
  assert result.output == 'Ethernet0: Set none loopback\n'
+
+ mock_config_db = MagicMock()
+ mock_config_db.get.side_effect = TypeError
+ mock_config_db_connector.return_value = mock_config_db
+ result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+ ["Ethernet0", "none"])
+ assert result.output == 'Ethernet0: subport is not present in CONFIG_DB\n'
+ assert result.exit_code == EXIT_FAIL
+
+ mock_config_db_connector.return_value = None
+ result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+ ["Ethernet0", "none"])
+ assert result.output == 'Ethernet0: Failed to connect to CONFIG_DB\n'
+ assert result.exit_code == EXIT_FAIL
+
+ mock_config_db_connector.return_value = MagicMock()
+ mock_sonic_v2_connector.return_value = None
+ result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
+ ["Ethernet0", "none"])
+ assert result.output == 'Ethernet0: Failed to connect to STATE_DB\n'
+ assert result.exit_code == EXIT_FAIL