From 80f0107127cfb52b12e7ffb5fd3ec2148c2b1c56 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:24:44 +0530 Subject: [PATCH 1/2] feature: Add pyzes support for driver APIs Add python bindings, unit tests, black box test coverage and README entries for: 1. zesDriverGetProperties 2. zesDriverGetExtensionProperties 3. zesDriverGetExtensionFunctionAddress 4. zesDriverEventListenEx 5. zesDriverEventRegisterExt 6. zesDriverEventListenExt Related-To: SYSM-433 Signed-off-by: Aviral Nigam --- bindings/sysman/python/README.md | 7 + .../source/examples/pyzes_black_box_test.py | 199 +++++++++++++++ bindings/sysman/python/source/pyzes.py | 210 ++++++++++++++++ .../python/test/unit_tests/test_driver.py | 237 ++++++++++++++++++ 4 files changed, 653 insertions(+) create mode 100644 bindings/sysman/python/test/unit_tests/test_driver.py diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..54dde551 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -158,6 +158,13 @@ SUPPORTED APIs | `zesDeviceGetProperties` | Device | 0.1.0 | None | | `zesDriverGetDeviceByUuidExp` | Device | 0.1.0 | Experimental API | | `zesDeviceProcessesGetState` | Device | 0.1.0 | None | +| **Driver Management** |-|-|-| +| `zesDriverGetProperties` | Driver | 0.1.3 | None | +| `zesDriverGetExtensionProperties` | Driver | 0.1.3 | None | +| `zesDriverGetExtensionFunctionAddress` | Driver | 0.1.3 | Returns a raw function address | +| `zesDriverEventListenEx` | Driver | 0.1.3 | None | +| `zesDriverEventRegisterExt` | Driver | 0.1.3 | None | +| `zesDriverEventListenExt` | Driver | 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..68ef55b9 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,33 @@ 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 is_root_user(): """Return whether the current user has root privileges on platforms that support it""" geteuid = getattr(os, "geteuid", None) @@ -840,6 +867,151 @@ def test_engine_modules(device_handle, device_index): return True +def print_received_events(device_count, num_device_events, events, driver_events=None): + """Print the events reported by an event listen call""" + print_verbose(f" Devices With Events: {num_device_events}") + for i in range(device_count): + if events[i] == 0: + continue + for bit in range(32): + flag = events[i] & (1 << bit) + if flag: + print_verbose( + f" Device {i} got {get_event_type_flags_string(flag)} event" + ) + + if driver_events is None: + return + if driver_events == 0: + print_verbose(" Driver Events: none") + if driver_events & pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT: + print_verbose(" Driver got INFO_LOG_CPER_DATA_AVAILABLE_EXT event") + print_verbose(" CPER records are available to read from the device info log") + other_events = ( + driver_events & ~pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT + ) + if other_events: + print_verbose( + f" Driver got unexpected events: {get_event_type_flags_string(other_events)}" + ) + + +def test_driver_module(driver_handle, driver_index, listen_timeout=0): + """Test driver properties, extensions, and event operations""" + print(f"\n---- Driver {driver_index} Driver Test ----") + + props = pz.zes_driver_properties_t() + props.stype = pz.ZES_STRUCTURE_TYPE_DRIVER_PROPERTIES + props.pNext = None + + rc = pz.zesDriverGetProperties(driver_handle, byref(props)) + if check_rc(f"zesDriverGetProperties(driver {driver_index})", rc): + print_verbose(" Properties:") + print_verbose( + f" UUID: {'-'.join(f'{props.uuid.id[i]:02x}' for i in range(16))}" + ) + print_verbose(f" Driver Version: {props.driverVersion}") + + ext_count = c_uint32(0) + rc = pz.zesDriverGetExtensionProperties(driver_handle, byref(ext_count), None) + if check_rc(f"zesDriverGetExtensionProperties(driver {driver_index}, count)", rc): + print_verbose(f" Found {ext_count.value} extension(s)") + if ext_count.value > 0: + ExtensionArray = pz.zes_driver_extension_properties_t * ext_count.value + extensions = ExtensionArray() + rc = pz.zesDriverGetExtensionProperties( + driver_handle, byref(ext_count), extensions + ) + if check_rc( + f"zesDriverGetExtensionProperties(driver {driver_index}, data)", rc + ): + for i in range(ext_count.value): + version = extensions[i].version + print_verbose( + f" {extensions[i].name.decode('utf-8', errors='replace')}: " + f"v{version >> 16}.{version & 0xFFFF}" + ) + + function_name = b"zesDriverGetDeviceByUuidExp" + function_address = c_void_p() + rc = pz.zesDriverGetExtensionFunctionAddress( + driver_handle, function_name, byref(function_address) + ) + if rc == pz.ZE_RESULT_SUCCESS: + print_verbose( + f" Extension Function Address ({function_name.decode()}): " + f"{function_address.value:#x}" + ) + else: + print_verbose( + f" Extension Function Address ({function_name.decode()}): " + f"Not available ({get_result_string(rc)})" + ) + + device_info = get_devices(driver_handle) + if device_info is None: + print_verbose(" Skipping event listen tests as no devices were found") + return True + + devices, device_count = device_info + EventsArray = pz.zes_event_type_flags_t * device_count + + # A timeout of 0 checks for pending events and returns immediately + print_verbose( + f" Listening for device events on {device_count} device(s) with a {listen_timeout} ms timeout" + ) + num_device_events = c_uint32(0) + events = EventsArray() + rc = pz.zesDriverEventListenEx( + driver_handle, + listen_timeout, + device_count, + devices, + byref(num_device_events), + events, + ) + if check_rc(f"zesDriverEventListenEx(driver {driver_index})", rc): + print_received_events(device_count, num_device_events.value, events) + + driver_events_to_register = pz.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT + rc = pz.zesDriverEventRegisterExt(driver_handle, driver_events_to_register) + if rc != pz.ZE_RESULT_SUCCESS: + print_verbose( + f" Driver Event Registration: Not available ({get_result_string(rc)})" + ) + return True + + print_verbose( + f" Registered driver events: {get_event_type_flags_string(driver_events_to_register)}" + ) + + print_verbose( + f" Listening for device and driver events with a {listen_timeout} ms timeout" + ) + num_device_events = c_uint32(0) + events = EventsArray() + driver_events = pz.zes_event_type_flags_t(0) + rc = pz.zesDriverEventListenExt( + driver_handle, + listen_timeout, + device_count, + devices, + byref(num_device_events), + events, + byref(driver_events), + ) + if check_rc(f"zesDriverEventListenExt(driver {driver_index})", rc): + print_received_events( + device_count, num_device_events.value, events, driver_events.value + ) + + # Clear the driver event registration made by this test + rc = pz.zesDriverEventRegisterExt(driver_handle, 0) + check_rc(f"zesDriverEventRegisterExt(driver {driver_index}, clear)", 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 +1567,9 @@ def run_all_tests(): # Test engine modules test_engine_modules(devices[device_idx], device_idx) + # Test driver properties, extensions, and events (driver scoped) + test_driver_module(drivers[driver_idx], driver_idx) + print("\n=== Test Completed ===") return True @@ -1413,6 +1588,8 @@ def main(): %(prog)s -f # Frequency tests only %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only + %(prog)s -d # Driver tests only + %(prog)s -d --listen-timeout 10000 # Driver tests, wait up to 10 s for events %(prog)s -h # Show help message""", formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -1450,8 +1627,20 @@ 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( + "-d", "--driver", action="store_true", help="Run only driver tests" + ) + parser.add_argument( + "--listen-timeout", + type=int, + default=0, + metavar="MS", + help="With -d, time in milliseconds to wait for events, default 0 returns immediately", + ) args = parser.parse_args() + if args.listen_timeout < 0: + parser.error("--listen-timeout must be a non-negative integer") # Check if any specific test is requested specific_test = ( @@ -1463,6 +1652,7 @@ def main(): or args.frequency or args.temperature or args.engine + or args.driver or args.all ) @@ -1489,6 +1679,15 @@ def main(): print("No devices available for testing") return 1 + # Driver tests are driver scoped, so run them once per driver + if args.driver: + for driver_idx in range(driver_count): + test_driver_module( + drivers[driver_idx], + driver_idx, + listen_timeout=args.listen_timeout, + ) + # Run selected tests on all devices for device_idx in range(device_count): if getattr(args, "global", False): diff --git a/bindings/sysman/python/source/pyzes.py b/bindings/sysman/python/source/pyzes.py index 25ab2e3d..3c4439da 100644 --- a/bindings/sysman/python/source/pyzes.py +++ b/bindings/sysman/python/source/pyzes.py @@ -262,6 +262,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 @@ -354,6 +375,7 @@ class zes_engine_handle_t(c_void_p): ZES_STRING_PROPERTY_SIZE = 64 # from zes_api.h ZE_MAX_DEVICE_NAME = 256 # from ze_api.h ZES_MAX_UUID_SIZE = 16 # from zes_api.h (uuid size for zes_uuid_t) +ZES_MAX_EXTENSION_NAME = 256 # from zes_api.h # Structure type enum values ZES_STRUCTURE_TYPE_DEVICE_PROPERTIES = 0x1 @@ -376,6 +398,7 @@ 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_DRIVER_PROPERTIES = 0x2F ## Core ze_device UUID struct ## @@ -555,6 +578,23 @@ class zes_uuid_t(_PrintableStructure): _fields_ = [("id", c_ubyte * ZES_MAX_UUID_SIZE)] +## Driver structures ## +class zes_driver_properties_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_DRIVER_PROPERTIES + ("pNext", c_void_p), + ("uuid", zes_uuid_t), # sysman driver instance UUID + ("driverVersion", c_uint32), # sysman driver version + ] + + +class zes_driver_extension_properties_t(_PrintableStructure): + _fields_ = [ + ("name", c_char * ZES_MAX_EXTENSION_NAME), # extension name + ("version", c_uint32), # extension version using ZE_MAKE_VERSION + ] + + ## Sysman zes_device_ext_properties_t (extension) ## class zes_device_ext_properties_t(_PrintableStructure): _fields_ = [ @@ -1556,3 +1596,173 @@ def zesEngineGetActivity(hEngine, pStats): retVal = funcPtr(hEngine, pStats) return retVal + + +## Driver functions ## + + +def zesDriverGetProperties(hDriver, pDriverProperties): + """Wraps API: + ze_result_t zesDriverGetProperties( + zes_driver_handle_t hDriver, + zes_driver_properties_t* pDriverProperties) + + Parameters: + hDriver: driver handle + pDriverProperties: POINTER(zes_driver_properties_t) - properties structure to fill + Returns: + ze_result_t - return code only, properties are filled into pDriverProperties + """ + funcPtr = getFunctionPointerList("zesDriverGetProperties") + funcPtr.argtypes = [zes_driver_handle_t, POINTER(zes_driver_properties_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDriver, pDriverProperties) + return retVal + + +def zesDriverGetExtensionProperties(hDriver, pCount, pExtensionProperties): + """Wraps API: + ze_result_t zesDriverGetExtensionProperties( + zes_driver_handle_t hDriver, + uint32_t* pCount, + zes_driver_extension_properties_t* pExtensionProperties) + + Parameters: + hDriver: driver handle + pCount: POINTER(c_uint32) + pExtensionProperties: POINTER(zes_driver_extension_properties_t) or None + Returns: + ze_result_t - return code only, extension properties are filled into pExtensionProperties + """ + funcPtr = getFunctionPointerList("zesDriverGetExtensionProperties") + funcPtr.argtypes = [ + zes_driver_handle_t, + POINTER(c_uint32), + POINTER(zes_driver_extension_properties_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDriver, pCount, pExtensionProperties) + return retVal + + +def zesDriverGetExtensionFunctionAddress(hDriver, name, ppFunctionAddress): + """Wraps API: + ze_result_t zesDriverGetExtensionFunctionAddress( + zes_driver_handle_t hDriver, + const char* name, + void** ppFunctionAddress) + + Parameters: + hDriver: driver handle + name: bytes - extension function name + ppFunctionAddress: POINTER(c_void_p) - function address to fill + Returns: + ze_result_t - return code only, function address is filled into ppFunctionAddress + """ + funcPtr = getFunctionPointerList("zesDriverGetExtensionFunctionAddress") + funcPtr.argtypes = [zes_driver_handle_t, c_char_p, POINTER(c_void_p)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDriver, name, ppFunctionAddress) + return retVal + + +def zesDriverEventListenEx( + hDriver, timeout, count, phDevices, pNumDeviceEvents, pEvents +): + """Wraps API: + ze_result_t zesDriverEventListenEx( + ze_driver_handle_t hDriver, + uint64_t timeout, + uint32_t count, + zes_device_handle_t* phDevices, + uint32_t* pNumDeviceEvents, + zes_event_type_flags_t* pEvents) + + Parameters: + hDriver: driver handle + timeout: c_uint64 - maximum wait in milliseconds; 0 returns immediately + count: c_uint32 - number of device handles in phDevices + phDevices: POINTER(zes_device_handle_t) - devices to listen to + pNumDeviceEvents: POINTER(c_uint32) - number of devices that generated events + pEvents: POINTER(zes_event_type_flags_t) - events for each device, at least count elements + Returns: + ze_result_t - return code only, events are filled into pEvents + """ + funcPtr = getFunctionPointerList("zesDriverEventListenEx") + funcPtr.argtypes = [ + zes_driver_handle_t, + c_uint64, + c_uint32, + POINTER(zes_device_handle_t), + POINTER(c_uint32), + POINTER(zes_event_type_flags_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDriver, timeout, count, phDevices, pNumDeviceEvents, pEvents) + return retVal + + +def zesDriverEventRegisterExt(hDriver, events): + """Wraps API: + ze_result_t zesDriverEventRegisterExt( + zes_driver_handle_t hDriver, + zes_event_type_flags_t events) + + Parameters: + hDriver: driver handle + events: zes_event_type_flags_t - driver scoped events to listen to, 0 to clear + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesDriverEventRegisterExt") + funcPtr.argtypes = [zes_driver_handle_t, zes_event_type_flags_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hDriver, events) + return retVal + + +def zesDriverEventListenExt( + hDriver, timeout, count, phDevices, pNumDeviceEvents, pEvents, pDriverEvents +): + """Wraps API: + ze_result_t zesDriverEventListenExt( + zes_driver_handle_t hDriver, + uint64_t timeout, + uint32_t count, + zes_device_handle_t* phDevices, + uint32_t* pNumDeviceEvents, + zes_event_type_flags_t* pEvents, + zes_event_type_flags_t* pDriverEvents) + + Parameters: + hDriver: driver handle + timeout: c_uint64 - maximum wait in milliseconds; 0 returns immediately + count: c_uint32 - number of device handles in phDevices + phDevices: POINTER(zes_device_handle_t) - devices to listen to + pNumDeviceEvents: POINTER(c_uint32) - number of devices that generated device scoped events + pEvents: POINTER(zes_event_type_flags_t) - device scoped events for each device + pDriverEvents: POINTER(zes_event_type_flags_t) or None - driver scoped events that occurred + Returns: + ze_result_t - return code only, events are filled into pEvents and pDriverEvents + """ + funcPtr = getFunctionPointerList("zesDriverEventListenExt") + funcPtr.argtypes = [ + zes_driver_handle_t, + c_uint64, + c_uint32, + POINTER(zes_device_handle_t), + POINTER(c_uint32), + POINTER(zes_event_type_flags_t), + POINTER(zes_event_type_flags_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr( + hDriver, timeout, count, phDevices, pNumDeviceEvents, pEvents, pDriverEvents + ) + return retVal diff --git a/bindings/sysman/python/test/unit_tests/test_driver.py b/bindings/sysman/python/test/unit_tests/test_driver.py new file mode 100644 index 00000000..91068fbb --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_driver.py @@ -0,0 +1,237 @@ +## +# 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 TestDriverFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDriverHandleWhenCallingZesDriverGetPropertiesThenCallSucceedsWithValidProperties( + self, mock_get_func + ): + def mock_get_properties(driver_handle, properties_ptr): + properties = properties_ptr._obj + properties.uuid.id[0] = 0xAB + properties.uuid.id[15] = 0xCD + properties.driverVersion = 0x10002 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_properties) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + properties = self.pyzes.zes_driver_properties_t() + properties.stype = self.pyzes.ZES_STRUCTURE_TYPE_DRIVER_PROPERTIES + + result = self.pyzes.zesDriverGetProperties(driver_handle, byref(properties)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(properties.uuid.id[0], 0xAB) + self.assertEqual(properties.uuid.id[15], 0xCD) + self.assertEqual(properties.driverVersion, 0x10002) + mock_get_func.assert_called_with("zesDriverGetProperties") + mock_func.assert_called_once() + + def test_GivenValidDriverHandleWhenCallingZesDriverGetExtensionPropertiesThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 3 + + def mock_get_extension_properties(driver_handle, count_ptr, properties_ptr): + count_ptr._obj.value = mock_count + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_extension_properties) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDriverGetExtensionProperties( + driver_handle, byref(count), None + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + mock_get_func.assert_called_with("zesDriverGetExtensionProperties") + mock_func.assert_called_once() + + def test_GivenValidDriverHandleWhenCallingZesDriverGetExtensionPropertiesWithArrayThenCallSucceedsWithProperties( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_extension_properties(driver_handle, count_ptr, properties_ptr): + count_ptr._obj.value = mock_count + properties_ptr[0].name = b"ZES_extension_info_logs" + properties_ptr[0].version = 0x10000 + properties_ptr[1].name = b"ZES_extension_power_limits" + properties_ptr[1].version = 0x10000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_extension_properties) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + count = c_uint32(mock_count) + properties = (self.pyzes.zes_driver_extension_properties_t * mock_count)() + + result = self.pyzes.zesDriverGetExtensionProperties( + driver_handle, byref(count), properties + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual(properties[0].name, b"ZES_extension_info_logs") + self.assertEqual(properties[0].version, 0x10000) + self.assertEqual(properties[1].name, b"ZES_extension_power_limits") + mock_get_func.assert_called_with("zesDriverGetExtensionProperties") + mock_func.assert_called_once() + + def test_GivenValidDriverHandleWhenCallingZesDriverGetExtensionFunctionAddressThenCallSucceedsWithAddress( + self, mock_get_func + ): + def mock_get_function_address(driver_handle, name, address_ptr): + self.assertEqual(name, b"zesDriverGetDeviceByUuidExp") + address_ptr._obj.value = 0xDEADBEEF + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_function_address) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + address = c_void_p() + + result = self.pyzes.zesDriverGetExtensionFunctionAddress( + driver_handle, b"zesDriverGetDeviceByUuidExp", byref(address) + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(address.value, 0xDEADBEEF) + mock_get_func.assert_called_with("zesDriverGetExtensionFunctionAddress") + mock_func.assert_called_once() + + def test_GivenValidDriverHandleWhenCallingZesDriverEventListenExThenCallSucceedsWithDeviceEvents( + self, mock_get_func + ): + device_count = 2 + + def mock_listen_ex( + driver_handle, timeout, count, devices, num_events_ptr, events + ): + self.assertEqual(timeout, 0) + self.assertEqual(count, device_count) + num_events_ptr._obj.value = 1 + events[0] = 0 + events[1] = self.pyzes.ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_listen_ex) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + devices = (self.pyzes.zes_device_handle_t * device_count)() + num_device_events = c_uint32(0) + events = (self.pyzes.zes_event_type_flags_t * device_count)() + + result = self.pyzes.zesDriverEventListenEx( + driver_handle, 0, device_count, devices, byref(num_device_events), events + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(num_device_events.value, 1) + self.assertEqual(events[0], 0) + self.assertEqual(events[1], self.pyzes.ZES_EVENT_TYPE_FLAG_TEMP_CRITICAL) + mock_get_func.assert_called_with("zesDriverEventListenEx") + mock_func.assert_called_once() + + def test_GivenValidDriverHandleWhenCallingZesDriverEventRegisterExtThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + events = self.pyzes.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT + + result = self.pyzes.zesDriverEventRegisterExt(driver_handle, events) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesDriverEventRegisterExt") + mock_func.assert_called_once_with(driver_handle, events) + + def test_GivenValidDriverHandleWhenCallingZesDriverEventListenExtThenCallSucceedsWithDriverEvents( + self, mock_get_func + ): + device_count = 1 + + def mock_listen_ext( + driver_handle, + timeout, + count, + devices, + num_events_ptr, + events, + driver_events_ptr, + ): + self.assertEqual(timeout, 0) + self.assertEqual(count, device_count) + num_events_ptr._obj.value = 0 + events[0] = 0 + driver_events_ptr._obj.value = ( + self.pyzes.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT + ) + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_listen_ext) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + devices = (self.pyzes.zes_device_handle_t * device_count)() + num_device_events = c_uint32(0) + events = (self.pyzes.zes_event_type_flags_t * device_count)() + driver_events = self.pyzes.zes_event_type_flags_t(0) + + result = self.pyzes.zesDriverEventListenExt( + driver_handle, + 0, + device_count, + devices, + byref(num_device_events), + events, + byref(driver_events), + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(num_device_events.value, 0) + self.assertEqual(events[0], 0) + self.assertEqual( + driver_events.value, + self.pyzes.ZES_EVENT_TYPE_FLAG_INFO_LOG_CPER_DATA_AVAILABLE_EXT, + ) + mock_get_func.assert_called_with("zesDriverEventListenExt") + mock_func.assert_called_once() + + +if __name__ == "__main__": + unittest.main() From 69aff0e03d0499059cdaec89939e73c7c3ba2ae9 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Mon, 28 Sep 2026 14:09:56 +0530 Subject: [PATCH 2/2] fix: Query a driver exposed extension function and default the event listen timeout to 10000 ms in pyzes black box test --- .../python/source/examples/pyzes_black_box_test.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 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 68ef55b9..205c1e8f 100755 --- a/bindings/sysman/python/source/examples/pyzes_black_box_test.py +++ b/bindings/sysman/python/source/examples/pyzes_black_box_test.py @@ -932,7 +932,8 @@ def test_driver_module(driver_handle, driver_index, listen_timeout=0): f"v{version >> 16}.{version & 0xFFFF}" ) - function_name = b"zesDriverGetDeviceByUuidExp" + # Only the address of this extension function is queried, the function is not called + function_name = b"zesIntelDriverGetPropertiesExp" function_address = c_void_p() rc = pz.zesDriverGetExtensionFunctionAddress( driver_handle, function_name, byref(function_address) @@ -1589,7 +1590,7 @@ def main(): %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only %(prog)s -d # Driver tests only - %(prog)s -d --listen-timeout 10000 # Driver tests, wait up to 10 s for events + %(prog)s -d --listen-timeout 0 # Driver tests, return from event listen immediately %(prog)s -h # Show help message""", formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -1633,9 +1634,9 @@ def main(): parser.add_argument( "--listen-timeout", type=int, - default=0, + default=10000, metavar="MS", - help="With -d, time in milliseconds to wait for events, default 0 returns immediately", + help="With -d, time in milliseconds to wait for events, default 10000, 0 returns immediately", ) args = parser.parse_args()