From 6b27cb1acc98b01f9cc7baf7d48d697c54725921 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:27:11 +0530 Subject: [PATCH 1/2] feature: Add pyzes support for device APIs Add python bindings, unit tests, black box test coverage and README entries for: 1. zesDeviceGetState 2. zesDeviceResetExt 3. zesDeviceEnumFirmwares 4. zesDeviceEventRegister 5. zesDeviceGetHealthStatusExt 6. zesDeviceSetHealthStatusExt Related-To: SYSM-434 Signed-off-by: Aviral Nigam --- bindings/sysman/python/README.md | 6 + .../source/examples/pyzes_black_box_test.py | 259 +++++++++++++++++- bindings/sysman/python/source/pyzes.py | 199 ++++++++++++++ .../python/test/unit_tests/test_device.py | 194 +++++++++++++ 4 files changed, 656 insertions(+), 2 deletions(-) create mode 100644 bindings/sysman/python/test/unit_tests/test_device.py diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..dfd8a4fc 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -158,6 +158,12 @@ SUPPORTED APIs | `zesDeviceGetProperties` | Device | 0.1.0 | None | | `zesDriverGetDeviceByUuidExp` | Device | 0.1.0 | Experimental API | | `zesDeviceProcessesGetState` | Device | 0.1.0 | None | +| `zesDeviceGetState` | Device | 0.1.3 | None | +| `zesDeviceResetExt` | Device | 0.1.3 | Linux: Requires superuser; all device state is lost | +| `zesDeviceEventRegister` | Device | 0.1.3 | None | +| `zesDeviceGetHealthStatusExt` | Device | 0.1.3 | None | +| `zesDeviceSetHealthStatusExt` | Device | 0.1.3 | Linux: Requires superuser; persisted to non-volatile memory | +| `zesDeviceEnumFirmwares` | Device | 0.1.3 | None | | **Memory Management** |-|-|-| | `zesDeviceEnumMemoryModules` | Memory | 0.1.0 | None | | `zesMemoryGetProperties` | Memory | 0.1.0 | None | 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..16e532ca 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,76 @@ def get_device_action_string(action): return action_map.get(action, f"UNKNOWN_DEVICE_ACTION_{action}") +def get_event_type_flags_string(events): + """Convert event type flags to a string of flag names""" + flag_map = { + pz.ZES_EVENT_TYPE_FLAG_DEVICE_DETACH: "DEVICE_DETACH", + pz.ZES_EVENT_TYPE_FLAG_DEVICE_ATTACH: "DEVICE_ATTACH", + pz.ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_ENTER: "DEVICE_SLEEP_STATE_ENTER", + pz.ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_EXIT: "DEVICE_SLEEP_STATE_EXIT", + pz.ZES_EVENT_TYPE_FLAG_FREQ_THROTTLED: "FREQ_THROTTLED", + pz.ZES_EVENT_TYPE_FLAG_ENERGY_THRESHOLD_CROSSED: "ENERGY_THRESHOLD_CROSSED", + pz.ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL: "TEMP_CRITICAL", + pz.ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD1: "TEMP_THRESHOLD1", + pz.ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD2: "TEMP_THRESHOLD2", + pz.ZES_EVENT_TYPE_FLAG_MEM_HEALTH: "MEM_HEALTH", + pz.ZES_EVENT_TYPE_FLAG_FABRIC_PORT_HEALTH: "FABRIC_PORT_HEALTH", + pz.ZES_EVENT_TYPE_FLAG_PCI_LINK_HEALTH: "PCI_LINK_HEALTH", + pz.ZES_EVENT_TYPE_FLAG_RAS_CORRECTABLE_ERRORS: "RAS_CORRECTABLE_ERRORS", + pz.ZES_EVENT_TYPE_FLAG_RAS_UNCORRECTABLE_ERRORS: "RAS_UNCORRECTABLE_ERRORS", + pz.ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED: "DEVICE_RESET_REQUIRED", + pz.ZES_EVENT_TYPE_FLAG_SURVIVABILITY_MODE_DETECTED: "SURVIVABILITY_MODE_DETECTED", + pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT: "INFO_LOG_CPER_DATA_AVAILABLE_EXT", + } + if events == 0: + return "None" + names = [name for flag, name in flag_map.items() if events & flag] + return " | ".join(names) if names else f"UNKNOWN_EVENTS_{events:#x}" + + +def get_reset_reason_flags_string(reasons): + """Convert reset reason flags to a string of flag names""" + flag_map = { + pz.ZES_RESET_REASON_FLAG_WEDGED: "WEDGED", + pz.ZES_RESET_REASON_FLAG_REPAIR: "REPAIR", + } + if reasons == 0: + return "None" + names = [name for flag, name in flag_map.items() if reasons & flag] + return " | ".join(names) if names else f"UNKNOWN_RESET_REASONS_{reasons:#x}" + + +def get_repair_status_string(status): + """Convert repair status enum to string""" + status_map = { + pz.ZES_REPAIR_STATUS_UNSUPPORTED: "ZES_REPAIR_STATUS_UNSUPPORTED", + pz.ZES_REPAIR_STATUS_NOT_PERFORMED: "ZES_REPAIR_STATUS_NOT_PERFORMED", + pz.ZES_REPAIR_STATUS_PERFORMED: "ZES_REPAIR_STATUS_PERFORMED", + } + return status_map.get(status, f"UNKNOWN_REPAIR_STATUS_{status}") + + +def get_health_status_string(health): + """Convert device health status enum to string""" + health_map = { + pz.ZES_DEVICE_HEALTH_STATUS_EXT_OK: "ZES_DEVICE_HEALTH_STATUS_EXT_OK", + pz.ZES_DEVICE_HEALTH_STATUS_EXT_WARNING: "ZES_DEVICE_HEALTH_STATUS_EXT_WARNING", + pz.ZES_DEVICE_HEALTH_STATUS_EXT_CRITICAL: "ZES_DEVICE_HEALTH_STATUS_EXT_CRITICAL", + pz.ZES_DEVICE_HEALTH_STATUS_EXT_FAILED: "ZES_DEVICE_HEALTH_STATUS_EXT_FAILED", + } + return health_map.get(health, f"UNKNOWN_HEALTH_STATUS_{health}") + + +def get_reset_type_string(reset_type): + """Convert reset type enum to string""" + type_map = { + pz.ZES_RESET_TYPE_WARM: "ZES_RESET_TYPE_WARM", + pz.ZES_RESET_TYPE_COLD: "ZES_RESET_TYPE_COLD", + pz.ZES_RESET_TYPE_FLR: "ZES_RESET_TYPE_FLR", + } + return type_map.get(reset_type, f"UNKNOWN_RESET_TYPE_{reset_type}") + + def is_root_user(): """Return whether the current user has root privileges on platforms that support it""" geteuid = getattr(os, "geteuid", None) @@ -840,6 +910,143 @@ def test_engine_modules(device_handle, device_index): return True +HEALTH_STATUS_ARGS = { + "ok": pz.ZES_DEVICE_HEALTH_STATUS_EXT_OK, + "warning": pz.ZES_DEVICE_HEALTH_STATUS_EXT_WARNING, + "critical": pz.ZES_DEVICE_HEALTH_STATUS_EXT_CRITICAL, + "failed": pz.ZES_DEVICE_HEALTH_STATUS_EXT_FAILED, +} + +RESET_TYPE_ARGS = { + "warm": pz.ZES_RESET_TYPE_WARM, + "cold": pz.ZES_RESET_TYPE_COLD, + "flr": pz.ZES_RESET_TYPE_FLR, +} + + +def print_device_state(device_handle, device_index): + """Read and print the reset reasons and repair status of a device""" + state = pz.zes_device_state_t() + state.stype = pz.ZES_STRUCTURE_TYPE_DEVICE_STATE + state.pNext = None + + rc = pz.zesDeviceGetState(device_handle, byref(state)) + if check_rc(f"zesDeviceGetState(device {device_index})", rc): + print_verbose(" State:") + print_verbose( + f" Reset Reasons: {get_reset_reason_flags_string(state.reset)}" + ) + print_verbose(f" Repair Status: {get_repair_status_string(state.repaired)}") + + +def test_device_reset(device_handle, device_index, force, reset_type): + """Reset a device with zesDeviceResetExt and print its state before and after""" + print(f"\n---- Device {device_index} Reset Test ----") + + if not is_root_user(): + print_verbose("Skipping zesDeviceResetExt due to insufficient permissions") + return True + + print_device_state(device_handle, device_index) + + reset_props = pz.zes_reset_properties_t() + reset_props.stype = pz.ZES_STRUCTURE_TYPE_RESET_PROPERTIES + reset_props.pNext = None + reset_props.force = force + reset_props.resetType = reset_type + + print_verbose( + f" Resetting device with type {get_reset_type_string(reset_type)} and force {bool(force)}" + ) + rc = pz.zesDeviceResetExt(device_handle, byref(reset_props)) + if not check_rc(f"zesDeviceResetExt(device {device_index})", rc): + return False + + print_verbose(" Device reset successfully") + print_device_state(device_handle, device_index) + return True + + +def test_device_state_module(device_handle, device_index, set_health=None): + """Test device state, health status, event registration, and firmware enumeration""" + print(f"\n---- Device {device_index} Device State Test ----") + + print_device_state(device_handle, device_index) + + # zesDeviceResetExt is only exercised with --reset as a reset loses all device + # state and may kill applications using the device. + + health = pz.zes_device_health_status_ext_t(0) + rc = pz.zesDeviceGetHealthStatusExt(device_handle, byref(health)) + if rc != pz.ZE_RESULT_SUCCESS: + print_verbose(f" Health Status: Not available ({get_result_string(rc)})") + else: + print_verbose(f" Health Status: {get_health_status_string(health.value)}") + + if is_root_user() and set_health is not None: + rc = pz.zesDeviceSetHealthStatusExt(device_handle, set_health) + if check_rc( + f"zesDeviceSetHealthStatusExt(device {device_index}, {get_health_status_string(set_health)})", + rc, + ): + read_back = pz.zes_device_health_status_ext_t(0) + rc = pz.zesDeviceGetHealthStatusExt(device_handle, byref(read_back)) + if check_rc( + f"zesDeviceGetHealthStatusExt(device {device_index}, verify)", rc + ): + status = "OK" if read_back.value == set_health else "MISMATCH" + print_verbose( + f" Set health status {get_health_status_string(set_health)}, read back " + f"{get_health_status_string(read_back.value)} ({status})" + ) + + # Restore the health status read before the test + rc = pz.zesDeviceSetHealthStatusExt(device_handle, health.value) + if check_rc( + f"zesDeviceSetHealthStatusExt(device {device_index}, restore)", rc + ): + print_verbose( + f" Restored health status to {get_health_status_string(health.value)}" + ) + elif is_root_user(): + # Write back the health status just read so the device state is unchanged + rc = pz.zesDeviceSetHealthStatusExt(device_handle, health.value) + if check_rc(f"zesDeviceSetHealthStatusExt(device {device_index})", rc): + print_verbose(" Set health status successfully") + else: + print_verbose( + " Skipping zesDeviceSetHealthStatusExt due to insufficient permissions" + ) + + events_to_register = pz.ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED + rc = pz.zesDeviceEventRegister(device_handle, events_to_register) + if rc != pz.ZE_RESULT_SUCCESS: + print_verbose( + f" Device Event Registration: Not available ({get_result_string(rc)})" + ) + else: + print_verbose( + f" Registered device events: {get_event_type_flags_string(events_to_register)}" + ) + # Clear the device event registration made by this test + rc = pz.zesDeviceEventRegister(device_handle, 0) + check_rc(f"zesDeviceEventRegister(device {device_index}, clear)", rc) + + firmware_count = c_uint32(0) + rc = pz.zesDeviceEnumFirmwares(device_handle, byref(firmware_count), None) + if check_rc(f"zesDeviceEnumFirmwares(device {device_index}, count)", rc): + print_verbose(f" Found {firmware_count.value} firmware component(s)") + if firmware_count.value > 0: + FirmwareArray = pz.zes_firmware_handle_t * firmware_count.value + firmware_handles = FirmwareArray() + rc = pz.zesDeviceEnumFirmwares( + device_handle, byref(firmware_count), firmware_handles + ) + check_rc(f"zesDeviceEnumFirmwares(device {device_index}, handles)", 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 ----") @@ -1374,6 +1581,9 @@ def run_all_tests(): # Test global device operations (properties and processes) test_global_operation(drivers[driver_idx], devices[device_idx], device_idx) + # Test device state, health, events, and firmware enumeration + test_device_state_module(devices[device_idx], device_idx) + # Test PCI module test_pci_module(devices[device_idx], device_idx) @@ -1406,7 +1616,9 @@ def main(): epilog="""Examples: %(prog)s -a # Run all tests %(prog)s -m # Memory tests only - %(prog)s -g # Global operations (device properties and processes) only + %(prog)s -g # Global operations (device properties, processes, and state) only + %(prog)s -g --set-health warning # Global operations, set then restore health status (root) + %(prog)s --reset noforce --reset-device 0 # Warm reset of device 0 with zesDeviceResetExt (root) %(prog)s -p # PCI tests only %(prog)s -C # ECC tests only %(prog)s -o # Power tests only @@ -1425,7 +1637,29 @@ def main(): "-g", "--global", action="store_true", - help="Run only global operations (device properties and processes)", + help="Run only global operations (device properties, processes, and state)", + ) + parser.add_argument( + "--set-health", + choices=sorted(HEALTH_STATUS_ARGS), + help="With -g, set the device health status, verify it and restore the original (requires root)", + ) + parser.add_argument( + "--reset", + choices=["force", "noforce"], + help="Run only the device reset test with zesDeviceResetExt on all devices (requires root)", + ) + parser.add_argument( + "--reset-type", + choices=list(RESET_TYPE_ARGS), + default="warm", + help="With --reset, type of reset to perform, default is warm", + ) + parser.add_argument( + "--reset-device", + type=int, + metavar="DEVICE", + help="With --reset, only reset the given device", ) parser.add_argument( "-o", "--power", action="store_true", help="Run only power-related tests" @@ -1452,11 +1686,14 @@ def main(): parser.add_argument("-e", "--engine", action="store_true", help="Run engine tests ") args = parser.parse_args() + if args.reset_device is not None and args.reset_device < 0: + parser.error("--reset-device must be a non-negative integer") # Check if any specific test is requested specific_test = ( args.memory or getattr(args, "global", False) + or args.reset is not None or args.pci or args.ecc or args.power @@ -1493,6 +1730,15 @@ def main(): for device_idx in range(device_count): if getattr(args, "global", False): test_global_operation(drivers[0], devices[device_idx], device_idx) + test_device_state_module( + devices[device_idx], + device_idx, + set_health=( + HEALTH_STATUS_ARGS[args.set_health] + if args.set_health is not None + else None + ), + ) if args.pci: test_pci_module(devices[device_idx], device_idx) @@ -1515,6 +1761,15 @@ def main(): if args.temperature: test_temperature_sensors(devices[device_idx], device_idx) + # Run the reset last since it loses all device state + if args.reset is not None and args.reset_device in (None, device_idx): + test_device_reset( + devices[device_idx], + device_idx, + args.reset == "force", + RESET_TYPE_ARGS[args.reset_type], + ) + 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..05d46aa9 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_firmware_handle_t(c_void_p): + pass + + ## ze_bool_t = c_uint8 @@ -172,6 +176,31 @@ class zes_engine_handle_t(c_void_p): ZES_DEVICE_TYPE_VPU = 5 ZES_DEVICE_TYPE_FORCE_UINT32 = 0x7FFFFFFF +## Device state, reset and health enums ## +zes_repair_status_t = c_int32 +ZES_REPAIR_STATUS_UNSUPPORTED = 0 +ZES_REPAIR_STATUS_NOT_PERFORMED = 1 +ZES_REPAIR_STATUS_PERFORMED = 2 +ZES_REPAIR_STATUS_FORCE_UINT32 = 0x7FFFFFFF + +zes_reset_reason_flags_t = c_uint32 +ZES_RESET_REASON_FLAG_WEDGED = 1 << 0 +ZES_RESET_REASON_FLAG_REPAIR = 1 << 1 +ZES_RESET_REASON_FLAG_FORCE_UINT32 = 0x7FFFFFFF + +zes_reset_type_t = c_int32 +ZES_RESET_TYPE_WARM = 0 +ZES_RESET_TYPE_COLD = 1 +ZES_RESET_TYPE_FLR = 2 +ZES_RESET_TYPE_FORCE_UINT32 = 0x7FFFFFFF + +zes_device_health_status_ext_t = c_int32 +ZES_DEVICE_HEALTH_STATUS_EXT_OK = 0 +ZES_DEVICE_HEALTH_STATUS_EXT_WARNING = 1 +ZES_DEVICE_HEALTH_STATUS_EXT_CRITICAL = 2 +ZES_DEVICE_HEALTH_STATUS_EXT_FAILED = 3 +ZES_DEVICE_HEALTH_STATUS_EXT_FORCE_UINT32 = 0x7FFFFFFF + # Memory type enumeration zes_mem_type_t = c_int32 ZES_MEM_TYPE_HBM = 0 @@ -262,6 +291,27 @@ class zes_engine_handle_t(c_void_p): ZES_FREQ_THROTTLE_REASON_FLAG_POWER = 1 << 9 ZES_FREQ_THROTTLE_REASON_FLAG_FORCE_UINT32 = 0x7FFFFFFF +## Event type flags ## +zes_event_type_flags_t = c_uint32 +ZES_EVENT_TYPE_FLAG_DEVICE_DETACH = 1 << 0 +ZES_EVENT_TYPE_FLAG_DEVICE_ATTACH = 1 << 1 +ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_ENTER = 1 << 2 +ZES_EVENT_TYPE_FLAG_DEVICE_SLEEP_STATE_EXIT = 1 << 3 +ZES_EVENT_TYPE_FLAG_FREQ_THROTTLED = 1 << 4 +ZES_EVENT_TYPE_FLAG_ENERGY_THRESHOLD_CROSSED = 1 << 5 +ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL = 1 << 6 +ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD1 = 1 << 7 +ZES_EVENT_TYPE_FLAG_TEMP_THRESHOLD2 = 1 << 8 +ZES_EVENT_TYPE_FLAG_MEM_HEALTH = 1 << 9 +ZES_EVENT_TYPE_FLAG_FABRIC_PORT_HEALTH = 1 << 10 +ZES_EVENT_TYPE_FLAG_PCI_LINK_HEALTH = 1 << 11 +ZES_EVENT_TYPE_FLAG_RAS_CORRECTABLE_ERRORS = 1 << 12 +ZES_EVENT_TYPE_FLAG_RAS_UNCORRECTABLE_ERRORS = 1 << 13 +ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED = 1 << 14 +ZES_EVENT_TYPE_FLAG_SURVIVABILITY_MODE_DETECTED = 1 << 15 +ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT = 1 << 16 +ZES_EVENT_TYPE_FLAG_FORCE_UINT32 = 0x7FFFFFFF + ## Temperature sensor enums ## zes_temp_sensors_t = c_int32 ZES_TEMP_SENSORS_GLOBAL = 0 @@ -365,6 +415,8 @@ class zes_engine_handle_t(c_void_p): ZES_STRUCTURE_TYPE_POWER_LIMIT_EXT_DESC = 0x27 ZES_STRUCTURE_TYPE_POWER_EXT_PROPERTIES = 0x28 ZES_STRUCTURE_TYPE_PROCESS_STATE = 0x16 +ZES_STRUCTURE_TYPE_DEVICE_STATE = 0x15 +ZES_STRUCTURE_TYPE_RESET_PROPERTIES = 0x2C ZES_STRUCTURE_TYPE_DEVICE_EXT_PROPERTIES = 0x2D # from zes_structure_type_t ZES_STRUCTURE_TYPE_SUBDEVICE_EXP_PROPERTIES = ( 0x00020004 # Experimental subdevice properties @@ -426,6 +478,26 @@ class zes_device_properties_t(_PrintableStructure): ] +## Sysman zes_device_state_t ## +class zes_device_state_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_DEVICE_STATE + ("pNext", c_void_p), + ("reset", zes_reset_reason_flags_t), # reasons the device needs a reset + ("repaired", zes_repair_status_t), # repair status + ] + + +## Sysman zes_reset_properties_t ## +class zes_reset_properties_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_RESET_PROPERTIES + ("pNext", c_void_p), + ("force", ze_bool_t), # forcibly kill applications using the device + ("resetType", zes_reset_type_t), # type of reset to perform + ] + + ## Sysman zes_process_state_t ## class zes_process_state_t(_PrintableStructure): _fields_ = [ @@ -1065,6 +1137,133 @@ def zesDeviceProcessesGetState(hDevice, pCount, pProcesses): return retVal +def zesDeviceGetState(hDevice, pState): + """Wraps API: + ze_result_t zesDeviceGetState( + zes_device_handle_t hDevice, + zes_device_state_t* pState) + + Parameters: + hDevice: device handle + pState: POINTER(zes_device_state_t) - state structure to fill + Returns: + ze_result_t - return code only, state is filled into pState + """ + funcPtr = getFunctionPointerList("zesDeviceGetState") + funcPtr.argtypes = [zes_device_handle_t, POINTER(zes_device_state_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDevice, pState) + return retVal + + +def zesDeviceResetExt(hDevice, pProperties): + """Wraps API: + ze_result_t zesDeviceResetExt( + zes_device_handle_t hDevice, + zes_reset_properties_t* pProperties) + + Performs a PCI bus reset of the device and all current device state is lost. + + Parameters: + hDevice: device handle + pProperties: POINTER(zes_reset_properties_t) - reset properties to apply + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesDeviceResetExt") + funcPtr.argtypes = [zes_device_handle_t, POINTER(zes_reset_properties_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDevice, pProperties) + return retVal + + +def zesDeviceEventRegister(hDevice, events): + """Wraps API: + ze_result_t zesDeviceEventRegister( + zes_device_handle_t hDevice, + zes_event_type_flags_t events) + + Parameters: + hDevice: device handle + events: zes_event_type_flags_t - events to listen to, 0 to clear + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesDeviceEventRegister") + funcPtr.argtypes = [zes_device_handle_t, zes_event_type_flags_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDevice, events) + return retVal + + +def zesDeviceGetHealthStatusExt(hDevice, pHealth): + """Wraps API: + ze_result_t zesDeviceGetHealthStatusExt( + zes_device_handle_t hDevice, + zes_device_health_status_ext_t* pHealth) + + Parameters: + hDevice: device handle + pHealth: POINTER(zes_device_health_status_ext_t) - current health status to fill + Returns: + ze_result_t - return code only, health status is filled into pHealth + """ + funcPtr = getFunctionPointerList("zesDeviceGetHealthStatusExt") + funcPtr.argtypes = [zes_device_handle_t, POINTER(zes_device_health_status_ext_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDevice, pHealth) + return retVal + + +def zesDeviceSetHealthStatusExt(hDevice, health): + """Wraps API: + ze_result_t zesDeviceSetHealthStatusExt( + zes_device_handle_t hDevice, + zes_device_health_status_ext_t health) + + The health status is persisted to non-volatile memory. + + Parameters: + hDevice: device handle + health: zes_device_health_status_ext_t - new health status + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesDeviceSetHealthStatusExt") + funcPtr.argtypes = [zes_device_handle_t, zes_device_health_status_ext_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDevice, health) + return retVal + + +def zesDeviceEnumFirmwares(hDevice, pCount, phFirmware): + """Wraps API: + ze_result_t zesDeviceEnumFirmwares( + zes_device_handle_t hDevice, + uint32_t* pCount, + zes_firmware_handle_t* phFirmware) + + Parameters: + hDevice: device handle + pCount: POINTER(c_uint32) + phFirmware: POINTER(zes_firmware_handle_t) or None + """ + funcPtr = getFunctionPointerList("zesDeviceEnumFirmwares") + funcPtr.argtypes = [ + zes_device_handle_t, + POINTER(c_uint32), + POINTER(zes_firmware_handle_t), + ] + funcPtr.restype = ze_result_t + retVal = funcPtr(hDevice, pCount, phFirmware) + return retVal + + ## Memory management functions ## def zesDeviceEnumMemoryModules(hDevice, pCount, phMemory): """Wraps API: diff --git a/bindings/sysman/python/test/unit_tests/test_device.py b/bindings/sysman/python/test/unit_tests/test_device.py new file mode 100644 index 00000000..4f5596e8 --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_device.py @@ -0,0 +1,194 @@ +## +# 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 TestDeviceFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDeviceHandleWhenCallingZesDeviceGetStateThenCallSucceedsWithValidState( + self, mock_get_func + ): + def mock_get_state(device_handle, state_ptr): + state = state_ptr._obj + state.reset = ( + self.pyzes.ZES_RESET_REASON_FLAG_WEDGED + | self.pyzes.ZES_RESET_REASON_FLAG_REPAIR + ) + state.repaired = self.pyzes.ZES_REPAIR_STATUS_PERFORMED + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_state) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + state = self.pyzes.zes_device_state_t() + state.stype = self.pyzes.ZES_STRUCTURE_TYPE_DEVICE_STATE + + result = self.pyzes.zesDeviceGetState(device_handle, byref(state)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual( + state.reset, + self.pyzes.ZES_RESET_REASON_FLAG_WEDGED + | self.pyzes.ZES_RESET_REASON_FLAG_REPAIR, + ) + self.assertEqual(state.repaired, self.pyzes.ZES_REPAIR_STATUS_PERFORMED) + mock_get_func.assert_called_with("zesDeviceGetState") + mock_func.assert_called_once() + + def test_GivenValidDeviceHandleWhenCallingZesDeviceResetExtThenCallSucceedsWithRequestedProperties( + self, mock_get_func + ): + def mock_reset_ext(device_handle, properties_ptr): + properties = properties_ptr._obj + self.assertEqual( + properties.stype, self.pyzes.ZES_STRUCTURE_TYPE_RESET_PROPERTIES + ) + self.assertEqual(properties.force, 0) + self.assertEqual(properties.resetType, self.pyzes.ZES_RESET_TYPE_FLR) + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_reset_ext) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + properties = self.pyzes.zes_reset_properties_t() + properties.stype = self.pyzes.ZES_STRUCTURE_TYPE_RESET_PROPERTIES + properties.force = 0 + properties.resetType = self.pyzes.ZES_RESET_TYPE_FLR + + result = self.pyzes.zesDeviceResetExt(device_handle, byref(properties)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesDeviceResetExt") + mock_func.assert_called_once() + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEventRegisterThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + events = ( + self.pyzes.ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED + | self.pyzes.ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL + ) + + result = self.pyzes.zesDeviceEventRegister(device_handle, events) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesDeviceEventRegister") + mock_func.assert_called_once_with(device_handle, events) + + def test_GivenValidDeviceHandleWhenCallingZesDeviceGetHealthStatusExtThenCallSucceedsWithHealthStatus( + self, mock_get_func + ): + def mock_get_health(device_handle, health_ptr): + health_ptr._obj.value = self.pyzes.ZES_DEVICE_HEALTH_STATUS_EXT_WARNING + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_health) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + health = self.pyzes.zes_device_health_status_ext_t(0) + + result = self.pyzes.zesDeviceGetHealthStatusExt(device_handle, byref(health)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(health.value, self.pyzes.ZES_DEVICE_HEALTH_STATUS_EXT_WARNING) + mock_get_func.assert_called_with("zesDeviceGetHealthStatusExt") + mock_func.assert_called_once() + + def test_GivenValidDeviceHandleWhenCallingZesDeviceSetHealthStatusExtThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + + result = self.pyzes.zesDeviceSetHealthStatusExt( + device_handle, self.pyzes.ZES_DEVICE_HEALTH_STATUS_EXT_OK + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesDeviceSetHealthStatusExt") + mock_func.assert_called_once_with( + device_handle, self.pyzes.ZES_DEVICE_HEALTH_STATUS_EXT_OK + ) + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEnumFirmwaresThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 3 + + def mock_enum_firmwares(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_firmwares) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDeviceEnumFirmwares(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("zesDeviceEnumFirmwares") + mock_func.assert_called_once() + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEnumFirmwaresWithArrayThenCallSucceedsWithHandles( + self, mock_get_func + ): + mock_count = 2 + + def mock_enum_firmwares(device_handle, count_ptr, handles_ptr): + count_ptr._obj.value = mock_count + handles_ptr[0] = self.pyzes.zes_firmware_handle_t(0x1000) + handles_ptr[1] = self.pyzes.zes_firmware_handle_t(0x2000) + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_enum_firmwares) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + count = c_uint32(mock_count) + handles = (self.pyzes.zes_firmware_handle_t * mock_count)() + + result = self.pyzes.zesDeviceEnumFirmwares(device_handle, byref(count), handles) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual(handles[0].value, 0x1000) + self.assertEqual(handles[1].value, 0x2000) + mock_get_func.assert_called_with("zesDeviceEnumFirmwares") + mock_func.assert_called_once() + + +if __name__ == "__main__": + unittest.main() From 0c8d4a5e364267120d1bce7f3213e3918d985808 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Mon, 28 Sep 2026 14:12:18 +0530 Subject: [PATCH 2/2] fix: Set health status only when requested in pyzes black box test --- .../source/examples/pyzes_black_box_test.py | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 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 16e532ca..5b9d5159 100755 --- a/bindings/sysman/python/source/examples/pyzes_black_box_test.py +++ b/bindings/sysman/python/source/examples/pyzes_black_box_test.py @@ -983,7 +983,13 @@ def test_device_state_module(device_handle, device_index, set_health=None): else: print_verbose(f" Health Status: {get_health_status_string(health.value)}") - if is_root_user() and set_health is not None: + if set_health is None: + pass + elif not is_root_user(): + print_verbose( + " Skipping zesDeviceSetHealthStatusExt due to insufficient permissions" + ) + else: rc = pz.zesDeviceSetHealthStatusExt(device_handle, set_health) if check_rc( f"zesDeviceSetHealthStatusExt(device {device_index}, {get_health_status_string(set_health)})", @@ -1008,15 +1014,6 @@ def test_device_state_module(device_handle, device_index, set_health=None): print_verbose( f" Restored health status to {get_health_status_string(health.value)}" ) - elif is_root_user(): - # Write back the health status just read so the device state is unchanged - rc = pz.zesDeviceSetHealthStatusExt(device_handle, health.value) - if check_rc(f"zesDeviceSetHealthStatusExt(device {device_index})", rc): - print_verbose(" Set health status successfully") - else: - print_verbose( - " Skipping zesDeviceSetHealthStatusExt due to insufficient permissions" - ) events_to_register = pz.ZES_EVENT_TYPE_FLAG_DEVICE_RESET_REQUIRED rc = pz.zesDeviceEventRegister(device_handle, events_to_register)