[sfputil] Add generic function to get lane mask by given subport and …

…lane count Signed-off-by: xinyu <xinyu0123@gmail.com>
sonic-net · Sep 10, 2024 · 2aaf92f · 2aaf92f
1 parent 4ede797
commit 2aaf92f
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 7 deletions.
diff --git a/sfputil/main.py b/sfputil/main.py
@@ -2032,9 +2032,9 @@ def loopback(port_name, loopback_mode, enable):
         sys.exit(EXIT_FAIL)
 
     if 'host-side' in loopback_mode:
-        lane_mask = ((1 << host_lane_count) - 1) << ((subport - 1) * host_lane_count)
+        lane_mask = get_subport_lane_mask(subport, host_lane_count)
     elif 'media-side' in loopback_mode:
-        lane_mask = ((1 << media_lane_count) - 1) << ((subport - 1) * media_lane_count)
+        lane_mask = get_subport_lane_mask(subport, media_lane_count)
     else:
         lane_mask = 0
 
@@ -2050,10 +2050,24 @@ def loopback(port_name, loopback_mode, enable):
         sys.exit(EXIT_FAIL)
 
     if status:
-        click.echo("{}: Set {} loopback".format(port_name, loopback_mode))
+        click.echo("{}: {} {} loopback".format(port_name, enable, loopback_mode))
     else:
-        click.echo("{}: Set {} loopback failed".format(port_name, loopback_mode))
+        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()
diff --git a/tests/sfputil_test.py b/tests/sfputil_test.py
@@ -1661,19 +1661,19 @@ def test_debug_loopback(self, mock_sonic_v2_connector, mock_config_db_connector,
         mock_sfp.get_xcvr_api = MagicMock(return_value=mock_api)
         result = runner.invoke(sfputil.cli.commands['debug'].commands['loopback'],
                                ["Ethernet0", "host-side-input", "enable"])
-        assert result.output == 'Ethernet0: Set host-side-input loopback\n'
+        assert result.output == 'Ethernet0: enable host-side-input loopback\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", "media-side-input", "enable"])
-        assert result.output == 'Ethernet0: Set media-side-input loopback\n'
+        assert result.output == 'Ethernet0: enable media-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", "media-side-output", "enable"])
-        assert result.output == 'Ethernet0: Set media-side-output loopback failed\n'
+        assert result.output == 'Ethernet0: enable media-side-output loopback failed\n'
         assert result.exit_code == EXIT_FAIL
 
         mock_api.set_loopback_mode.return_value = True
@@ -1709,3 +1709,14 @@ def test_debug_loopback(self, mock_sonic_v2_connector, mock_config_db_connector,
                                ["Ethernet0", "media-side-input", "enable"])
         assert result.output == 'Ethernet0: Failed to connect to STATE_DB\n'
         assert result.exit_code == EXIT_FAIL
+
+    @pytest.mark.parametrize("subport, lane_count, expected_mask", [
+        (1, 1, 0x1),
+        (1, 4, 0xf),
+        (2, 1, 0x2),
+        (2, 4, 0xf0),
+        (3, 2, 0x30),
+        (4, 1, 0x8),
+    ])
+    def test_get_subport_lane_mask(self, subport, lane_count, expected_mask):
+        assert sfputil.get_subport_lane_mask(subport, lane_count) == expected_mask