"""Unit tests for the runtime device/provider selection in inference_pool. Covers: - _resolve_ov_mode — the OpenVINO device-mode branch matrix. - _resolve_onnx_providers — the ONNX execution-provider ordering with camstack's pin + availability filter. Pure-python: stubs numpy / PIL / postprocessors in sys.modules so the module imports without the ML runtime installed (inference_pool uses `from __future__ import annotations`, so annotations never touch the stubs). Run: python3 -m unittest test_inference_pool_device_selection -v """ from __future__ import annotations import sys import types import unittest # --------------------------------------------------------------------------- # Stub the heavy third-party imports BEFORE importing inference_pool. # --------------------------------------------------------------------------- if "numpy" not in sys.modules: sys.modules["numpy"] = types.ModuleType("numpy") if "PIL" not in sys.modules: _pil = types.ModuleType("PIL") _pil_image = types.ModuleType("PIL.Image") _pil.Image = _pil_image sys.modules["PIL"] = _pil sys.modules["PIL.Image"] = _pil_image if "postprocessors" not in sys.modules: _pp = types.ModuleType("postprocessors") _pp.POSTPROCESSORS = {} sys.modules["postprocessors"] = _pp from inference_pool import ( ONNX_COREML_EP, ONNX_CPU_EP, ONNX_CUDA_EP, _resolve_onnx_providers, _resolve_ov_mode, ) class ResolveOvModeTest(unittest.TestCase): """Branch matrix of the OpenVINO device switch.""" def test_n100_intel_igpu_and_cpu_is_explicit_gpu(self) -> None: # Intel iGPU + CPU, no NPU, no dGPU — compiles explicit GPU. self.assertEqual( _resolve_ov_mode( ["CPU", "GPU"], { "CPU": "Intel(R) N100", "GPU": "Intel(R) UHD Graphics [0x46d1] (iGPU)", }, ), "GPU", ) def test_hub_npu_gpu_cpu_is_auto_npu_gpu_cpu(self) -> None: self.assertEqual( _resolve_ov_mode( ["CPU", "GPU", "NPU"], { "CPU": "Intel(R) Core(TM) Ultra 5 125H", "GPU": "Intel(R) Arc(TM) Graphics (iGPU)", "NPU": "Intel(R) AI Boost", }, ), "AUTO:NPU,GPU,CPU", ) def test_npu_without_gpu_is_auto_npu_cpu(self) -> None: self.assertEqual( _resolve_ov_mode( ["CPU", "NPU"], { "CPU": "Intel(R) Core(TM) Ultra", "NPU": "Intel(R) AI Boost", }, ), "AUTO:NPU,CPU", ) def test_nvidia_dgpu_is_auto_dgpu_cpu(self) -> None: self.assertEqual( _resolve_ov_mode( ["CPU", "GPU.0"], { "CPU": "Intel(R) Core(TM) i5", "GPU.0": "NVIDIA GeForce RTX 3060 (dGPU)", }, ), "AUTO:GPU.0,CPU", ) def test_multiple_nvidia_dgpus_join_in_enumeration_order(self) -> None: self.assertEqual( _resolve_ov_mode( ["CPU", "GPU.0", "GPU.1"], { "CPU": "Intel(R) Core(TM) i9", "GPU.0": "NVIDIA GeForce RTX 4090 (dGPU)", "GPU.1": "NVIDIA GeForce RTX 4090 (dGPU)", }, ), "AUTO:GPU.0,GPU.1,CPU", ) def test_nvidia_dgpu_beats_intel_igpu(self) -> None: # dGPU branch is checked before the plain-gpu branch (elif order). self.assertEqual( _resolve_ov_mode( ["CPU", "GPU.0", "GPU.1"], { "CPU": "Intel(R) Core(TM) i7", "GPU.0": "Intel(R) UHD Graphics (iGPU)", "GPU.1": "NVIDIA GeForce RTX 3080 (dGPU)", }, ), "AUTO:GPU.1,CPU", ) def test_npu_wins_over_nvidia_dgpu(self) -> None: # npu is checked FIRST — dGPUs only matter when no NPU exists. self.assertEqual( _resolve_ov_mode( ["CPU", "GPU.0", "NPU"], { "CPU": "Intel(R) Core(TM) Ultra", "GPU.0": "NVIDIA GeForce RTX 3060 (dGPU)", "NPU": "Intel(R) AI Boost", }, ), "AUTO:NPU,GPU,CPU", ) def test_cpu_only_is_auto(self) -> None: self.assertEqual( _resolve_ov_mode(["CPU"], {"CPU": "Intel(R) N100"}), "AUTO", ) def test_empty_available_is_auto(self) -> None: self.assertEqual(_resolve_ov_mode([], {}), "AUTO") def test_device_without_full_name_is_not_classified(self) -> None: # A device whose # FULL_DEVICE_NAME query failed is skipped entirely. self.assertEqual( _resolve_ov_mode(["CPU", "GPU"], {"CPU": "Intel(R) N100"}), "AUTO", ) class ResolveOnnxProvidersTest(unittest.TestCase): """ONNX EP ordering + camstack's pin/availability filter.""" def test_darwin_auto_prefers_coreml_cpu_last(self) -> None: self.assertEqual( _resolve_onnx_providers( "", "darwin", "arm64", [ONNX_COREML_EP, ONNX_CPU_EP], ), [ONNX_COREML_EP, ONNX_CPU_EP], ) def test_linux_x86_auto_prefers_cuda_with_device_id(self) -> None: self.assertEqual( _resolve_onnx_providers( "auto", "linux", "x86_64", [ONNX_CUDA_EP, ONNX_CPU_EP], ), [(ONNX_CUDA_EP, {"device_id": 0}), ONNX_CPU_EP], ) def test_windows_amd64_auto_prefers_cuda(self) -> None: self.assertEqual( _resolve_onnx_providers( "", "win32", "AMD64", [ONNX_CUDA_EP, ONNX_CPU_EP], ), [(ONNX_CUDA_EP, {"device_id": 0}), ONNX_CPU_EP], ) def test_plain_wheel_drops_unavailable_cuda(self) -> None: # camstack ships the plain onnxruntime wheel (no CUDA EP) — the # unavailable EP is filtered so session creation cannot raise. self.assertEqual( _resolve_onnx_providers("auto", "linux", "x86_64", [ONNX_CPU_EP]), [ONNX_CPU_EP], ) def test_linux_arm64_auto_is_cpu_only(self) -> None: self.assertEqual( _resolve_onnx_providers( "", "linux", "aarch64", [ONNX_CUDA_EP, ONNX_CPU_EP], ), [ONNX_CPU_EP], ) def test_darwin_never_matches_win_substring(self) -> None: # "win" in "darwin" is True — the helper must NOT add CUDA on an # Intel Mac (a substring "win" in platform check would misfire). self.assertEqual( _resolve_onnx_providers( "", "darwin", "x86_64", [ONNX_COREML_EP, ONNX_CUDA_EP, ONNX_CPU_EP], ), [ONNX_COREML_EP, ONNX_CPU_EP], ) def test_cpu_pin_is_cpu_only(self) -> None: self.assertEqual( _resolve_onnx_providers( "cpu", "linux", "x86_64", [ONNX_CUDA_EP, ONNX_CPU_EP], ), [ONNX_CPU_EP], ) def test_cuda_pin_orders_cuda_first_cpu_last(self) -> None: self.assertEqual( _resolve_onnx_providers( "cuda", "linux", "x86_64", [ONNX_CUDA_EP, ONNX_CPU_EP], ), [(ONNX_CUDA_EP, {"device_id": 0}), ONNX_CPU_EP], ) def test_cuda_pin_respects_device_id(self) -> None: self.assertEqual( _resolve_onnx_providers( "cuda", "linux", "x86_64", [ONNX_CUDA_EP, ONNX_CPU_EP], cuda_device_id=1, ), [(ONNX_CUDA_EP, {"device_id": 1}), ONNX_CPU_EP], ) def test_coreml_pin_orders_coreml_first(self) -> None: self.assertEqual( _resolve_onnx_providers( "coreml", "darwin", "arm64", [ONNX_COREML_EP, ONNX_CPU_EP], ), [ONNX_COREML_EP, ONNX_CPU_EP], ) def test_pinned_but_unavailable_ep_falls_back_to_cpu(self) -> None: self.assertEqual( _resolve_onnx_providers("coreml", "linux", "x86_64", [ONNX_CPU_EP]), [ONNX_CPU_EP], ) if __name__ == "__main__": unittest.main()