From c92d98ff1648d677a0cdd03e3a4177ead0c2b244 Mon Sep 17 00:00:00 2001 From: Aviral Nigam Date: Wed, 23 Sep 2026 11:18:06 +0530 Subject: [PATCH] feature: Add pyzes support for VF management, temperature config, engine activity ext and standby APIs Add python bindings, unit tests, black box test coverage and README entries for: 1. zesDeviceEnumEnabledVFExp 2. zesVFManagementGetVFCapabilitiesExp2 3. zesVFManagementGetVFEngineUtilizationExp2 4. zesVFManagementGetVFMemoryUtilizationExp2 5. zesTemperatureSetConfig 6. zesEngineGetActivityExt 7. zesDeviceEnumStandbyDomains 8. zesStandbyGetProperties 9. zesStandbyGetMode 10. zesStandbySetMode Related-To: SYSM-431 Signed-off-by: Aviral Nigam --- bindings/sysman/python/README.md | 12 + .../source/examples/pyzes_black_box_test.py | 265 ++++++++++++++++ bindings/sysman/python/source/pyzes.py | 298 ++++++++++++++++++ .../python/test/unit_tests/test_engine.py | 53 ++++ .../python/test/unit_tests/test_standby.py | 119 +++++++ .../test/unit_tests/test_temperature.py | 31 ++ .../test/unit_tests/test_vf_management.py | 162 ++++++++++ 7 files changed, 940 insertions(+) create mode 100644 bindings/sysman/python/test/unit_tests/test_standby.py create mode 100644 bindings/sysman/python/test/unit_tests/test_vf_management.py diff --git a/bindings/sysman/python/README.md b/bindings/sysman/python/README.md index 8bafe655..a55c0550 100644 --- a/bindings/sysman/python/README.md +++ b/bindings/sysman/python/README.md @@ -182,10 +182,22 @@ SUPPORTED APIs | `zesTemperatureGetProperties` | Temperature | 0.1.0 | None | | `zesTemperatureGetConfig` | Temperature | 0.1.0 | None | | `zesTemperatureGetState` | Temperature | 0.1.0 | Linux: Requires superuser or read permissions for telem nodes | +| `zesTemperatureSetConfig` | Temperature | 0.1.3 | Linux: Requires superuser | | **Engine Management** |-|-|-| | `zesDeviceEnumEngineGroups` | Engine | 0.1.0 | Linux: Shows "no handles found" error when not in superuser mode | | `zesEngineGetProperties` | Engine | 0.1.0 | None | | `zesEngineGetActivity` | Engine | 0.1.0 | None | +| `zesEngineGetActivityExt` | Engine | 0.1.3 | None | +| **Standby Management** |-|-|-| +| `zesDeviceEnumStandbyDomains` | Standby | 0.1.3 | None | +| `zesStandbyGetProperties` | Standby | 0.1.3 | None | +| `zesStandbyGetMode` | Standby | 0.1.3 | None | +| `zesStandbySetMode` | Standby | 0.1.3 | Linux: Requires superuser | +| **Virtual Function (VF) Management** |-|-|-| +| `zesDeviceEnumEnabledVFExp` | VF Management | 0.1.3 | Experimental API | +| `zesVFManagementGetVFCapabilitiesExp2` | VF Management | 0.1.3 | Experimental API | +| `zesVFManagementGetVFEngineUtilizationExp2` | VF Management | 0.1.3 | Experimental API | +| `zesVFManagementGetVFMemoryUtilizationExp2` | VF Management | 0.1.3 | Experimental API | | **PCI Management** |-|-|-| | `zesDevicePciGetProperties` | PCI | 0.1.2 | Linux: Requires superuser or read permissions for telem nodes | | `zesDevicePciGetState` | PCI | 0.1.2 | Linux: Requires superuser or read permissions for telem nodes | 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..b67ca063 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,23 @@ def get_device_action_string(action): return action_map.get(action, f"UNKNOWN_DEVICE_ACTION_{action}") +def get_standby_type_string(standby_type): + """Convert standby type enum to string""" + type_map = { + pz.ZES_STANDBY_TYPE_GLOBAL: "ZES_STANDBY_TYPE_GLOBAL", + } + return type_map.get(standby_type, f"UNKNOWN_STANDBY_TYPE_{standby_type}") + + +def get_standby_promo_mode_string(mode): + """Convert standby promo mode enum to string""" + mode_map = { + pz.ZES_STANDBY_PROMO_MODE_DEFAULT: "ZES_STANDBY_PROMO_MODE_DEFAULT", + pz.ZES_STANDBY_PROMO_MODE_NEVER: "ZES_STANDBY_PROMO_MODE_NEVER", + } + return mode_map.get(mode, f"UNKNOWN_STANDBY_PROMO_MODE_{mode}") + + def is_root_user(): """Return whether the current user has root privileges on platforms that support it""" geteuid = getattr(os, "geteuid", None) @@ -837,6 +854,222 @@ def test_engine_modules(device_handle, device_index): print_verbose(f" Active Time: {engineStats.activeTime}") print_verbose(f" Timestamp: {engineStats.timestamp}") + for i in range(engine_count.value): + ext_count = c_uint32(0) + rc = pz.zesEngineGetActivityExt(engine_handles[i], byref(ext_count), None) + if rc != pz.ZE_RESULT_SUCCESS: + print_verbose( + f"\n Engine Module {i} Activity Ext: Not available ({get_result_string(rc)})" + ) + continue + + if ext_count.value == 0: + continue + + # Take two samples to derive the utilization of the PF and each VF + EngineStatsArray = pz.zes_engine_stats_t * ext_count.value + ext_stats = EngineStatsArray() + rc = pz.zesEngineGetActivityExt(engine_handles[i], byref(ext_count), ext_stats) + if not check_rc(f"zesEngineGetActivityExt(engine {i}, sample 1)", rc): + continue + + time.sleep(0.15) + ext_stats2 = EngineStatsArray() + rc = pz.zesEngineGetActivityExt(engine_handles[i], byref(ext_count), ext_stats2) + if not check_rc(f"zesEngineGetActivityExt(engine {i}, sample 2)", rc): + continue + + print_verbose(f"\n Engine Module {i} Activity Ext:") + for j in range(ext_count.value): + label = "PF" if j == 0 else f"VF {j}" + print_verbose( + f" {label}: Active Time: {ext_stats2[j].activeTime}, Timestamp: {ext_stats2[j].timestamp}" + ) + delta_active = ext_stats2[j].activeTime - ext_stats[j].activeTime + delta_time = ext_stats2[j].timestamp - ext_stats[j].timestamp + if delta_time > 0: + print_verbose( + f" Utilization: {100.0 * delta_active / delta_time:.2f}%" + ) + else: + print_verbose(" Utilization: unavailable due to zero delta time") + + return True + + +def test_standby_module(device_handle, device_index): + """Test standby domain enumeration, properties, and mode operations""" + print(f"\n---- Device {device_index} Standby Domains Test ----") + + standby_count = c_uint32(0) + rc = pz.zesDeviceEnumStandbyDomains(device_handle, byref(standby_count), None) + if not check_rc(f"zesDeviceEnumStandbyDomains(device {device_index}, count)", rc): + return False + + if standby_count.value == 0: + print_verbose("No standby domains found on this device") + return True + + print_verbose(f"Found {standby_count.value} standby domain(s)") + + StandbyArray = pz.zes_standby_handle_t * standby_count.value + standby_handles = StandbyArray() + + rc = pz.zesDeviceEnumStandbyDomains( + device_handle, byref(standby_count), standby_handles + ) + if not check_rc(f"zesDeviceEnumStandbyDomains(device {device_index}, handles)", rc): + return False + + for i in range(standby_count.value): + print_verbose(f"\n Standby Domain {i}:") + + props = pz.zes_standby_properties_t() + props.stype = pz.ZES_STRUCTURE_TYPE_STANDBY_PROPERTIES + props.pNext = None + + rc = pz.zesStandbyGetProperties(standby_handles[i], byref(props)) + if not check_rc(f"zesStandbyGetProperties(standby {i})", rc): + continue + + print_verbose(f" Type: {get_standby_type_string(props.type)}") + print_verbose(f" On Subdevice: {bool(props.onSubdevice)}") + if props.onSubdevice: + print_verbose(f" Subdevice ID: {props.subdeviceId}") + + mode = pz.zes_standby_promo_mode_t(0) + rc = pz.zesStandbyGetMode(standby_handles[i], byref(mode)) + if not check_rc(f"zesStandbyGetMode(standby {i})", rc): + continue + + print_verbose(f" Mode: {get_standby_promo_mode_string(mode.value)}") + + if not is_root_user(): + print_verbose( + " Skipping zesStandbySetMode due to insufficient permissions" + ) + continue + + for new_mode in ( + pz.ZES_STANDBY_PROMO_MODE_DEFAULT, + pz.ZES_STANDBY_PROMO_MODE_NEVER, + ): + mode_name = get_standby_promo_mode_string(new_mode) + rc = pz.zesStandbySetMode(standby_handles[i], new_mode) + if not check_rc(f"zesStandbySetMode(standby {i}, {mode_name})", rc): + continue + + read_back = pz.zes_standby_promo_mode_t(0) + rc = pz.zesStandbyGetMode(standby_handles[i], byref(read_back)) + if check_rc(f"zesStandbyGetMode(standby {i}, verify)", rc): + status = "OK" if read_back.value == new_mode else "MISMATCH" + print_verbose( + f" Set mode {mode_name}, read back " + f"{get_standby_promo_mode_string(read_back.value)} ({status})" + ) + + # Restore the mode read before the test + rc = pz.zesStandbySetMode(standby_handles[i], mode.value) + if check_rc(f"zesStandbySetMode(standby {i}, restore)", rc): + print_verbose( + f" Restored standby mode to {get_standby_promo_mode_string(mode.value)}" + ) + + return True + + +def test_vf_management_module(device_handle, device_index): + """Test enabled VF enumeration, capabilities, and utilization operations""" + print(f"\n---- Device {device_index} VF Management Test ----") + + vf_count = c_uint32(0) + rc = pz.zesDeviceEnumEnabledVFExp(device_handle, byref(vf_count), None) + if not check_rc(f"zesDeviceEnumEnabledVFExp(device {device_index}, count)", rc): + return False + + if vf_count.value == 0: + print_verbose("No enabled VFs found on this device") + return True + + print_verbose(f"Found {vf_count.value} enabled VF(s)") + + VFArray = pz.zes_vf_handle_t * vf_count.value + vf_handles = VFArray() + + rc = pz.zesDeviceEnumEnabledVFExp(device_handle, byref(vf_count), vf_handles) + if not check_rc(f"zesDeviceEnumEnabledVFExp(device {device_index}, handles)", rc): + return False + + for i in range(vf_count.value): + print_verbose(f"\n VF {i}:") + + capability = pz.zes_vf_exp2_capabilities_t() + capability.stype = pz.ZES_STRUCTURE_TYPE_VF_EXP2_CAPABILITIES + capability.pNext = None + + rc = pz.zesVFManagementGetVFCapabilitiesExp2(vf_handles[i], byref(capability)) + if not check_rc(f"zesVFManagementGetVFCapabilitiesExp2(vf {i})", rc): + continue + + print_verbose(" Capabilities:") + print_verbose(f" VF ID: {capability.vfID}") + print_verbose( + f" BDF: {capability.address.domain:04X}:{capability.address.bus:02X}:" + f"{capability.address.device:02X}.{capability.address.function:X}" + ) + print_verbose(f" Device Memory Size: {capability.vfDeviceMemSize} bytes") + + mem_count = c_uint32(0) + rc = pz.zesVFManagementGetVFMemoryUtilizationExp2( + vf_handles[i], byref(mem_count), None + ) + if check_rc(f"zesVFManagementGetVFMemoryUtilizationExp2(vf {i}, count)", rc): + if mem_count.value > 0: + MemUtilArray = pz.zes_vf_util_mem_exp2_t * mem_count.value + mem_util = MemUtilArray() + for j in range(mem_count.value): + mem_util[j].stype = pz.ZES_STRUCTURE_TYPE_VF_UTIL_MEM_EXP2 + mem_util[j].pNext = None + + rc = pz.zesVFManagementGetVFMemoryUtilizationExp2( + vf_handles[i], byref(mem_count), mem_util + ) + if check_rc( + f"zesVFManagementGetVFMemoryUtilizationExp2(vf {i}, data)", rc + ): + print_verbose(" Memory Utilization:") + for j in range(mem_count.value): + print_verbose( + f" {get_memory_location_string(mem_util[j].vfMemLocation)}: " + f"{mem_util[j].vfMemUtilized} bytes" + ) + + engine_count = c_uint32(0) + rc = pz.zesVFManagementGetVFEngineUtilizationExp2( + vf_handles[i], byref(engine_count), None + ) + if check_rc(f"zesVFManagementGetVFEngineUtilizationExp2(vf {i}, count)", rc): + if engine_count.value > 0: + EngineUtilArray = pz.zes_vf_util_engine_exp2_t * engine_count.value + engine_util = EngineUtilArray() + for j in range(engine_count.value): + engine_util[j].stype = pz.ZES_STRUCTURE_TYPE_VF_UTIL_ENGINE_EXP2 + engine_util[j].pNext = None + + rc = pz.zesVFManagementGetVFEngineUtilizationExp2( + vf_handles[i], byref(engine_count), engine_util + ) + if check_rc( + f"zesVFManagementGetVFEngineUtilizationExp2(vf {i}, data)", rc + ): + print_verbose(" Engine Utilization:") + for j in range(engine_count.value): + print_verbose( + f" {get_engine_type_string(engine_util[j].vfEngineType)}: " + f"Active Counter: {engine_util[j].activeCounterValue}, " + f"Sampling Counter: {engine_util[j].samplingCounterValue}" + ) + return True @@ -1336,6 +1569,16 @@ def test_temperature_sensors(device_handle, device_index): if temp_config.threshold2.threshold >= 0 else " Threshold 2: Not set" ) + + if is_root_user(): + # Write back the config just read so the device configuration is unchanged + rc = pz.zesTemperatureSetConfig(temp_handles[i], byref(temp_config)) + if check_rc(f"zesTemperatureSetConfig(temperature {i})", rc): + print_verbose(" Set temperature config successfully") + else: + print_verbose( + " Skipping zesTemperatureSetConfig due to insufficient permissions" + ) else: print_verbose( f" Temperature Config: Not available ({get_result_string(rc)})" @@ -1395,6 +1638,12 @@ def run_all_tests(): # Test engine modules test_engine_modules(devices[device_idx], device_idx) + # Test standby domains + test_standby_module(devices[device_idx], device_idx) + + # Test VF management + test_vf_management_module(devices[device_idx], device_idx) + print("\n=== Test Completed ===") return True @@ -1413,6 +1662,8 @@ def main(): %(prog)s -f # Frequency tests only %(prog)s -t # Temperature tests only %(prog)s -e # Engine tests only + %(prog)s -s # Standby tests only + %(prog)s -V # VF management tests only %(prog)s -h # Show help message""", formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -1450,6 +1701,12 @@ 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( + "-s", "--standby", action="store_true", help="Run only standby tests" + ) + parser.add_argument( + "-V", "--vf", action="store_true", help="Run only VF management tests" + ) args = parser.parse_args() @@ -1463,6 +1720,8 @@ def main(): or args.frequency or args.temperature or args.engine + or args.standby + or args.vf or args.all ) @@ -1515,6 +1774,12 @@ def main(): if args.temperature: test_temperature_sensors(devices[device_idx], device_idx) + if args.standby: + test_standby_module(devices[device_idx], device_idx) + + if args.vf: + test_vf_management_module(devices[device_idx], device_idx) + 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..013cf19f 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_standby_handle_t(c_void_p): + pass + + +class zes_vf_handle_t(c_void_p): + pass + + ## ze_bool_t = c_uint8 @@ -294,6 +302,16 @@ class zes_engine_handle_t(c_void_p): ZES_ENGINE_GROUP_MEDIA_CODEC_SINGLE = 14 ZES_ENGINE_GROUP_FORCE_UINT32 = 0x7FFFFFFF +## Standby enums ## +zes_standby_type_t = c_int32 +ZES_STANDBY_TYPE_GLOBAL = 0 +ZES_STANDBY_TYPE_FORCE_UINT32 = 0x7FFFFFFF + +zes_standby_promo_mode_t = c_int32 +ZES_STANDBY_PROMO_MODE_DEFAULT = 0 +ZES_STANDBY_PROMO_MODE_NEVER = 1 +ZES_STANDBY_PROMO_MODE_FORCE_UINT32 = 0x7FFFFFFF + ze_result_t = c_int32 ZE_RESULT_SUCCESS = 0 ZE_RESULT_NOT_READY = 1 @@ -376,6 +394,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_STANDBY_PROPERTIES = 0x13 +ZES_STRUCTURE_TYPE_VF_UTIL_MEM_EXP2 = 0x00020009 # Experimental VF memory utilization +ZES_STRUCTURE_TYPE_VF_UTIL_ENGINE_EXP2 = ( + 0x00020010 # Experimental VF engine utilization +) +ZES_STRUCTURE_TYPE_VF_EXP2_CAPABILITIES = 0x00020011 # Experimental VF capabilities ## Core ze_device UUID struct ## @@ -779,6 +803,50 @@ class zes_engine_stats_t(_PrintableStructure): _fmt_ = {"activeTime": "%d", "timestamp": "%d"} +## Standby structures ## +class zes_standby_properties_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_STANDBY_PROPERTIES + ("pNext", c_void_p), + ("type", zes_standby_type_t), # standby hardware component + ("onSubdevice", ze_bool_t), # is on subdevice + ("subdeviceId", c_uint32), # subdevice ID + ] + + +## VF management structures ## +class zes_vf_exp2_capabilities_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_VF_EXP2_CAPABILITIES + ("pNext", c_void_p), + ("address", zes_pci_address_t), # VF BDF address + ("vfDeviceMemSize", c_uint64), # VF memory size in bytes + ("vfID", c_uint32), # VF ID + ] + _fmt_ = {"vfDeviceMemSize": "%d bytes"} + + +class zes_vf_util_mem_exp2_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_VF_UTIL_MEM_EXP2 + ("pNext", c_void_p), + ("vfMemLocation", zes_mem_loc_t), # memory location (system, device) + ("vfMemUtilized", c_uint64), # utilized memory size in bytes + ] + _fmt_ = {"vfMemUtilized": "%d bytes"} + + +class zes_vf_util_engine_exp2_t(_PrintableStructure): + _fields_ = [ + ("stype", c_int32), # ZES_STRUCTURE_TYPE_VF_UTIL_ENGINE_EXP2 + ("pNext", c_void_p), + ("vfEngineType", zes_engine_group_t), # engine group + ("activeCounterValue", c_uint64), # active counter + ("samplingCounterValue", c_uint64), # sampling counter + ] + _fmt_ = {"activeCounterValue": "%d", "samplingCounterValue": "%d"} + + ## Function access ## _getFunctionPointerList: Dict[str, Any] = dict() @@ -1492,6 +1560,26 @@ def zesTemperatureGetState(hTemperature, pTemperature): return retVal +def zesTemperatureSetConfig(hTemperature, pConfig): + """Wraps API: + ze_result_t zesTemperatureSetConfig( + zes_temp_handle_t hTemperature, + const zes_temp_config_t* pConfig) + + Parameters: + hTemperature: temperature handle + pConfig: POINTER(zes_temp_config_t) - new configuration + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesTemperatureSetConfig") + funcPtr.argtypes = [zes_temp_handle_t, POINTER(zes_temp_config_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hTemperature, pConfig) + return retVal + + ## Engine functions ## @@ -1556,3 +1644,213 @@ def zesEngineGetActivity(hEngine, pStats): retVal = funcPtr(hEngine, pStats) return retVal + + +def zesEngineGetActivityExt(hEngine, pCount, pStats): + """Wraps API: + ze_result_t zesEngineGetActivityExt( + zes_engine_handle_t hEngine, + uint32_t* pCount, + zes_engine_stats_t* pStats) + + Parameters: + hEngine: engine handle + pCount: POINTER(c_uint32) + pStats: POINTER(zes_engine_stats_t) or None - PF stats at index 0 followed by VF stats + Returns: + ze_result_t - return code only, stats are filled into pStats + """ + funcPtr = getFunctionPointerList("zesEngineGetActivityExt") + funcPtr.argtypes = [ + zes_engine_handle_t, + POINTER(c_uint32), + POINTER(zes_engine_stats_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hEngine, pCount, pStats) + return retVal + + +## Standby functions ## + + +def zesDeviceEnumStandbyDomains(hDevice, pCount, phStandby): + """Wraps API: + ze_result_t zesDeviceEnumStandbyDomains( + zes_device_handle_t hDevice, + uint32_t* pCount, + zes_standby_handle_t* phStandby) + + Parameters: + hDevice: device handle + pCount: POINTER(c_uint32) + phStandby: POINTER(zes_standby_handle_t) or None + """ + funcPtr = getFunctionPointerList("zesDeviceEnumStandbyDomains") + funcPtr.argtypes = [ + zes_device_handle_t, + POINTER(c_uint32), + POINTER(zes_standby_handle_t), + ] + funcPtr.restype = ze_result_t + retVal = funcPtr(hDevice, pCount, phStandby) + return retVal + + +def zesStandbyGetProperties(hStandby, pProperties): + """Wraps API: + ze_result_t zesStandbyGetProperties( + zes_standby_handle_t hStandby, + zes_standby_properties_t* pProperties) + + Parameters: + hStandby: standby handle + pProperties: POINTER(zes_standby_properties_t) - properties structure to fill + Returns: + ze_result_t - return code only, properties are filled into pProperties + """ + funcPtr = getFunctionPointerList("zesStandbyGetProperties") + funcPtr.argtypes = [zes_standby_handle_t, POINTER(zes_standby_properties_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hStandby, pProperties) + return retVal + + +def zesStandbyGetMode(hStandby, pMode): + """Wraps API: + ze_result_t zesStandbyGetMode( + zes_standby_handle_t hStandby, + zes_standby_promo_mode_t* pMode) + + Parameters: + hStandby: standby handle + pMode: POINTER(zes_standby_promo_mode_t) - current standby mode to fill + Returns: + ze_result_t - return code only, mode is filled into pMode + """ + funcPtr = getFunctionPointerList("zesStandbyGetMode") + funcPtr.argtypes = [zes_standby_handle_t, POINTER(zes_standby_promo_mode_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hStandby, pMode) + return retVal + + +def zesStandbySetMode(hStandby, mode): + """Wraps API: + ze_result_t zesStandbySetMode( + zes_standby_handle_t hStandby, + zes_standby_promo_mode_t mode) + + Parameters: + hStandby: standby handle + mode: zes_standby_promo_mode_t - new standby mode + Returns: + ze_result_t - return code only + """ + funcPtr = getFunctionPointerList("zesStandbySetMode") + funcPtr.argtypes = [zes_standby_handle_t, zes_standby_promo_mode_t] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hStandby, mode) + return retVal + + +## VF management functions ## + + +def zesDeviceEnumEnabledVFExp(hDevice, pCount, phVFhandle): + """Wraps API: + ze_result_t zesDeviceEnumEnabledVFExp( + zes_device_handle_t hDevice, + uint32_t* pCount, + zes_vf_handle_t* phVFhandle) + + Parameters: + hDevice: device handle + pCount: POINTER(c_uint32) + phVFhandle: POINTER(zes_vf_handle_t) or None + """ + funcPtr = getFunctionPointerList("zesDeviceEnumEnabledVFExp") + funcPtr.argtypes = [ + zes_device_handle_t, + POINTER(c_uint32), + POINTER(zes_vf_handle_t), + ] + funcPtr.restype = ze_result_t + retVal = funcPtr(hDevice, pCount, phVFhandle) + return retVal + + +def zesVFManagementGetVFCapabilitiesExp2(hVFhandle, pCapability): + """Wraps API: + ze_result_t zesVFManagementGetVFCapabilitiesExp2( + zes_vf_handle_t hVFhandle, + zes_vf_exp2_capabilities_t* pCapability) + + Parameters: + hVFhandle: VF handle + pCapability: POINTER(zes_vf_exp2_capabilities_t) - capability structure to fill + Returns: + ze_result_t - return code only, capability is filled into pCapability + """ + funcPtr = getFunctionPointerList("zesVFManagementGetVFCapabilitiesExp2") + funcPtr.argtypes = [zes_vf_handle_t, POINTER(zes_vf_exp2_capabilities_t)] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hVFhandle, pCapability) + return retVal + + +def zesVFManagementGetVFEngineUtilizationExp2(hVFhandle, pCount, pEngineUtil): + """Wraps API: + ze_result_t zesVFManagementGetVFEngineUtilizationExp2( + zes_vf_handle_t hVFhandle, + uint32_t* pCount, + zes_vf_util_engine_exp2_t* pEngineUtil) + + Parameters: + hVFhandle: VF handle + pCount: POINTER(c_uint32) + pEngineUtil: POINTER(zes_vf_util_engine_exp2_t) or None + Returns: + ze_result_t - return code only, engine utilization is filled into pEngineUtil + """ + funcPtr = getFunctionPointerList("zesVFManagementGetVFEngineUtilizationExp2") + funcPtr.argtypes = [ + zes_vf_handle_t, + POINTER(c_uint32), + POINTER(zes_vf_util_engine_exp2_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hVFhandle, pCount, pEngineUtil) + return retVal + + +def zesVFManagementGetVFMemoryUtilizationExp2(hVFhandle, pCount, pMemUtil): + """Wraps API: + ze_result_t zesVFManagementGetVFMemoryUtilizationExp2( + zes_vf_handle_t hVFhandle, + uint32_t* pCount, + zes_vf_util_mem_exp2_t* pMemUtil) + + Parameters: + hVFhandle: VF handle + pCount: POINTER(c_uint32) + pMemUtil: POINTER(zes_vf_util_mem_exp2_t) or None + Returns: + ze_result_t - return code only, memory utilization is filled into pMemUtil + """ + funcPtr = getFunctionPointerList("zesVFManagementGetVFMemoryUtilizationExp2") + funcPtr.argtypes = [ + zes_vf_handle_t, + POINTER(c_uint32), + POINTER(zes_vf_util_mem_exp2_t), + ] + funcPtr.restype = ze_result_t + + retVal = funcPtr(hVFhandle, pCount, pMemUtil) + return retVal diff --git a/bindings/sysman/python/test/unit_tests/test_engine.py b/bindings/sysman/python/test/unit_tests/test_engine.py index 0f928b5c..77bb7ba3 100644 --- a/bindings/sysman/python/test/unit_tests/test_engine.py +++ b/bindings/sysman/python/test/unit_tests/test_engine.py @@ -100,6 +100,59 @@ def mock_get_activity(engine_handle, stats_ptr): mock_get_func.assert_called_with("zesEngineGetActivity") mock_func.assert_called_once() + def test_GivenValidEngineHandleWhenCallingZesEngineGetActivityExtThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 3 + + def mock_get_activity_ext(engine_handle, count_ptr, stats_ptr): + count_ptr._obj.value = mock_count + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_activity_ext) + mock_get_func.return_value = mock_func + + engine_handle = self.pyzes.zes_engine_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesEngineGetActivityExt(engine_handle, byref(count), None) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + mock_get_func.assert_called_with("zesEngineGetActivityExt") + mock_func.assert_called_once() + + def test_GivenValidEngineHandleWhenCallingZesEngineGetActivityExtWithArrayThenCallSucceedsWithStats( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_activity_ext(engine_handle, count_ptr, stats_ptr): + count_ptr._obj.value = mock_count + stats_ptr[0].activeTime = 1000 + stats_ptr[0].timestamp = 5000 + stats_ptr[1].activeTime = 400 + stats_ptr[1].timestamp = 5000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_activity_ext) + mock_get_func.return_value = mock_func + + engine_handle = self.pyzes.zes_engine_handle_t() + count = c_uint32(mock_count) + stats = (self.pyzes.zes_engine_stats_t * mock_count)() + + result = self.pyzes.zesEngineGetActivityExt(engine_handle, byref(count), stats) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual(stats[0].activeTime, 1000) + self.assertEqual(stats[0].timestamp, 5000) + self.assertEqual(stats[1].activeTime, 400) + self.assertEqual(stats[1].timestamp, 5000) + mock_get_func.assert_called_with("zesEngineGetActivityExt") + mock_func.assert_called_once() + if __name__ == "__main__": unittest.main() diff --git a/bindings/sysman/python/test/unit_tests/test_standby.py b/bindings/sysman/python/test/unit_tests/test_standby.py new file mode 100644 index 00000000..583a8070 --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_standby.py @@ -0,0 +1,119 @@ +## +# 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 TestStandbyFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEnumStandbyDomainsThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 1 + + def mock_enum_standby(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_standby) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDeviceEnumStandbyDomains( + 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("zesDeviceEnumStandbyDomains") + mock_func.assert_called_once() + + def test_GivenValidStandbyHandleWhenCallingZesStandbyGetPropertiesThenCallSucceedsWithValidProperties( + self, mock_get_func + ): + def mock_get_properties(standby_handle, properties_ptr): + properties = properties_ptr._obj + properties.type = self.pyzes.ZES_STANDBY_TYPE_GLOBAL + 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 + + standby_handle = self.pyzes.zes_standby_handle_t() + properties = self.pyzes.zes_standby_properties_t() + properties.stype = self.pyzes.ZES_STRUCTURE_TYPE_STANDBY_PROPERTIES + + result = self.pyzes.zesStandbyGetProperties(standby_handle, byref(properties)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(properties.type, self.pyzes.ZES_STANDBY_TYPE_GLOBAL) + self.assertEqual(properties.onSubdevice, 1) + self.assertEqual(properties.subdeviceId, 1) + mock_get_func.assert_called_with("zesStandbyGetProperties") + mock_func.assert_called_once() + + def test_GivenValidStandbyHandleWhenCallingZesStandbyGetModeThenCallSucceedsWithMode( + self, mock_get_func + ): + def mock_get_mode(standby_handle, mode_ptr): + mode_ptr._obj.value = self.pyzes.ZES_STANDBY_PROMO_MODE_NEVER + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_mode) + mock_get_func.return_value = mock_func + + standby_handle = self.pyzes.zes_standby_handle_t() + mode = self.pyzes.zes_standby_promo_mode_t(0) + + result = self.pyzes.zesStandbyGetMode(standby_handle, byref(mode)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(mode.value, self.pyzes.ZES_STANDBY_PROMO_MODE_NEVER) + mock_get_func.assert_called_with("zesStandbyGetMode") + mock_func.assert_called_once() + + def test_GivenValidStandbyHandleWhenCallingZesStandbySetModeThenCallSucceeds( + self, mock_get_func + ): + mock_func = MagicMock(return_value=self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.return_value = mock_func + + standby_handle = self.pyzes.zes_standby_handle_t() + + result = self.pyzes.zesStandbySetMode( + standby_handle, self.pyzes.ZES_STANDBY_PROMO_MODE_DEFAULT + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesStandbySetMode") + mock_func.assert_called_once_with( + standby_handle, self.pyzes.ZES_STANDBY_PROMO_MODE_DEFAULT + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/bindings/sysman/python/test/unit_tests/test_temperature.py b/bindings/sysman/python/test/unit_tests/test_temperature.py index adde881c..ae23e6ec 100644 --- a/bindings/sysman/python/test/unit_tests/test_temperature.py +++ b/bindings/sysman/python/test/unit_tests/test_temperature.py @@ -144,6 +144,37 @@ def mock_get_state(temp_handle, temperature_ptr): mock_get_func.assert_called_with("zesTemperatureGetState") mock_func.assert_called_once() + def test_GivenValidTemperatureHandleWhenCallingZesTemperatureSetConfigThenCallSucceeds( + self, mock_get_func + ): + def mock_set_config(temp_handle, config_ptr): + config = config_ptr._obj + self.assertEqual(config.stype, self.pyzes.ZES_STRUCTURE_TYPE_TEMP_CONFIG) + self.assertEqual(config.enableCritical, 1) + self.assertEqual(config.threshold1.enableLowToHigh, 1) + self.assertEqual(config.threshold1.threshold, 85.0) + self.assertEqual(config.threshold2.enableHighToLow, 1) + self.assertEqual(config.threshold2.threshold, 60.0) + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_set_config) + mock_get_func.return_value = mock_func + + temp_handle = self.pyzes.zes_temp_handle_t() + config = self.pyzes.zes_temp_config_t() + config.stype = self.pyzes.ZES_STRUCTURE_TYPE_TEMP_CONFIG + config.enableCritical = 1 + config.threshold1.enableLowToHigh = 1 + config.threshold1.threshold = 85.0 + config.threshold2.enableHighToLow = 1 + config.threshold2.threshold = 60.0 + + result = self.pyzes.zesTemperatureSetConfig(temp_handle, byref(config)) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + mock_get_func.assert_called_with("zesTemperatureSetConfig") + mock_func.assert_called_once() + if __name__ == "__main__": unittest.main() diff --git a/bindings/sysman/python/test/unit_tests/test_vf_management.py b/bindings/sysman/python/test/unit_tests/test_vf_management.py new file mode 100644 index 00000000..db3fdf7e --- /dev/null +++ b/bindings/sysman/python/test/unit_tests/test_vf_management.py @@ -0,0 +1,162 @@ +## +# 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 TestVFManagementFunctions(unittest.TestCase): + def setUp(self): + import pyzes + + self.pyzes = pyzes + + def test_GivenValidDeviceHandleWhenCallingZesDeviceEnumEnabledVFExpThenCallSucceedsWithValidCount( + self, mock_get_func + ): + mock_count = 2 + + def mock_enum_vf(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_vf) + mock_get_func.return_value = mock_func + + device_handle = self.pyzes.zes_device_handle_t() + count = c_uint32(0) + + result = self.pyzes.zesDeviceEnumEnabledVFExp(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("zesDeviceEnumEnabledVFExp") + mock_func.assert_called_once() + + def test_GivenValidVFHandleWhenCallingZesVFManagementGetVFCapabilitiesExp2ThenCallSucceedsWithCapabilities( + self, mock_get_func + ): + def mock_get_capabilities(vf_handle, capability_ptr): + capability = capability_ptr._obj + capability.address.domain = 0 + capability.address.bus = 0x3A + capability.address.device = 0 + capability.address.function = 1 + capability.vfDeviceMemSize = 4 * 1024 * 1024 * 1024 + capability.vfID = 1 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_capabilities) + mock_get_func.return_value = mock_func + + vf_handle = self.pyzes.zes_vf_handle_t() + capability = self.pyzes.zes_vf_exp2_capabilities_t() + capability.stype = self.pyzes.ZES_STRUCTURE_TYPE_VF_EXP2_CAPABILITIES + + result = self.pyzes.zesVFManagementGetVFCapabilitiesExp2( + vf_handle, byref(capability) + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(capability.address.bus, 0x3A) + self.assertEqual(capability.address.function, 1) + self.assertEqual(capability.vfDeviceMemSize, 4 * 1024 * 1024 * 1024) + self.assertEqual(capability.vfID, 1) + mock_get_func.assert_called_with("zesVFManagementGetVFCapabilitiesExp2") + mock_func.assert_called_once() + + def test_GivenValidVFHandleWhenCallingZesVFManagementGetVFEngineUtilizationExp2ThenCallSucceedsWithUtilization( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_engine_util(vf_handle, count_ptr, util_ptr): + count_ptr._obj.value = mock_count + util_ptr[0].vfEngineType = self.pyzes.ZES_ENGINE_GROUP_COMPUTE_ALL + util_ptr[0].activeCounterValue = 1200 + util_ptr[0].samplingCounterValue = 5000 + util_ptr[1].vfEngineType = self.pyzes.ZES_ENGINE_GROUP_COPY_ALL + util_ptr[1].activeCounterValue = 300 + util_ptr[1].samplingCounterValue = 5000 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_engine_util) + mock_get_func.return_value = mock_func + + vf_handle = self.pyzes.zes_vf_handle_t() + count = c_uint32(mock_count) + engine_util = (self.pyzes.zes_vf_util_engine_exp2_t * mock_count)() + for i in range(mock_count): + engine_util[i].stype = self.pyzes.ZES_STRUCTURE_TYPE_VF_UTIL_ENGINE_EXP2 + + result = self.pyzes.zesVFManagementGetVFEngineUtilizationExp2( + vf_handle, byref(count), engine_util + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual( + engine_util[0].vfEngineType, self.pyzes.ZES_ENGINE_GROUP_COMPUTE_ALL + ) + self.assertEqual(engine_util[0].activeCounterValue, 1200) + self.assertEqual(engine_util[0].samplingCounterValue, 5000) + self.assertEqual( + engine_util[1].vfEngineType, self.pyzes.ZES_ENGINE_GROUP_COPY_ALL + ) + self.assertEqual(engine_util[1].activeCounterValue, 300) + mock_get_func.assert_called_with("zesVFManagementGetVFEngineUtilizationExp2") + mock_func.assert_called_once() + + def test_GivenValidVFHandleWhenCallingZesVFManagementGetVFMemoryUtilizationExp2ThenCallSucceedsWithUtilization( + self, mock_get_func + ): + mock_count = 2 + + def mock_get_mem_util(vf_handle, count_ptr, util_ptr): + count_ptr._obj.value = mock_count + util_ptr[0].vfMemLocation = self.pyzes.ZES_MEM_LOC_SYSTEM + util_ptr[0].vfMemUtilized = 1024 + util_ptr[1].vfMemLocation = self.pyzes.ZES_MEM_LOC_DEVICE + util_ptr[1].vfMemUtilized = 2048 + return self.pyzes.ZE_RESULT_SUCCESS + + mock_func = MagicMock(side_effect=mock_get_mem_util) + mock_get_func.return_value = mock_func + + vf_handle = self.pyzes.zes_vf_handle_t() + count = c_uint32(mock_count) + mem_util = (self.pyzes.zes_vf_util_mem_exp2_t * mock_count)() + for i in range(mock_count): + mem_util[i].stype = self.pyzes.ZES_STRUCTURE_TYPE_VF_UTIL_MEM_EXP2 + + result = self.pyzes.zesVFManagementGetVFMemoryUtilizationExp2( + vf_handle, byref(count), mem_util + ) + + self.assertEqual(result, self.pyzes.ZE_RESULT_SUCCESS) + self.assertEqual(count.value, mock_count) + self.assertEqual(mem_util[0].vfMemLocation, self.pyzes.ZES_MEM_LOC_SYSTEM) + self.assertEqual(mem_util[0].vfMemUtilized, 1024) + self.assertEqual(mem_util[1].vfMemLocation, self.pyzes.ZES_MEM_LOC_DEVICE) + self.assertEqual(mem_util[1].vfMemUtilized, 2048) + mock_get_func.assert_called_with("zesVFManagementGetVFMemoryUtilizationExp2") + mock_func.assert_called_once() + + +if __name__ == "__main__": + unittest.main()