From 0c00b59af9c9ce7220ac39317b0441266867b9ab Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:11:52 +0530 Subject: [PATCH 1/2] feature: Add RAS module support to L0 Sysman python binding Added following RAS APIs to python binding: 1. zesDeviceEnumRasErrorSets 2. zesRasGetProperties 3. zesRasGetSupportedCategoriesExp 4. zesRasGetStateExp2 5. zesRasGetConfigExp 6. zesRasSetConfigExp 7. zesRasClearStateExp Along with the required handle, enums, structure types and structures (zes_ras_properties_t, zes_ras_state_exp2_t, zes_ras_config_exp_t), unit tests, a RAS section in the black box test and README updates. Related-To: SYSM-429 Signed-off-by: Aviral Nigam --- bindings/sysman/python/README.md | 8 + .../source/examples/pyzes_black_box_test.py | 227 +++++++++++++++++ bindings/sysman/python/source/pyzes.py | 224 ++++++++++++++++ .../sysman/python/test/unit_tests/test_ras.py | 239 ++++++++++++++++++ 4 files changed, 698 insertions(+) create mode 100644 bindings/sysman/python/test/unit_tests/test_ras.py diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..6987cabc 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -195,6 +195,14 @@ SUPPORTED APIs | `zesDeviceEccConfigurable` | ECC | 0.1.2 | None | | `zesDeviceGetEccState` | ECC | 0.1.2 | None | | `zesDeviceSetEccState` | ECC | 0.1.2 | None | +| **RAS (Reliability, Availability, Serviceability)** |-|-|-| +| `zesDeviceEnumRasErrorSets` | RAS | 0.1.3 | None | +| `zesRasGetProperties` | RAS | 0.1.3 | None | +| `zesRasGetSupportedCategoriesExp` | RAS | 0.1.3 | Experimental API | +| `zesRasGetStateExp2` | RAS | 0.1.3 | Experimental API | +| `zesRasGetConfigExp` | RAS | 0.1.3 | Experimental API | +| `zesRasSetConfigExp` | RAS | 0.1.3 | Experimental API. Linux: Requires superuser | +| `zesRasClearStateExp` | RAS | 0.1.3 | Experimental API. Linux: Requires superuser | RELEASE NOTES ------------- diff --git a/bindings/sysman/python/source/examples/pyzes_black_box_test.py b/bindings/sysman/python/source/examples/pyzes_black_box_test.py index 135191c3..79f08909 100755 --- a/bindings/sysman/python/source/examples/pyzes_black_box_test.py +++ b/bindings/sysman/python/source/examples/pyzes_black_box_test.py @@ -321,6 +321,35 @@ def get_device_action_string(action): return action_map.get(action, f"UNKNOWN_DEVICE_ACTION_{action}") +def get_ras_error_type_string(error_type): + """Convert RAS error type enum to string""" + type_map = { + pz.ZES_RAS_ERROR_TYPE_CORRECTABLE: "ZES_RAS_ERROR_TYPE_CORRECTABLE", + pz.ZES_RAS_ERROR_TYPE_UNCORRECTABLE: "ZES_RAS_ERROR_TYPE_UNCORRECTABLE", + } + return type_map.get(error_type, f"UNKNOWN_RAS_ERROR_TYPE_{error_type}") + + +def get_ras_error_category_string(category): + """Convert RAS error category enum to string""" + category_map = { + pz.ZES_RAS_ERROR_CATEGORY_EXP_RESET: "ZES_RAS_ERROR_CATEGORY_EXP_RESET", + pz.ZES_RAS_ERROR_CATEGORY_EXP_PROGRAMMING_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_PROGRAMMING_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_DRIVER_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_DRIVER_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_COMPUTE_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_COMPUTE_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_NON_COMPUTE_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_NON_COMPUTE_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_DISPLAY_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_DISPLAY_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_MEMORY_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_MEMORY_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_SCALE_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_SCALE_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_L3FABRIC_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_L3FABRIC_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_PCIE_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_PCIE_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_FABRIC_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_FABRIC_ERRORS", + pz.ZES_RAS_ERROR_CATEGORY_EXP_SOC_INTERNAL_ERRORS: "ZES_RAS_ERROR_CATEGORY_EXP_SOC_INTERNAL_ERRORS", + } + return category_map.get(category, f"UNKNOWN_RAS_ERROR_CATEGORY_{category}") + + def is_root_user(): """Return whether the current user has root privileges on platforms that support it""" geteuid = getattr(os, "geteuid", None) @@ -840,6 +869,175 @@ def test_engine_modules(device_handle, device_index): return True +def get_ras_states(ras_handle, ras_index, categories, category_count, label): + """Read RAS error counters for the given categories, returns None on failure""" + StateArray = pz.zes_ras_state_exp2_t * category_count + states = StateArray() + for j in range(category_count): + states[j].stype = pz.ZES_STRUCTURE_TYPE_RAS_STATE_EXP2 + states[j].pNext = None + + rc = pz.zesRasGetStateExp2(ras_handle, category_count, categories, states) + if not check_rc(f"zesRasGetStateExp2(ras {ras_index}, {label})", rc): + return None + return states + + +def get_ras_configs(ras_handle, ras_index, categories, category_count, label): + """Read RAS thresholds for the given categories, returns None on failure""" + ConfigArray = pz.zes_ras_config_exp_t * category_count + configs = ConfigArray() + for j in range(category_count): + configs[j].stype = pz.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + configs[j].pNext = None + configs[j].category = categories[j] + + rc = pz.zesRasGetConfigExp(ras_handle, category_count, configs) + if not check_rc(f"zesRasGetConfigExp(ras {ras_index}, {label})", rc): + return None + return configs + + +def test_ras_module(device_handle, device_index, set_threshold=None, clear_state=False): + """Test RAS error set enumeration, properties, state, and config operations""" + print(f"\n---- Device {device_index} RAS Test ----") + + ras_count = c_uint32(0) + rc = pz.zesDeviceEnumRasErrorSets(device_handle, byref(ras_count), None) + if not check_rc(f"zesDeviceEnumRasErrorSets(device {device_index}, count)", rc): + return False + + if ras_count.value == 0: + print_verbose("No RAS error sets found on this device") + return True + + print_verbose(f"Found {ras_count.value} RAS error set(s)") + + RasArray = pz.zes_ras_handle_t * ras_count.value + ras_handles = RasArray() + + rc = pz.zesDeviceEnumRasErrorSets(device_handle, byref(ras_count), ras_handles) + if not check_rc(f"zesDeviceEnumRasErrorSets(device {device_index}, handles)", rc): + return False + + for i in range(ras_count.value): + print_verbose(f"\n RAS Error Set {i}:") + + props = pz.zes_ras_properties_t() + props.stype = pz.ZES_STRUCTURE_TYPE_RAS_PROPERTIES + props.pNext = None + + rc = pz.zesRasGetProperties(ras_handles[i], byref(props)) + if not check_rc(f"zesRasGetProperties(ras {i})", rc): + continue + + print_verbose(f" Type: {get_ras_error_type_string(props.type)}") + print_verbose(f" On Subdevice: {bool(props.onSubdevice)}") + if props.onSubdevice: + print_verbose(f" Subdevice ID: {props.subdeviceId}") + + category_count = c_uint32(0) + rc = pz.zesRasGetSupportedCategoriesExp( + ras_handles[i], byref(category_count), None + ) + if not check_rc(f"zesRasGetSupportedCategoriesExp(ras {i}, count)", rc): + continue + + if category_count.value == 0: + print_verbose(" No supported RAS error categories") + continue + + count = category_count.value + CategoryArray = pz.zes_ras_error_category_exp_t * count + categories = CategoryArray() + rc = pz.zesRasGetSupportedCategoriesExp( + ras_handles[i], byref(category_count), categories + ) + if not check_rc(f"zesRasGetSupportedCategoriesExp(ras {i}, categories)", rc): + continue + + states = get_ras_states(ras_handles[i], i, categories, count, "current") + if states is None: + continue + + configs = get_ras_configs(ras_handles[i], i, categories, count, "current") + + print_verbose(" Error Counters:") + for j in range(count): + line = f" {get_ras_error_category_string(categories[j])}: {states[j].errorCounter}" + if configs is not None: + line += f" (threshold: {configs[j].threshold})" + print_verbose(line) + + if not is_root_user(): + print_verbose( + " Skipping zesRasSetConfigExp and zesRasClearStateExp tests due to insufficient permissions" + ) + continue + + if configs is not None and set_threshold is not None: + ConfigArray = pz.zes_ras_config_exp_t * count + new_configs = ConfigArray() + for j in range(count): + new_configs[j].stype = pz.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + new_configs[j].pNext = None + new_configs[j].category = categories[j] + new_configs[j].threshold = set_threshold + + print_verbose(f" Setting threshold {set_threshold} for all categories") + rc = pz.zesRasSetConfigExp(ras_handles[i], count, new_configs) + if check_rc(f"zesRasSetConfigExp(ras {i}, threshold={set_threshold})", rc): + verify = get_ras_configs(ras_handles[i], i, categories, count, "verify") + if verify is not None: + print_verbose(" Thresholds after set:") + for j in range(count): + status = ( + "OK" if verify[j].threshold == set_threshold else "MISMATCH" + ) + print_verbose( + f" {get_ras_error_category_string(categories[j])}: {verify[j].threshold} ({status})" + ) + + # Restore the thresholds read before the test + print_verbose(" Restoring original thresholds") + rc = pz.zesRasSetConfigExp(ras_handles[i], count, configs) + check_rc(f"zesRasSetConfigExp(ras {i}, restore)", rc) + elif configs is not None: + # Write back the thresholds just read so the device configuration is unchanged + rc = pz.zesRasSetConfigExp(ras_handles[i], count, configs) + check_rc(f"zesRasSetConfigExp(ras {i})", rc) + + if clear_state: + print_verbose(" Clearing all RAS error categories") + for j in range(count): + rc = pz.zesRasClearStateExp(ras_handles[i], categories[j]) + check_rc( + f"zesRasClearStateExp(ras {i}, {get_ras_error_category_string(categories[j])})", + rc, + ) + + after = get_ras_states(ras_handles[i], i, categories, count, "after clear") + if after is not None: + print_verbose(" Error Counters (before -> after clear):") + for j in range(count): + print_verbose( + f" {get_ras_error_category_string(categories[j])}: " + f"{states[j].errorCounter} -> {after[j].errorCounter}" + ) + else: + # Only clear categories that have no errors so that no counter data is lost + for j in range(count): + if states[j].errorCounter != 0: + continue + rc = pz.zesRasClearStateExp(ras_handles[i], categories[j]) + check_rc( + f"zesRasClearStateExp(ras {i}, {get_ras_error_category_string(categories[j])})", + rc, + ) + + return True + + def test_memory_modules(device_handle, device_index): """Test memory module enumeration and operations""" print(f"\n---- Device {device_index} Memory Modules Test ----") @@ -1395,6 +1593,9 @@ def run_all_tests(): # Test engine modules test_engine_modules(devices[device_idx], device_idx) + # Test RAS module + test_ras_module(devices[device_idx], device_idx) + print("\n=== Test Completed ===") return True @@ -1413,6 +1614,9 @@ def main(): %(prog)s -f # Frequency tests only %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only + %(prog)s -r # RAS tests only + %(prog)s -r --set-threshold 5 # RAS tests, set then restore threshold 5 (root) + %(prog)s -r --clear-ras-state # RAS tests, clear all error counters (root) %(prog)s -h # Show help message""", formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -1450,8 +1654,22 @@ def main(): version="Python Level Zero Sysman Black Box Test v1.0", ) parser.add_argument("-e", "--engine", action="store_true", help="Run engine tests ") + parser.add_argument("-r", "--ras", action="store_true", help="Run only RAS tests") + parser.add_argument( + "--set-threshold", + type=int, + metavar="VALUE", + help="With -r, set the RAS threshold of all categories, verify it and restore the original (requires root)", + ) + parser.add_argument( + "--clear-ras-state", + action="store_true", + help="With -r, clear all RAS error counters and print them before and after (requires root)", + ) args = parser.parse_args() + if args.set_threshold is not None and args.set_threshold < 0: + parser.error("--set-threshold must be a non-negative integer") # Check if any specific test is requested specific_test = ( @@ -1463,6 +1681,7 @@ def main(): or args.frequency or args.temperature or args.engine + or args.ras or args.all ) @@ -1515,6 +1734,14 @@ def main(): if args.temperature: test_temperature_sensors(devices[device_idx], device_idx) + if args.ras: + test_ras_module( + devices[device_idx], + device_idx, + set_threshold=args.set_threshold, + clear_state=args.clear_ras_state, + ) + success = True return 0 if success else 1 diff --git a/bindings/sysman/python/source/pyzes.py b/bindings/sysman/python/source/pyzes.py index 25ab2e3d..f0fdfbfc 100644 --- a/bindings/sysman/python/source/pyzes.py +++ b/bindings/sysman/python/source/pyzes.py @@ -148,6 +148,10 @@ class zes_engine_handle_t(c_void_p): pass +class zes_ras_handle_t(c_void_p): + pass + + ## ze_bool_t = c_uint8 @@ -294,6 +298,28 @@ class zes_engine_handle_t(c_void_p): ZES_ENGINE_GROUP_MEDIA_CODEC_SINGLE = 14 ZES_ENGINE_GROUP_FORCE_UINT32 = 0x7FFFFFFF +## RAS enums ## +zes_ras_error_type_t = c_int32 +ZES_RAS_ERROR_TYPE_CORRECTABLE = 0 +ZES_RAS_ERROR_TYPE_UNCORRECTABLE = 1 +ZES_RAS_ERROR_TYPE_FORCE_UINT32 = 0x7FFFFFFF + +zes_ras_error_category_exp_t = c_int32 +ZES_RAS_ERROR_CATEGORY_EXP_RESET = 0 +ZES_RAS_ERROR_CATEGORY_EXP_PROGRAMMING_ERRORS = 1 +ZES_RAS_ERROR_CATEGORY_EXP_DRIVER_ERRORS = 2 +ZES_RAS_ERROR_CATEGORY_EXP_COMPUTE_ERRORS = 3 +ZES_RAS_ERROR_CATEGORY_EXP_NON_COMPUTE_ERRORS = 4 +ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS = 5 +ZES_RAS_ERROR_CATEGORY_EXP_DISPLAY_ERRORS = 6 +ZES_RAS_ERROR_CATEGORY_EXP_MEMORY_ERRORS = 7 +ZES_RAS_ERROR_CATEGORY_EXP_SCALE_ERRORS = 8 +ZES_RAS_ERROR_CATEGORY_EXP_L3FABRIC_ERRORS = 9 +ZES_RAS_ERROR_CATEGORY_EXP_PCIE_ERRORS = 10 +ZES_RAS_ERROR_CATEGORY_EXP_FABRIC_ERRORS = 11 +ZES_RAS_ERROR_CATEGORY_EXP_SOC_INTERNAL_ERRORS = 12 +ZES_RAS_ERROR_CATEGORY_EXP_FORCE_UINT32 = 0x7FFFFFFF + ze_result_t = c_int32 ZE_RESULT_SUCCESS = 0 ZE_RESULT_NOT_READY = 1 @@ -376,6 +402,9 @@ class zes_engine_handle_t(c_void_p): ZES_STRUCTURE_TYPE_TEMP_PROPERTIES = 0x14 ZES_STRUCTURE_TYPE_TEMP_CONFIG = 0x23 ZES_STRUCTURE_TYPE_ENGINE_PROPERTIES = 0x5 +ZES_STRUCTURE_TYPE_RAS_PROPERTIES = 0xF +ZES_STRUCTURE_TYPE_RAS_STATE_EXP2 = 0x00020015 # Experimental RAS state +ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP = 0x00020016 # Experimental RAS config ## Core ze_device UUID struct ## @@ -779,6 +808,36 @@ class zes_engine_stats_t(_PrintableStructure): _fmt_ = {"activeTime": "%d", "timestamp": "%d"} +## RAS structures ## +class zes_ras_properties_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_RAS_PROPERTIES + ("pNext", c_void_p), + ("type", zes_ras_error_type_t), # correctable/uncorrectable + ("onSubdevice", ze_bool_t), # is on subdevice + ("subdeviceId", c_uint32), # subdevice ID + ] + + +class zes_ras_state_exp2_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_RAS_STATE_EXP2 + ("pNext", c_void_p), + ("errorCounter", c_uint64), # error counter for the requested category + ] + _fmt_ = {"errorCounter": "%d"} + + +class zes_ras_config_exp_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + ("pNext", c_void_p), + ("category", zes_ras_error_category_exp_t), # RAS error category + ("threshold", c_uint64), # threshold to trigger RAS events (0 disables) + ] + _fmt_ = {"threshold": "%d"} + + ## Function access ## _getFunctionPointerList: Dict[str, Any] = dict() @@ -1556,3 +1615,168 @@ def zesEngineGetActivity(hEngine, pStats): retVal = funcPtr(hEngine, pStats) return retVal + + +## RAS functions ## + + +def zesDeviceEnumRasErrorSets(hDevice, pCount, phRas): + """Wraps API: + ze_result_t zesDeviceEnumRasErrorSets( + zes_device_handle_t hDevice, + uint32_t* pCount, + zes_ras_handle_t* phRas) + + Parameters: + hDevice: device handle + pCount: POINTER(c_uint32) + phRas: POINTER(zes_ras_handle_t) or None + """ + funcPtr = getFunctionPointerList("zesDeviceEnumRasErrorSets") + funcPtr.argtypes = [ + zes_device_handle_t, + POINTER(c_uint32), + POINTER(zes_ras_handle_t), + ] + funcPtr.restype = ze_result_t + retVal = funcPtr(hDevice, pCount, phRas) + return retVal + + +def zesRasGetProperties(hRas, pProperties): + """Wraps API: + ze_result_t zesRasGetProperties( + zes_ras_handle_t hRas, + zes_ras_properties_t* pProperties) + + Parameters: + hRas: RAS handle + pProperties: POINTER(zes_ras_properties_t) - properties structure to fill + Returns: + ze_result_t - return code only, properties are filled into pProperties + """ + funcPtr = getFunctionPointerList("zesRasGetProperties") + funcPtr.argtypes = [zes_ras_handle_t, POINTER(zes_ras_properties_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, pProperties) + return retVal + + +def zesRasGetSupportedCategoriesExp(hRas, pCount, pCategories): + """Wraps API: + ze_result_t zesRasGetSupportedCategoriesExp( + zes_ras_handle_t hRas, + uint32_t* pCount, + zes_ras_error_category_exp_t* pCategories) + + Parameters: + hRas: RAS handle + pCount: POINTER(c_uint32) + pCategories: POINTER(zes_ras_error_category_exp_t) or None + Returns: + ze_result_t - return code only, supported categories are filled into pCategories + """ + funcPtr = getFunctionPointerList("zesRasGetSupportedCategoriesExp") + funcPtr.argtypes = [ + zes_ras_handle_t, + POINTER(c_uint32), + POINTER(zes_ras_error_category_exp_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, pCount, pCategories) + return retVal + + +def zesRasGetStateExp2(hRas, count, pCategories, pState): + """Wraps API: + ze_result_t zesRasGetStateExp2( + zes_ras_handle_t hRas, + const uint32_t count, + const zes_ras_error_category_exp_t* pCategories, + zes_ras_state_exp2_t* pState) + + Parameters: + hRas: RAS handle + count: number of elements in pCategories and pState + pCategories: POINTER(zes_ras_error_category_exp_t) - categories to query + pState: POINTER(zes_ras_state_exp2_t) - states to fill, pState[i] maps to pCategories[i] + Returns: + ze_result_t - return code only, error counters are filled into pState + """ + funcPtr = getFunctionPointerList("zesRasGetStateExp2") + funcPtr.argtypes = [ + zes_ras_handle_t, + c_uint32, + POINTER(zes_ras_error_category_exp_t), + POINTER(zes_ras_state_exp2_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, count, pCategories, pState) + return retVal + + +def zesRasGetConfigExp(hRas, count, pConfig): + """Wraps API: + ze_result_t zesRasGetConfigExp( + zes_ras_handle_t hRas, + const uint32_t count, + zes_ras_config_exp_t* pConfig) + + Parameters: + hRas: RAS handle + count: number of elements in pConfig + pConfig: POINTER(zes_ras_config_exp_t) - category set by caller, threshold filled by driver + Returns: + ze_result_t - return code only, thresholds are filled into pConfig + """ + funcPtr = getFunctionPointerList("zesRasGetConfigExp") + funcPtr.argtypes = [zes_ras_handle_t, c_uint32, POINTER(zes_ras_config_exp_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, count, pConfig) + return retVal + + +def zesRasSetConfigExp(hRas, count, pConfig): + """Wraps API: + ze_result_t zesRasSetConfigExp( + zes_ras_handle_t hRas, + const uint32_t count, + const zes_ras_config_exp_t* pConfig) + + Parameters: + hRas: RAS handle + count: number of elements in pConfig + pConfig: POINTER(zes_ras_config_exp_t) - thresholds to set per category + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesRasSetConfigExp") + funcPtr.argtypes = [zes_ras_handle_t, c_uint32, POINTER(zes_ras_config_exp_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, count, pConfig) + return retVal + + +def zesRasClearStateExp(hRas, category): + """Wraps API: + ze_result_t zesRasClearStateExp( + zes_ras_handle_t hRas, + zes_ras_error_category_exp_t category) + + Parameters: + hRas: RAS handle + category: zes_ras_error_category_exp_t - category whose error counter is cleared + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesRasClearStateExp") + funcPtr.argtypes = [zes_ras_handle_t, zes_ras_error_category_exp_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hRas, category) + return retVal diff --git a/bindings/sysman/python/test/unit_tests/test_ras.py b/bindings/sysman/python/test/unit_tests/test_ras.py new file mode 100644 index 00000000..edceb552 --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_ras.py @@ -0,0 +1,239 @@ +## +# Copyright (C) 2026 Intel Corporation +# +# SPDX-License-Identifier: MIT +# +## + +import os +import sys +import unittest +from ctypes import * +from unittest.mock import MagicMock, patch + +# Add the source directory to Python path so we can import pyzes +script_dir = os.path.dirname(os.path.abspath(__file__)) +source_dir = os.path.join(script_dir, "..", "..", "source") +source_dir = os.path.abspath(source_dir) +if source_dir not in sys.path: + sys.path.insert(0, source_dir) + + +@patch("pyzes.getFunctionPointerList") +class TestRasFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEnumRasErrorSetsThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 2 + + def mock_enum_ras(device_handle, count_ptr, handles_ptr): + count_ptr._obj.value = mock_count + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_enum_ras) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDeviceEnumRasErrorSets(device_handle, byref(count), None) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + mock_get_func.assert_called_with("zesDeviceEnumRasErrorSets") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasGetPropertiesThenCallSucceedsWithValidProperties( + self, mock_get_func + ): + def mock_get_properties(ras_handle, properties_ptr): + properties = properties_ptr._obj + properties.type = self.pyzes.ZES_RAS_ERROR_TYPE_UNCORRECTABLE + properties.onSubdevice = 1 + properties.subdeviceId = 1 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_properties) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + properties = self.pyzes.zes_ras_properties_t() + properties.stype = self.pyzes.ZES_STRUCTURE_TYPE_RAS_PROPERTIES + + result = self.pyzes.zesRasGetProperties(ras_handle, byref(properties)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(properties.type, self.pyzes.ZES_RAS_ERROR_TYPE_UNCORRECTABLE) + self.assertEqual(properties.onSubdevice, 1) + self.assertEqual(properties.subdeviceId, 1) + mock_get_func.assert_called_with("zesRasGetProperties") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasGetSupportedCategoriesExpThenCallSucceedsWithCategories( + self, mock_get_func + ): + mock_count = 3 + + def mock_get_categories(ras_handle, count_ptr, categories_ptr): + count_ptr._obj.value = mock_count + categories_ptr[0] = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET + categories_ptr[1] = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS + categories_ptr[2] = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_MEMORY_ERRORS + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_categories) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + count = c_uint32(mock_count) + categories = (self.pyzes.zes_ras_error_category_exp_t * mock_count)() + + result = self.pyzes.zesRasGetSupportedCategoriesExp( + ras_handle, byref(count), categories + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual(categories[0], self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET) + self.assertEqual( + categories[1], self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS + ) + self.assertEqual( + categories[2], self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_MEMORY_ERRORS + ) + mock_get_func.assert_called_with("zesRasGetSupportedCategoriesExp") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasGetStateExp2ThenCallSucceedsWithErrorCounters( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_state(ras_handle, count, categories_ptr, state_ptr): + self.assertEqual(count, mock_count) + self.assertEqual( + categories_ptr[0], self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET + ) + self.assertEqual( + categories_ptr[1], self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_PCIE_ERRORS + ) + state_ptr[0].errorCounter = 5 + state_ptr[1].errorCounter = 12 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_state) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + categories = (self.pyzes.zes_ras_error_category_exp_t * mock_count)( + self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET, + self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_PCIE_ERRORS, + ) + states = (self.pyzes.zes_ras_state_exp2_t * mock_count)() + for i in range(mock_count): + states[i].stype = self.pyzes.ZES_STRUCTURE_TYPE_RAS_STATE_EXP2 + states[i].pNext = None + + result = self.pyzes.zesRasGetStateExp2( + ras_handle, mock_count, categories, states + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(states[0].errorCounter, 5) + self.assertEqual(states[1].errorCounter, 12) + mock_get_func.assert_called_with("zesRasGetStateExp2") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasGetConfigExpThenCallSucceedsWithThresholds( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_config(ras_handle, count, config_ptr): + self.assertEqual(count, mock_count) + self.assertEqual( + config_ptr[0].category, + self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_COMPUTE_ERRORS, + ) + self.assertEqual( + config_ptr[1].category, + self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_FABRIC_ERRORS, + ) + config_ptr[0].threshold = 100 + config_ptr[1].threshold = 0 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_config) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + configs = (self.pyzes.zes_ras_config_exp_t * mock_count)() + configs[0].stype = self.pyzes.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + configs[0].category = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_COMPUTE_ERRORS + configs[1].stype = self.pyzes.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + configs[1].category = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_FABRIC_ERRORS + + result = self.pyzes.zesRasGetConfigExp(ras_handle, mock_count, configs) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(configs[0].threshold, 100) + self.assertEqual(configs[1].threshold, 0) + mock_get_func.assert_called_with("zesRasGetConfigExp") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasSetConfigExpThenCallSucceeds( + self, mock_get_func + ): + def mock_set_config(ras_handle, count, config_ptr): + self.assertEqual(count, 1) + self.assertEqual( + config_ptr[0].stype, self.pyzes.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + ) + self.assertEqual( + config_ptr[0].category, + self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS, + ) + self.assertEqual(config_ptr[0].threshold, 50) + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_set_config) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + configs = (self.pyzes.zes_ras_config_exp_t * 1)() + configs[0].stype = self.pyzes.ZES_STRUCTURE_TYPE_RAS_CONFIG_EXP + configs[0].category = self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_CACHE_ERRORS + configs[0].threshold = 50 + + result = self.pyzes.zesRasSetConfigExp(ras_handle, 1, configs) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesRasSetConfigExp") + mock_func.assert_called_once() + + def test_GivenValidRasHandleWhenCallingZesRasClearStateExpThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + ras_handle = self.pyzes.zes_ras_handle_t() + + result = self.pyzes.zesRasClearStateExp( + ras_handle, self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesRasClearStateExp") + mock_func.assert_called_once_with( + ras_handle, self.pyzes.ZES_RAS_ERROR_CATEGORY_EXP_RESET + ) + + +if __name__ == "__main__": + unittest.main() From bdf673ab5e4b0f47ecd1d2d83f9be309c4e91ccf Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Mon, 28 Sep 2026 14:05:44 +0530 Subject: [PATCH 2/2] fix: Set RAS config and clear RAS state only when requested in pyzes black box test --- .../source/examples/pyzes_black_box_test.py | 17 +++-------------- 1 file changed, 3 insertions(+), 14 deletions(-) diff --git a/bindings/sysman/python/source/examples/pyzes_black_box_test.py b/bindings/sysman/python/source/examples/pyzes_black_box_test.py index 79f08909..6fe75a8d 100755 --- a/bindings/sysman/python/source/examples/pyzes_black_box_test.py +++ b/bindings/sysman/python/source/examples/pyzes_black_box_test.py @@ -969,6 +969,9 @@ def test_ras_module(device_handle, device_index, set_threshold=None, clear_state line += f" (threshold: {configs[j].threshold})" print_verbose(line) + if set_threshold is None and not clear_state: + continue + if not is_root_user(): print_verbose( " Skipping zesRasSetConfigExp and zesRasClearStateExp tests due to insufficient permissions" @@ -1002,10 +1005,6 @@ def test_ras_module(device_handle, device_index, set_threshold=None, clear_state print_verbose(" Restoring original thresholds") rc = pz.zesRasSetConfigExp(ras_handles[i], count, configs) check_rc(f"zesRasSetConfigExp(ras {i}, restore)", rc) - elif configs is not None: - # Write back the thresholds just read so the device configuration is unchanged - rc = pz.zesRasSetConfigExp(ras_handles[i], count, configs) - check_rc(f"zesRasSetConfigExp(ras {i})", rc) if clear_state: print_verbose(" Clearing all RAS error categories") @@ -1024,16 +1023,6 @@ def test_ras_module(device_handle, device_index, set_threshold=None, clear_state f" {get_ras_error_category_string(categories[j])}: " f"{states[j].errorCounter} -> {after[j].errorCounter}" ) - else: - # Only clear categories that have no errors so that no counter data is lost - for j in range(count): - if states[j].errorCounter != 0: - continue - rc = pz.zesRasClearStateExp(ras_handles[i], categories[j]) - check_rc( - f"zesRasClearStateExp(ras {i}, {get_ras_error_category_string(categories[j])})", - rc, - ) return True