From fbd1a414084b8c21769c2548ff5e3942cf3ea4a2 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:21:44 +0530 Subject: [PATCH] feature: Add pyzes support for info log APIs Add python bindings, unit tests, black box test coverage and README entries for: 1. zesDriverEnumInfoLogsExt 2. zesInfoLogGetPropertiesExt 3. zesInfoLogCreateInstanceExt 4. zesInfoLogInstanceReadWithMetadataExt 5. zesInfoLogInstancePeekWithMetadataExt 6. zesInfoLogInstanceDeleteExt Related-To: SYSM-432 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 | 269 +++++++++++++ .../python/test/unit_tests/test_info_log.py | 359 ++++++++++++++++++ 4 files changed, 834 insertions(+) create mode 100644 bindings/sysman/python/test/unit_tests/test_info_log.py diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..bbf738b6 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -195,6 +195,13 @@ SUPPORTED APIs | `zesDeviceEccConfigurable` | ECC | 0.1.2 | None | | `zesDeviceGetEccState` | ECC | 0.1.2 | None | | `zesDeviceSetEccState` | ECC | 0.1.2 | None | +| **Info Log Management** |-|-|-| +| `zesDriverEnumInfoLogsExt` | Info Log | 0.1.3 | None | +| `zesInfoLogGetPropertiesExt` | Info Log | 0.1.3 | None | +| `zesInfoLogCreateInstanceExt` | Info Log | 0.1.3 | None | +| `zesInfoLogInstanceReadWithMetadataExt` | Info Log | 0.1.3 | Records returned are consumed | +| `zesInfoLogInstancePeekWithMetadataExt` | Info Log | 0.1.3 | Requires `isPeekDataSupported` | +| `zesInfoLogInstanceDeleteExt` | Info Log | 0.1.3 | None | 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..ad7a5f56 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,36 @@ def get_device_action_string(action): return action_map.get(action, f"UNKNOWN_DEVICE_ACTION_{action}") +def get_info_log_type_string(info_log_type): + """Convert info log type enum to string""" + type_map = { + pz.ZES_INFO_LOG_TYPE_EXT_DEVICE: "ZES_INFO_LOG_TYPE_EXT_DEVICE", + } + return type_map.get(info_log_type, f"UNKNOWN_INFO_LOG_TYPE_{info_log_type}") + + +def get_info_log_format_string(info_log_format): + """Convert info log format enum to string""" + format_map = { + pz.ZES_INFO_LOG_FORMAT_EXT_CPER: "ZES_INFO_LOG_FORMAT_EXT_CPER", + } + return format_map.get(info_log_format, f"UNKNOWN_INFO_LOG_FORMAT_{info_log_format}") + + +def get_info_log_record_type_string(record_type): + """Convert info log record type enum to string""" + record_type_map = { + pz.ZES_INFO_LOG_RECORD_TYPE_EXT_UNKNOWN: "ZES_INFO_LOG_RECORD_TYPE_EXT_UNKNOWN", + pz.ZES_INFO_LOG_RECORD_TYPE_EXT_INFORMATIONAL: "ZES_INFO_LOG_RECORD_TYPE_EXT_INFORMATIONAL", + pz.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED: "ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED", + pz.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_RECOVERABLE: "ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_RECOVERABLE", + pz.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL: "ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL", + } + return record_type_map.get( + record_type, f"UNKNOWN_INFO_LOG_RECORD_TYPE_{record_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 +870,162 @@ def test_engine_modules(device_handle, device_index): return True +INFO_LOG_READ_TIMEOUT_MS = 1000 + + +def test_info_log_module(driver_handle, driver_index): + """Test info log enumeration, properties, and collection instance operations""" + print(f"\n---- Driver {driver_index} Info Log Test ----") + + info_log_count = c_uint32(0) + rc = pz.zesDriverEnumInfoLogsExt(driver_handle, byref(info_log_count), None) + if not check_rc(f"zesDriverEnumInfoLogsExt(driver {driver_index}, count)", rc): + return False + + if info_log_count.value == 0: + print_verbose("No info logs found on this driver") + return True + + print_verbose(f"Found {info_log_count.value} info log(s)") + + InfoLogArray = pz.zes_info_log_handle_t * info_log_count.value + info_log_handles = InfoLogArray() + + rc = pz.zesDriverEnumInfoLogsExt( + driver_handle, byref(info_log_count), info_log_handles + ) + if not check_rc(f"zesDriverEnumInfoLogsExt(driver {driver_index}, handles)", rc): + return False + + for i in range(info_log_count.value): + print_verbose(f"\n Info Log {i}:") + + props = pz.zes_info_log_ext_properties_t() + props.stype = pz.ZES_STRUCTURE_TYPE_INFO_LOG_EXT_PROPERTIES + props.pNext = None + + rc = pz.zesInfoLogGetPropertiesExt(info_log_handles[i], byref(props)) + if not check_rc(f"zesInfoLogGetPropertiesExt(info log {i})", rc): + continue + + print_verbose(f" Type: {get_info_log_type_string(props.infoLogType)}") + print_verbose(f" Format: {get_info_log_format_string(props.infoLogFormat)}") + print_verbose( + f" Named Instance Supported: {bool(props.isNamedInstanceSupported)}" + ) + print_verbose(f" Peek Data Supported: {bool(props.isPeekDataSupported)}") + + # Use the default buffer and default buffer size + desc = pz.zes_info_log_instance_ext_desc_t() + desc.stype = pz.ZES_STRUCTURE_TYPE_INFO_LOG_INSTANCE_EXT_DESC + desc.pNext = None + desc.pBufferSizeInKb = None + + instance_handle = pz.zes_info_log_instance_handle_t() + rc = pz.zesInfoLogCreateInstanceExt( + info_log_handles[i], None, byref(desc), byref(instance_handle) + ) + if rc != pz.ZE_RESULT_SUCCESS: + print_verbose( + f" Collection Instance: Not available ({get_result_string(rc)})" + ) + continue + + print_verbose(" Created collection instance successfully") + + # Only a query call is made, so records in the shared default buffer are not consumed + size = c_uint32(0) + record_count = c_uint32(0) + rc = pz.zesInfoLogInstanceReadWithMetadataExt( + instance_handle, + INFO_LOG_READ_TIMEOUT_MS, + byref(size), + None, + byref(record_count), + None, + None, + ) + if check_rc(f"zesInfoLogInstanceReadWithMetadataExt(info log {i}, query)", rc): + print_verbose( + f" Read Query: {record_count.value} record(s), {size.value} bytes" + ) + + if props.isPeekDataSupported: + test_info_log_peek(instance_handle, i) + + rc = pz.zesInfoLogInstanceDeleteExt(instance_handle) + if check_rc(f"zesInfoLogInstanceDeleteExt(info log {i})", rc): + print_verbose(" Deleted collection instance successfully") + + return True + + +def test_info_log_peek(instance_handle, info_log_index): + """Peek at the records held by an info log collection instance without consuming them""" + size = c_uint32(0) + record_count = c_uint32(0) + rc = pz.zesInfoLogInstancePeekWithMetadataExt( + instance_handle, + INFO_LOG_READ_TIMEOUT_MS, + byref(size), + None, + byref(record_count), + None, + None, + ) + if not check_rc( + f"zesInfoLogInstancePeekWithMetadataExt(info log {info_log_index}, query)", rc + ): + return + + print_verbose(f" Peek Query: {record_count.value} record(s), {size.value} bytes") + if size.value == 0 or record_count.value == 0: + return + + buffer = (c_uint8 * size.value)() + descriptors = (pz.zes_info_log_metadata_ext_t * record_count.value)() + for j in range(record_count.value): + descriptors[j].stype = pz.ZES_STRUCTURE_TYPE_INFO_LOG_METADATA_EXT + descriptors[j].pNext = None + + status = pz.zes_info_log_read_status_ext_t() + status.stype = pz.ZES_STRUCTURE_TYPE_INFO_LOG_READ_STATUS_EXT + status.pNext = None + + rc = pz.zesInfoLogInstancePeekWithMetadataExt( + instance_handle, + INFO_LOG_READ_TIMEOUT_MS, + byref(size), + buffer, + byref(record_count), + descriptors, + byref(status), + ) + if rc == pz.ZE_RESULT_WARNING_DROPPED_DATA: + print_verbose(f" Dropped Records: {status.droppedRecordCount}") + elif not check_rc( + f"zesInfoLogInstancePeekWithMetadataExt(info log {info_log_index}, data)", rc + ): + return + + print_verbose(f" Peeked {record_count.value} record(s), {size.value} bytes:") + for j in range(record_count.value): + record = descriptors[j] + print_verbose(f" Record {j}:") + print_verbose( + f" Type: {get_info_log_record_type_string(record.recordType)}" + ) + print_verbose( + f" BDF: {record.address.domain:04X}:{record.address.bus:02X}:" + f"{record.address.device:02X}.{record.address.function:X}" + ) + print_verbose(f" Timestamp: {record.timestamp} ns") + print_verbose(f" Offset: {record.offset}") + print_verbose(f" Length: {record.lengthOfData} bytes") + print_verbose(f" Consumed Data Size: {status.consumedDataSize} bytes") + print_verbose(f" Has Data To Read: {bool(status.hasDataToRead)}") + + 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 +1581,9 @@ def run_all_tests(): # Test engine modules test_engine_modules(devices[device_idx], device_idx) + # Test info logs (driver scoped) + test_info_log_module(drivers[driver_idx], driver_idx) + print("\n=== Test Completed ===") return True @@ -1413,6 +1602,7 @@ def main(): %(prog)s -f # Frequency tests only %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only + %(prog)s -l # Info log tests only %(prog)s -h # Show help message""", formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -1450,6 +1640,9 @@ 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( + "-l", "--infolog", action="store_true", help="Run only info log tests" + ) args = parser.parse_args() @@ -1463,6 +1656,7 @@ def main(): or args.frequency or args.temperature or args.engine + or args.infolog or args.all ) @@ -1489,6 +1683,11 @@ def main(): print("No devices available for testing") return 1 + # Info logs are driver scoped, so run them once per driver + if args.infolog: + for driver_idx in range(driver_count): + test_info_log_module(drivers[driver_idx], driver_idx) + # 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..f3a484b9 100644 --- a/bindings/sysman/python/source/pyzes.py +++ b/bindings/sysman/python/source/pyzes.py @@ -148,6 +148,14 @@ class zes_engine_handle_t(c_void_p): pass +class zes_info_log_handle_t(c_void_p): + pass + + +class zes_info_log_instance_handle_t(c_void_p): + pass + + ## ze_bool_t = c_uint8 @@ -294,6 +302,23 @@ class zes_engine_handle_t(c_void_p): ZES_ENGINE_GROUP_MEDIA_CODEC_SINGLE = 14 ZES_ENGINE_GROUP_FORCE_UINT32 = 0x7FFFFFFF +## Info log enums ## +zes_info_log_type_ext_t = c_int32 +ZES_INFO_LOG_TYPE_EXT_DEVICE = 0 +ZES_INFO_LOG_TYPE_EXT_FORCE_UINT32 = 0x7FFFFFFF + +zes_info_log_format_ext_t = c_int32 +ZES_INFO_LOG_FORMAT_EXT_CPER = 0 +ZES_INFO_LOG_FORMAT_EXT_FORCE_UINT32 = 0x7FFFFFFF + +zes_info_log_record_type_ext_t = c_int32 +ZES_INFO_LOG_RECORD_TYPE_EXT_UNKNOWN = 0 +ZES_INFO_LOG_RECORD_TYPE_EXT_INFORMATIONAL = 1 +ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED = 2 +ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_RECOVERABLE = 3 +ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL = 4 +ZES_INFO_LOG_RECORD_TYPE_EXT_FORCE_UINT32 = 0x7FFFFFFF + ze_result_t = c_int32 ZE_RESULT_SUCCESS = 0 ZE_RESULT_NOT_READY = 1 @@ -376,6 +401,12 @@ 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_INFO_LOG_EXT_PROPERTIES = 0x00020020 # Info log properties +ZES_STRUCTURE_TYPE_INFO_LOG_INSTANCE_EXT_DESC = ( + 0x00020021 # Info log instance descriptor +) +ZES_STRUCTURE_TYPE_INFO_LOG_METADATA_EXT = 0x00020022 # Info log record metadata +ZES_STRUCTURE_TYPE_INFO_LOG_READ_STATUS_EXT = 0x00020023 # Info log read status ## Core ze_device UUID struct ## @@ -779,6 +810,51 @@ class zes_engine_stats_t(_PrintableStructure): _fmt_ = {"activeTime": "%d", "timestamp": "%d"} +## Info log structures ## +class zes_info_log_ext_properties_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_INFO_LOG_EXT_PROPERTIES + ("pNext", c_void_p), + ("infoLogType", zes_info_log_type_ext_t), # type of info carried + ("infoLogFormat", zes_info_log_format_ext_t), # format of records + ("isNamedInstanceSupported", ze_bool_t), # named instances supported + ("isPeekDataSupported", ze_bool_t), # peek without consuming supported + ] + + +class zes_info_log_instance_ext_desc_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_INFO_LOG_INSTANCE_EXT_DESC + ("pNext", c_void_p), + ("pBufferSizeInKb", POINTER(c_uint32)), # optional buffer size in KB + ] + + +class zes_info_log_metadata_ext_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_INFO_LOG_METADATA_EXT + ("pNext", c_void_p), + ("address", zes_pci_address_t), # PCI address of originating device + ("uuid", zes_uuid_t), # UUID of originating device + ("timestamp", c_uint64), # record time in nanoseconds + ("lengthOfData", c_uint32), # record length in bytes + ("offset", c_uint32), # record byte offset in data buffer + ("recordType", zes_info_log_record_type_ext_t), # record type + ] + _fmt_ = {"timestamp": "%d ns", "lengthOfData": "%d bytes"} + + +class zes_info_log_read_status_ext_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_INFO_LOG_READ_STATUS_EXT + ("pNext", c_void_p), + ("droppedRecordCount", c_int64), # dropped records, -1 if unknown + ("consumedDataSize", c_uint64), # bytes read out of the instance + ("hasDataToRead", ze_bool_t), # records remain to be read + ] + _fmt_ = {"consumedDataSize": "%d bytes"} + + ## Function access ## _getFunctionPointerList: Dict[str, Any] = dict() @@ -1556,3 +1632,196 @@ def zesEngineGetActivity(hEngine, pStats): retVal = funcPtr(hEngine, pStats) return retVal + + +## Info log functions ## + + +def zesDriverEnumInfoLogsExt(hDriver, pCount, phInfoLogs): + """Wraps API: + ze_result_t zesDriverEnumInfoLogsExt( + zes_driver_handle_t hDriver, + uint32_t* pCount, + zes_info_log_handle_t* phInfoLogs) + + Parameters: + hDriver: driver handle + pCount: POINTER(c_uint32) + phInfoLogs: POINTER(zes_info_log_handle_t) or None + """ + funcPtr = getFunctionPointerList("zesDriverEnumInfoLogsExt") + funcPtr.argtypes = [ + zes_driver_handle_t, + POINTER(c_uint32), + POINTER(zes_info_log_handle_t), + ] + funcPtr.restype = ze_result_t + retVal = funcPtr(hDriver, pCount, phInfoLogs) + return retVal + + +def zesInfoLogGetPropertiesExt(hInfoLog, pProperties): + """Wraps API: + ze_result_t zesInfoLogGetPropertiesExt( + zes_info_log_handle_t hInfoLog, + zes_info_log_ext_properties_t* pProperties) + + Parameters: + hInfoLog: info log handle + pProperties: POINTER(zes_info_log_ext_properties_t) - properties structure to fill + Returns: + ze_result_t - return code only, properties are filled into pProperties + """ + funcPtr = getFunctionPointerList("zesInfoLogGetPropertiesExt") + funcPtr.argtypes = [zes_info_log_handle_t, POINTER(zes_info_log_ext_properties_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hInfoLog, pProperties) + return retVal + + +def zesInfoLogCreateInstanceExt(hInfoLog, pInstanceName, pDesc, phInfoLogInstance): + """Wraps API: + ze_result_t zesInfoLogCreateInstanceExt( + zes_info_log_handle_t hInfoLog, + const char* pInstanceName, + zes_info_log_instance_ext_desc_t* pDesc, + zes_info_log_instance_handle_t* phInfoLogInstance) + + Parameters: + hInfoLog: info log handle + pInstanceName: bytes or None - name of the instance; None uses the default buffer + pDesc: POINTER(zes_info_log_instance_ext_desc_t) - instance descriptor + phInfoLogInstance: POINTER(zes_info_log_instance_handle_t) - created instance handle + Returns: + ze_result_t - return code only, instance handle is filled into phInfoLogInstance + """ + funcPtr = getFunctionPointerList("zesInfoLogCreateInstanceExt") + funcPtr.argtypes = [ + zes_info_log_handle_t, + c_char_p, + POINTER(zes_info_log_instance_ext_desc_t), + POINTER(zes_info_log_instance_handle_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hInfoLog, pInstanceName, pDesc, phInfoLogInstance) + return retVal + + +def zesInfoLogInstanceReadWithMetadataExt( + hInfoLogInstance, timeout, pSize, pBuffer, pRecordCount, pDescriptors, pReadStatus +): + """Wraps API: + ze_result_t zesInfoLogInstanceReadWithMetadataExt( + zes_info_log_instance_handle_t hInfoLogInstance, + uint64_t timeout, + uint32_t* pSize, + uint8_t* pBuffer, + uint32_t* pRecordCount, + zes_info_log_metadata_ext_t* pDescriptors, + zes_info_log_read_status_ext_t* pReadStatus) + + Records returned are consumed and are not returned again. + + Parameters: + hInfoLogInstance: info log collection instance handle + timeout: c_uint64 - maximum search time in milliseconds + pSize: POINTER(c_uint32) - size of pBuffer in bytes; 0 makes this a query call + pBuffer: POINTER(c_uint8) or None - buffer for record data + pRecordCount: POINTER(c_uint32) - number of elements in pDescriptors; 0 makes this a query call + pDescriptors: POINTER(zes_info_log_metadata_ext_t) or None - metadata for each record + pReadStatus: POINTER(zes_info_log_read_status_ext_t) or None - status of this call + Returns: + ze_result_t - return code only, records are filled into pBuffer and pDescriptors + """ + funcPtr = getFunctionPointerList("zesInfoLogInstanceReadWithMetadataExt") + funcPtr.argtypes = [ + zes_info_log_instance_handle_t, + c_uint64, + POINTER(c_uint32), + POINTER(c_uint8), + POINTER(c_uint32), + POINTER(zes_info_log_metadata_ext_t), + POINTER(zes_info_log_read_status_ext_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr( + hInfoLogInstance, + timeout, + pSize, + pBuffer, + pRecordCount, + pDescriptors, + pReadStatus, + ) + return retVal + + +def zesInfoLogInstancePeekWithMetadataExt( + hInfoLogInstance, timeout, pSize, pBuffer, pRecordCount, pDescriptors, pReadStatus +): + """Wraps API: + ze_result_t zesInfoLogInstancePeekWithMetadataExt( + zes_info_log_instance_handle_t hInfoLogInstance, + uint64_t timeout, + uint32_t* pSize, + uint8_t* pBuffer, + uint32_t* pRecordCount, + zes_info_log_metadata_ext_t* pDescriptors, + zes_info_log_read_status_ext_t* pReadStatus) + + Records returned are not consumed. + + Parameters: + hInfoLogInstance: info log collection instance handle + timeout: c_uint64 - maximum search time in milliseconds + pSize: POINTER(c_uint32) - size of pBuffer in bytes; 0 makes this a query call + pBuffer: POINTER(c_uint8) or None - buffer for record data + pRecordCount: POINTER(c_uint32) - number of elements in pDescriptors; 0 makes this a query call + pDescriptors: POINTER(zes_info_log_metadata_ext_t) or None - metadata for each record + pReadStatus: POINTER(zes_info_log_read_status_ext_t) or None - status of this call + Returns: + ze_result_t - return code only, records are filled into pBuffer and pDescriptors + """ + funcPtr = getFunctionPointerList("zesInfoLogInstancePeekWithMetadataExt") + funcPtr.argtypes = [ + zes_info_log_instance_handle_t, + c_uint64, + POINTER(c_uint32), + POINTER(c_uint8), + POINTER(c_uint32), + POINTER(zes_info_log_metadata_ext_t), + POINTER(zes_info_log_read_status_ext_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr( + hInfoLogInstance, + timeout, + pSize, + pBuffer, + pRecordCount, + pDescriptors, + pReadStatus, + ) + return retVal + + +def zesInfoLogInstanceDeleteExt(hInfoLogInstance): + """Wraps API: + ze_result_t zesInfoLogInstanceDeleteExt( + zes_info_log_instance_handle_t hInfoLogInstance) + + Parameters: + hInfoLogInstance: info log collection instance handle to delete + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesInfoLogInstanceDeleteExt") + funcPtr.argtypes = [zes_info_log_instance_handle_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hInfoLogInstance) + return retVal diff --git a/bindings/sysman/python/test/unit_tests/test_info_log.py b/bindings/sysman/python/test/unit_tests/test_info_log.py new file mode 100644 index 00000000..7947cd24 --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_info_log.py @@ -0,0 +1,359 @@ +## +# 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 TestInfoLogFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDriverHandleWhenCallingZesDriverEnumInfoLogsExtThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 1 + + def mock_enum_info_logs(driver_handle, count_ptr, handles_ptr): + count_ptr._obj.value = mock_count + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_enum_info_logs) + mock_get_func.return_value = mock_func + + driver_handle = self.pyzes.zes_driver_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDriverEnumInfoLogsExt(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("zesDriverEnumInfoLogsExt") + mock_func.assert_called_once() + + def test_GivenValidInfoLogHandleWhenCallingZesInfoLogGetPropertiesExtThenCallSucceedsWithValidProperties( + self, mock_get_func + ): + def mock_get_properties(info_log_handle, properties_ptr): + properties = properties_ptr._obj + properties.infoLogType = self.pyzes.ZES_INFO_LOG_TYPE_EXT_DEVICE + properties.infoLogFormat = self.pyzes.ZES_INFO_LOG_FORMAT_EXT_CPER + properties.isNamedInstanceSupported = 1 + properties.isPeekDataSupported = 1 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_properties) + mock_get_func.return_value = mock_func + + info_log_handle = self.pyzes.zes_info_log_handle_t() + properties = self.pyzes.zes_info_log_ext_properties_t() + properties.stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_EXT_PROPERTIES + + result = self.pyzes.zesInfoLogGetPropertiesExt( + info_log_handle, byref(properties) + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual( + properties.infoLogType, self.pyzes.ZES_INFO_LOG_TYPE_EXT_DEVICE + ) + self.assertEqual( + properties.infoLogFormat, self.pyzes.ZES_INFO_LOG_FORMAT_EXT_CPER + ) + self.assertEqual(properties.isNamedInstanceSupported, 1) + self.assertEqual(properties.isPeekDataSupported, 1) + mock_get_func.assert_called_with("zesInfoLogGetPropertiesExt") + mock_func.assert_called_once() + + def test_GivenValidInfoLogHandleWhenCallingZesInfoLogCreateInstanceExtThenCallSucceedsWithInstanceHandle( + self, mock_get_func + ): + def mock_create_instance(info_log_handle, name, desc_ptr, instance_ptr): + self.assertEqual(name, b"pyzes_instance") + desc = desc_ptr._obj + self.assertEqual( + desc.stype, self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_INSTANCE_EXT_DESC + ) + # Driver reports the buffer size it actually applied + desc.pBufferSizeInKb[0] = 64 + instance_ptr._obj.value = 0x1234 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_create_instance) + mock_get_func.return_value = mock_func + + info_log_handle = self.pyzes.zes_info_log_handle_t() + buffer_size_kb = c_uint32(0) + desc = self.pyzes.zes_info_log_instance_ext_desc_t() + desc.stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_INSTANCE_EXT_DESC + desc.pBufferSizeInKb = pointer(buffer_size_kb) + instance_handle = self.pyzes.zes_info_log_instance_handle_t() + + result = self.pyzes.zesInfoLogCreateInstanceExt( + info_log_handle, b"pyzes_instance", byref(desc), byref(instance_handle) + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(buffer_size_kb.value, 64) + self.assertEqual(instance_handle.value, 0x1234) + mock_get_func.assert_called_with("zesInfoLogCreateInstanceExt") + mock_func.assert_called_once() + + def test_GivenValidInstanceHandleWhenCallingZesInfoLogInstanceReadWithMetadataExtWithZeroSizeThenQueryReturnsSizeAndRecordCount( + self, mock_get_func + ): + def mock_query( + instance_handle, + timeout, + size_ptr, + buffer, + count_ptr, + descriptors, + status_ptr, + ): + self.assertEqual(timeout, 1000) + self.assertIsNone(buffer) + self.assertIsNone(descriptors) + size_ptr._obj.value = 512 + count_ptr._obj.value = 2 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_query) + mock_get_func.return_value = mock_func + + instance_handle = self.pyzes.zes_info_log_instance_handle_t() + size = c_uint32(0) + record_count = c_uint32(0) + + result = self.pyzes.zesInfoLogInstanceReadWithMetadataExt( + instance_handle, 1000, byref(size), None, byref(record_count), None, None + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(size.value, 512) + self.assertEqual(record_count.value, 2) + mock_get_func.assert_called_with("zesInfoLogInstanceReadWithMetadataExt") + mock_func.assert_called_once() + + def test_GivenValidInstanceHandleWhenCallingZesInfoLogInstanceReadWithMetadataExtWithBuffersThenRecordsAndStatusAreReturned( + self, mock_get_func + ): + def mock_read( + instance_handle, + timeout, + size_ptr, + buffer, + count_ptr, + descriptors, + status_ptr, + ): + buffer[0] = 0x43 + buffer[300] = 0x50 + descriptors[0].lengthOfData = 300 + descriptors[0].offset = 0 + descriptors[0].timestamp = 123456789 + descriptors[0].recordType = ( + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED + ) + descriptors[0].address.bus = 0x3A + descriptors[1].lengthOfData = 212 + descriptors[1].offset = 300 + descriptors[1].recordType = ( + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL + ) + status = status_ptr._obj + status.droppedRecordCount = 0 + status.consumedDataSize = 4096 + status.hasDataToRead = 0 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_read) + mock_get_func.return_value = mock_func + + instance_handle = self.pyzes.zes_info_log_instance_handle_t() + size = c_uint32(512) + record_count = c_uint32(2) + buffer = (c_uint8 * size.value)() + descriptors = (self.pyzes.zes_info_log_metadata_ext_t * record_count.value)() + for i in range(record_count.value): + descriptors[i].stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_METADATA_EXT + status = self.pyzes.zes_info_log_read_status_ext_t() + status.stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_READ_STATUS_EXT + + result = self.pyzes.zesInfoLogInstanceReadWithMetadataExt( + instance_handle, + 1000, + byref(size), + buffer, + byref(record_count), + descriptors, + byref(status), + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(buffer[0], 0x43) + self.assertEqual(buffer[300], 0x50) + self.assertEqual(descriptors[0].lengthOfData, 300) + self.assertEqual(descriptors[0].timestamp, 123456789) + self.assertEqual( + descriptors[0].recordType, + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED, + ) + self.assertEqual(descriptors[0].address.bus, 0x3A) + self.assertEqual(descriptors[1].offset, 300) + self.assertEqual( + descriptors[1].recordType, + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL, + ) + self.assertEqual(status.droppedRecordCount, 0) + self.assertEqual(status.consumedDataSize, 4096) + self.assertEqual(status.hasDataToRead, 0) + mock_get_func.assert_called_with("zesInfoLogInstanceReadWithMetadataExt") + mock_func.assert_called_once() + + def test_GivenValidInstanceHandleWhenCallingZesInfoLogInstancePeekWithMetadataExtWithZeroSizeThenQueryReturnsSizeAndRecordCount( + self, mock_get_func + ): + def mock_query( + instance_handle, + timeout, + size_ptr, + buffer, + count_ptr, + descriptors, + status_ptr, + ): + self.assertEqual(timeout, 1000) + self.assertIsNone(buffer) + self.assertIsNone(descriptors) + size_ptr._obj.value = 512 + count_ptr._obj.value = 2 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_query) + mock_get_func.return_value = mock_func + + instance_handle = self.pyzes.zes_info_log_instance_handle_t() + size = c_uint32(0) + record_count = c_uint32(0) + + result = self.pyzes.zesInfoLogInstancePeekWithMetadataExt( + instance_handle, 1000, byref(size), None, byref(record_count), None, None + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(size.value, 512) + self.assertEqual(record_count.value, 2) + mock_get_func.assert_called_with("zesInfoLogInstancePeekWithMetadataExt") + mock_func.assert_called_once() + + def test_GivenValidInstanceHandleWhenCallingZesInfoLogInstancePeekWithMetadataExtWithBuffersThenRecordsAndStatusAreReturned( + self, mock_get_func + ): + def mock_read( + instance_handle, + timeout, + size_ptr, + buffer, + count_ptr, + descriptors, + status_ptr, + ): + buffer[0] = 0x43 + buffer[300] = 0x50 + descriptors[0].lengthOfData = 300 + descriptors[0].offset = 0 + descriptors[0].timestamp = 123456789 + descriptors[0].recordType = ( + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED + ) + descriptors[0].address.bus = 0x3A + descriptors[1].lengthOfData = 212 + descriptors[1].offset = 300 + descriptors[1].recordType = ( + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL + ) + status = status_ptr._obj + status.droppedRecordCount = 0 + status.consumedDataSize = 4096 + status.hasDataToRead = 0 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_read) + mock_get_func.return_value = mock_func + + instance_handle = self.pyzes.zes_info_log_instance_handle_t() + size = c_uint32(512) + record_count = c_uint32(2) + buffer = (c_uint8 * size.value)() + descriptors = (self.pyzes.zes_info_log_metadata_ext_t * record_count.value)() + for i in range(record_count.value): + descriptors[i].stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_METADATA_EXT + status = self.pyzes.zes_info_log_read_status_ext_t() + status.stype = self.pyzes.ZES_STRUCTURE_TYPE_INFO_LOG_READ_STATUS_EXT + + result = self.pyzes.zesInfoLogInstancePeekWithMetadataExt( + instance_handle, + 1000, + byref(size), + buffer, + byref(record_count), + descriptors, + byref(status), + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(buffer[0], 0x43) + self.assertEqual(buffer[300], 0x50) + self.assertEqual(descriptors[0].lengthOfData, 300) + self.assertEqual(descriptors[0].timestamp, 123456789) + self.assertEqual( + descriptors[0].recordType, + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_CORRECTED, + ) + self.assertEqual(descriptors[0].address.bus, 0x3A) + self.assertEqual(descriptors[1].offset, 300) + self.assertEqual( + descriptors[1].recordType, + self.pyzes.ZES_INFO_LOG_RECORD_TYPE_EXT_ERROR_FATAL, + ) + self.assertEqual(status.droppedRecordCount, 0) + self.assertEqual(status.consumedDataSize, 4096) + self.assertEqual(status.hasDataToRead, 0) + mock_get_func.assert_called_with("zesInfoLogInstancePeekWithMetadataExt") + mock_func.assert_called_once() + + def test_GivenValidInstanceHandleWhenCallingZesInfoLogInstanceDeleteExtThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + instance_handle = self.pyzes.zes_info_log_instance_handle_t(0x1234) + + result = self.pyzes.zesInfoLogInstanceDeleteExt(instance_handle) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesInfoLogInstanceDeleteExt") + mock_func.assert_called_once_with(instance_handle) + + +if __name__ == "__main__": + unittest.main()