[SmartSwitch] Define DPU helper functions (#20724)

Why I did it
Add new DPU helper functions to avoid code duplication

How I did it
Define new is_dpu, get_num_dpus helpers.
Define get_platform_json_data and get_dpu_info helper function for common code and refactor code accordingly.

How to verify it
Verified the changes on smart switch

src/sonic-py-common/sonic_py_common/device_info.py

@@ -168,6 +168,34 @@ def get_platform_and_hwsku():
     return (platform, hwsku)
 
 
+def get_platform_json_data():
+    """
+    Retrieve the data from platform.json file
+
+    Returns:
+        A dictionary containing the key/value pairs as found in the platform.json file
+    """
+    platform = get_platform()
+    if not platform:
+        return None
+
+    platform_path = get_path_to_platform_dir()
+    if not platform_path:
+        return None
+
+    platform_json = os.path.join(platform_path, PLATFORM_JSON_FILE)
+    if not os.path.isfile(platform_json):
+        return None
+
+    try:
+        with open(platform_json, 'r') as f:
+            platform_data = json.loads(f.read())
+            return platform_data
+    except (json.JSONDecodeError, IOError, TypeError, ValueError):
+        # Handle any file reading and JSON parsing errors
+        return None
+
+
 def get_asic_conf_file_path():
     """
     Retrieves the path to the ASIC configuration file on the device
@@ -582,15 +610,30 @@ def is_smartswitch():
     if not platform:
         return False
 
-    # get platform.json file path
-    platform_json = os.path.join(HOST_DEVICE_PATH, platform, "platform.json")
-    try:
-        with open(platform_json, 'r') as f:
-            platform_cfg = json.loads(f.read())
-            return "DPUS" in platform_cfg
-    except IOError:
+    # Retrieve platform.json data
+    platform_data = get_platform_json_data()
+    if platform_data:
+        return "DPUS" in platform_data
+
+    return False
+
+
+def is_dpu():
+    # Get platform
+    platform = get_platform()
+    if not platform:
         return False
 
+    if not is_smartswitch():
+        return False
+
+    # Retrieve platform.json data
+    platform_data = get_platform_json_data()
+    if platform_data:
+        return 'DPU' in platform_data
+
+    return False
+
 
 def is_supervisor():
     platform_env_conf_file_path = get_platform_env_conf_file_path()
@@ -892,37 +935,63 @@ def is_frontend_port_present_in_host():
     return True
 
 
+def get_dpu_info():
+    """
+    Retrieves the DPU information from platform.json file.
+
+    Returns:
+        A dictionary containing the DPU information.
+    """
+
+    platform = get_platform()
+    if not platform:
+        return {}
+
+    # Retrieve platform.json data
+    platform_data = get_platform_json_data()
+    if not platform_data:
+        return {}
+
+    if "DPUS" in platform_data:
+        return platform_data["DPUS"]
+    elif 'DPU' in platform_data:
+        return platform_data['DPU']
+    else:
+        return {}
+
+
 def get_num_dpus():
     """
     Retrieves the number of DPUs from platform.json file.
 
-    Args:
-
     Returns:
         A integer to indicate the number of DPUs.
     """
 
-    platform = get_platform()
-    if not platform:
+    if is_dpu():
         return 0
 
-    # Get Platform path.
-    platform_path = get_path_to_platform_dir()
+    dpu_info = get_dpu_info()
+    if dpu_info is not None and len(dpu_info) > 0:
+        return len(dpu_info)
 
-    if os.path.isfile(os.path.join(platform_path, PLATFORM_JSON_FILE)):
-        json_file = os.path.join(platform_path, PLATFORM_JSON_FILE)
+    return 0
 
-        try:
-            with open(json_file, 'r') as file:
-                platform_data = json.load(file)
-        except (json.JSONDecodeError, IOError, TypeError, ValueError):
-            # Handle any file reading and JSON parsing errors
-            return 0
-
-        # Convert to lower case avoid case sensitive.
-        data = {k.lower(): v for k, v in platform_data.items()}
-        DPUs = data.get('dpus', None)
-        if DPUs is not None and len(DPUs) > 0:
-            return len(DPUs)
 
-    return 0
+def get_dpu_list():
+    """
+    Retrieves the list of DPUs from platform.json file.
+
+    Returns:
+        A list indicating the list of DPUs.
+        For example, ['dpu0', 'dpu1', 'dpu2']
+    """
+
+    if is_dpu():
+        return []
+
+    dpu_info = get_dpu_info()
+    if dpu_info is not None and len(dpu_info) > 0:
+        return list(dpu_info)
+
+    return []

src/sonic-py-common/tests/device_info_test.py

@@ -10,6 +10,7 @@
     import mock
 
 import pytest
+import json
 
 from sonic_py_common import device_info
 
@@ -162,6 +163,176 @@ def test_get_platform_info(self, mock_hwsku, mock_machine_info, mock_sonic_ver,
         assert mock_hwsku.called_once()
         mock_cfg_inst.get_table.assert_called_once_with("DEVICE_METADATA")
 
+    @mock.patch("os.path.isfile")
+    @mock.patch("{}.open".format(BUILTINS))
+    @mock.patch("sonic_py_common.device_info.get_path_to_platform_dir")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_get_platform_json_data(self, mock_get_platform, mock_get_path_to_platform_dir, mock_open, mock_isfile):
+        mock_get_platform.return_value = "x86_64-mlnx_msn2700-r0"
+        mock_get_path_to_platform_dir.return_value = "/usr/share/sonic/device"
+        mock_isfile.return_value = True
+        platform_json_data = {
+            "chassis": {
+                "name": "MSN2700"
+            }
+        }
+        open_mocked = mock.mock_open(read_data=json.dumps(platform_json_data))
+        mock_open.side_effect = open_mocked
+
+        result = device_info.get_platform_json_data()
+        assert result == platform_json_data
+        assert mock_open.call_count == 1
+
+        # Test case where platform is None
+        mock_get_platform.return_value = None
+        result = device_info.get_platform_json_data()
+        assert result is None
+
+        # Test case where platform path is None
+        mock_get_platform.return_value = "x86_64-mlnx_msn2700-r0"
+        mock_get_path_to_platform_dir.return_value = None
+        result = device_info.get_platform_json_data()
+        assert result is None
+
+        # Test case where platform.json file does not exist
+        mock_get_path_to_platform_dir.return_value = "/usr/share/sonic/device"
+        mock_isfile.return_value = False
+        result = device_info.get_platform_json_data()
+        assert result is None
+
+        # Test case where JSON decoding fails
+        mock_isfile.return_value = True
+        open_mocked = mock.mock_open(read_data="invalid json")
+        mock_open.side_effect = open_mocked
+        result = device_info.get_platform_json_data()
+        assert result is None
+
+    @mock.patch("sonic_py_common.device_info.get_platform_json_data")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_is_smartswitch(self, mock_get_platform, mock_get_platform_json_data):
+        # Test case where platform is None
+        mock_get_platform.return_value = None
+        assert device_info.is_smartswitch() == False
+
+        # Test case where platform.json data is None
+        mock_get_platform.return_value="x86_64-mlnx_msn2700-r0"
+        mock_get_platform_json_data.return_value=None
+        assert device_info.is_smartswitch() == False
+
+        # Test case where platform.json data does not contain "DPUS"
+        mock_get_platform_json_data.return_value={}
+        assert device_info.is_smartswitch() == False
+
+        # Test case where platform.json data contains "DPUS"
+        mock_get_platform_json_data.return_value={"DPUS": {}}
+        assert device_info.is_smartswitch() == True
+
+    @mock.patch("sonic_py_common.device_info.get_platform_json_data")
+    @mock.patch("sonic_py_common.device_info.is_smartswitch")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_is_dpu(self, mock_get_platform, mock_is_smartswitch, mock_get_platform_json_data):
+        # Test case where platform is None
+        mock_get_platform.return_value=None
+        assert device_info.is_dpu() == False
+
+        # Test case where platform is not a smart switch
+        mock_get_platform.return_value="x86_64-mlnx_msn2700-r0"
+        mock_is_smartswitch.return_value=False
+        assert device_info.is_dpu() == False
+
+        # Test case where platform is a smart switch but no DPU data in platform.json
+        mock_is_smartswitch.return_value=True
+        mock_get_platform_json_data.return_value={}
+        assert device_info.is_dpu() == False
+
+        # Test case where platform is a smart switch and DPU data is present in platform.json
+        mock_get_platform_json_data.return_value={"DPU": {}}
+        assert device_info.is_dpu() == True
+
+        # Test case where platform is a smart switch and DPU data is present in platform.json
+        mock_get_platform_json_data.return_value={"DPUS": {}}
+        assert device_info.is_dpu() == False
+
+    @mock.patch("sonic_py_common.device_info.get_platform_json_data")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_get_dpu_info(self, mock_get_platform, mock_get_platform_json_data):
+        # Test case where platform is None
+        mock_get_platform.return_value = None
+        assert device_info.get_dpu_info() == {}
+
+        # Test case where platform.json data is None
+        mock_get_platform.return_value = "x86_64-mlnx_msn2700-r0"
+        mock_get_platform_json_data.return_value = None
+        assert device_info.get_dpu_info() == {}
+
+        # Test case where platform.json data does not contain "DPUS" or "DPU"
+        mock_get_platform_json_data.return_value = {}
+        assert device_info.get_dpu_info() == {}
+
+        # Test case where platform.json data contains "DPUS"
+        mock_get_platform_json_data.return_value = {"DPUS": {"dpu0": {}, "dpu1": {}}}
+        assert device_info.get_dpu_info() == {"dpu0": {}, "dpu1": {}}
+
+        # Test case where platform.json data contains "DPU"
+        mock_get_platform_json_data.return_value = {"DPU": {"dpu0": {}}}
+        assert device_info.get_dpu_info() == {"dpu0": {}}
+
+        # Test case where platform.json data does not contain "DPU" or "DPUS"
+        mock_get_platform_json_data.return_value = {"chassis": {}}
+        assert device_info.get_dpu_info() == {}
+
+    @mock.patch("sonic_py_common.device_info.get_platform_json_data")
+    @mock.patch("sonic_py_common.device_info.is_dpu")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_get_num_dpus(self, mock_get_platform, mock_is_dpu, mock_get_platform_json_data):
+        # Test case where platform is None
+        mock_get_platform.return_value = None
+        assert device_info.get_num_dpus() == 0
+
+        # Test case where platform is a DPU
+        mock_get_platform.return_value = "x86_64-mlnx_msn2700-r0"
+        mock_is_dpu.return_value = True
+        assert device_info.get_num_dpus() == 0
+
+        # Test case where platform.json data is None
+        mock_is_dpu.return_value = False
+        mock_get_platform_json_data.return_value = None
+        assert device_info.get_num_dpus() == 0
+
+        # Test case where platform.json data does not contain DPUs
+        mock_get_platform_json_data.return_value = {}
+        assert device_info.get_num_dpus() == 0
+
+        # Test case where platform.json data contains DPUs
+        mock_get_platform_json_data.return_value = {"DPUS": {"dpu0": {}, "dpu1": {}}}
+        assert device_info.get_num_dpus() == 2
+
+    @mock.patch("sonic_py_common.device_info.get_platform_json_data")
+    @mock.patch("sonic_py_common.device_info.is_dpu")
+    @mock.patch("sonic_py_common.device_info.get_platform")
+    def test_get_dpu_list(self, mock_get_platform, mock_is_dpu, mock_get_platform_json_data):
+        # Test case where platform is None
+        mock_get_platform.return_value = None
+        assert device_info.get_dpu_list() == []
+
+        # Test case where platform is a DPU
+        mock_get_platform.return_value = "x86_64-mlnx_msn2700-r0"
+        mock_is_dpu.return_value = True
+        assert device_info.get_dpu_list() == []
+
+        # Test case where platform.json data is None
+        mock_is_dpu.return_value = False
+        mock_get_platform_json_data.return_value = None
+        assert device_info.get_dpu_list() == []
+
+        # Test case where platform.json data does not contain DPUs
+        mock_get_platform_json_data.return_value = {}
+        assert device_info.get_dpu_list() == []
+
+        # Test case where platform.json data contains DPUs
+        mock_get_platform_json_data.return_value = {"DPUS": {"dpu0": {}, "dpu1": {}}}
+        assert device_info.get_dpu_list() == ["dpu0", "dpu1"]
+
     @classmethod
     def teardown_class(cls):
         print("TEARDOWN")