diff --git a/src/winml/modelkit/analyze/analyzer.py b/src/winml/modelkit/analyze/analyzer.py index 60e5126f0..b747e03e9 100644 --- a/src/winml/modelkit/analyze/analyzer.py +++ b/src/winml/modelkit/analyze/analyzer.py @@ -581,8 +581,9 @@ def analyze( # Load ONNX model try: - # Load without strict validation to allow custom attributes like hierarchy_tag - model_proto = onnx.load(str(model_file), load_external_data=True) + # Load without external data — static analysis only needs graph structure, + # shapes, and small embedded constants; not multi-GB weight tensors. + model_proto = onnx.load(str(model_file), load_external_data=False) # Skip onnx.checker.check_model() which rejects custom attributes except (OSError, FileNotFoundError) as e: raise RuntimeError(f"Failed to load ONNX model: {e}") from e diff --git a/src/winml/modelkit/analyze/core/onnx_loader.py b/src/winml/modelkit/analyze/core/onnx_loader.py index 4ae7e4c56..5945f9508 100644 --- a/src/winml/modelkit/analyze/core/onnx_loader.py +++ b/src/winml/modelkit/analyze/core/onnx_loader.py @@ -148,7 +148,7 @@ def load(self) -> ONNXModel: # FR-001: Load ONNX model from file try: logger.info("Loading ONNX model from: %s", self._model_path) - model_proto = onnx.load(str(self._model_path)) + model_proto = onnx.load(str(self._model_path), load_external_data=False) except Exception as e: # FR-038: Provide clear error message raise ONNXLoadError( diff --git a/src/winml/modelkit/analyze/core/runtime_checker.py b/src/winml/modelkit/analyze/core/runtime_checker.py index 4c8e65054..b48d98438 100644 --- a/src/winml/modelkit/analyze/core/runtime_checker.py +++ b/src/winml/modelkit/analyze/core/runtime_checker.py @@ -139,6 +139,7 @@ def _get_query(self) -> RuntimeCheckerQuery: model_proto=model_proto, ep_name=self._ep, device_type=self._device, + model_path=self._model.model_path, dynamic_axis_strict_mode=self._dynamic_axis_strict_mode, ) diff --git a/src/winml/modelkit/analyze/core/runtime_checker_query.py b/src/winml/modelkit/analyze/core/runtime_checker_query.py index 8ac93183b..bb91798e5 100644 --- a/src/winml/modelkit/analyze/core/runtime_checker_query.py +++ b/src/winml/modelkit/analyze/core/runtime_checker_query.py @@ -188,8 +188,7 @@ def _expand_snapshot_payload( base_opset = payload.get(SNAPSHOT_BASE_OPSET_KEY) if not isinstance(base_opset, int): logger.warning( - "Delta snapshot %s in %s missing integer %s; " - "applying delta on empty base.", + "Delta snapshot %s in %s missing integer %s; applying delta on empty base.", file_name, zip_path, SNAPSHOT_BASE_OPSET_KEY, @@ -217,8 +216,7 @@ def _expand_snapshot_payload( base_raw = _read_json_from_zip(base_zip_path, base_file_name) if base_raw is None: logger.warning( - "Base snapshot %s not found in %s (from %s); " - "applying delta on empty base.", + "Base snapshot %s not found in %s (from %s); applying delta on empty base.", base_file_name, base_zip_path, file_name, @@ -633,6 +631,103 @@ def _normalize_type_var_annotation(type_value: str) -> str: return SupportedONNXType.normalize_annotation(type_value) +# Query conditions are later normalized with make_hashable, which materializes +# numpy arrays. Keep external initializer loading bounded until the broader +# condition pipeline can stay lazy end-to-end. +MAX_EXTERNAL_INITIALIZER_BYTES_FOR_QUERY = 1024 * 1024 + + +def _get_external_tensor_info(tensor: onnx.TensorProto) -> tuple[str | None, int, int | None]: + """Extract location, offset, and length metadata for an external tensor.""" + info = {entry.key: entry.value for entry in tensor.external_data} + location = info.get("location") + offset = int(info.get("offset", "0")) + length = int(info["length"]) if "length" in info else None + return location, offset, length + + +def _tensor_proto_dtype_to_np_dtype(tensor_type: int) -> np.dtype[Any]: + """Convert a TensorProto dtype enum to a numpy dtype.""" + try: + from onnx.helper import tensor_dtype_to_np_dtype as onnx_tensor_dtype_to_np_dtype + except ImportError: + from onnx.mapping import TENSOR_TYPE_TO_NP_TYPE + + return np.dtype(TENSOR_TYPE_TO_NP_TYPE[tensor_type]) + + return np.dtype(onnx_tensor_dtype_to_np_dtype(tensor_type)) + + +def _try_load_external_initializer_array( + tensor: onnx.TensorProto, + model_path: str | Path | None, +) -> np.ndarray | None: + """Load a small external initializer from disk for runtime-query conditions. + + Returns None when the model path is unavailable, metadata is incomplete, + the sidecar file is missing, or the tensor is too large for the current + hash-based condition pipeline. + """ + if tensor.data_location != onnx.TensorProto.EXTERNAL or model_path is None: + return None + + location, offset, length = _get_external_tensor_info(tensor) + if location is None: + return None + + model_path = Path(model_path) + data_path = Path(location) + if not data_path.is_absolute(): + data_path = model_path.parent / data_path + if not data_path.exists(): + return None + + try: + np_dtype = _tensor_proto_dtype_to_np_dtype(tensor.data_type) + shape = tuple(tensor.dims) + numel = int(np.prod(shape)) if shape else 1 + expected_bytes = numel * np_dtype.itemsize + if length is not None and length < expected_bytes: + logger.debug( + "External initializer %s length %s is smaller than expected %s", + tensor.name, + length, + expected_bytes, + ) + return None + if expected_bytes > MAX_EXTERNAL_INITIALIZER_BYTES_FOR_QUERY: + logger.debug( + "Skipping external initializer %s for query conditions because %s bytes exceeds %s", + tensor.name, + expected_bytes, + MAX_EXTERNAL_INITIALIZER_BYTES_FOR_QUERY, + ) + return None + + with data_path.open("rb") as f: + f.seek(offset) + arr = np.fromfile(f, dtype=np_dtype, count=numel) + + if arr.size != numel: + logger.debug( + "External initializer %s only yielded %s elements, expected %s", + tensor.name, + arr.size, + numel, + ) + return None + + return arr.reshape(shape) + except Exception as e: + logger.debug( + "Failed to read external initializer %s from %s: %s", + tensor.name, + data_path, + e, + ) + return None + + def _get_pattern_type_var_conditions( pattern_match: PatternMatchResult, gen: Any, @@ -753,6 +848,7 @@ def get_query_conditions_for_node( domain: ONNXDomain, input_to_dq: dict[str, QDQTypeInfo], output_to_q: dict[str, QDQTypeInfo], + model_path: str | Path | None = None, dynamic_axis_strict_mode: bool = False, ) -> tuple[dict[str, Any], list[str], bool]: """Extract query conditions for runtime checking of an ONNX node. @@ -766,6 +862,7 @@ def get_query_conditions_for_node( domain: ONNX domain of the node. input_to_dq: Maps DQ output name -> QDQTypeInfo (for nodes consuming DQ outputs). output_to_q: Maps Q input name -> QDQTypeInfo (for nodes producing Q inputs). + model_path: Path to the source ONNX model when external data may need to be resolved. dynamic_axis_strict_mode: If False (default), maps any dynamic axes to (0,) for matching against first_axis test data. If True, preserves exact indices. @@ -855,8 +952,8 @@ def update_conditions_( input_name: str, is_variadic: bool, is_constant: bool, - shape: tuple | None = None, - value: tuple | None = None, + shape: tuple | list | None = None, + value: Any = None, ): dyn_axes = _compute_dynamic_axes(shape, is_constant) if is_variadic: @@ -927,8 +1024,31 @@ def update_conditions_( if inp_name in initializers: init = initializers[inp_name] - arr = numpy_helper.to_array(init) - update_conditions_(conditions, input_name, is_variadic, True, arr.shape, arr) + # External data initializers (large weights) may not have data loaded; + # keep the known shape but treat them as non-constant when the payload + # is unavailable so downstream rules do not assume *_value is usable. + # External-data tensors keep their payload out-of-line, so the typed + # TensorProto data fields should also be empty; raw_data is therefore + # the relevant signal for whether bytes were loaded into this proto. + external_arr = None + if init.data_location == onnx.TensorProto.EXTERNAL and not init.raw_data: + external_arr = _try_load_external_initializer_array(init, model_path) + + if external_arr is not None: + update_conditions_( + conditions, + input_name, + is_variadic, + True, + external_arr.shape, + external_arr, + ) + else: + shape = tuple(init.dims) if init.dims is not None else None + update_conditions_(conditions, input_name, is_variadic, False, shape, None) + else: + arr = numpy_helper.to_array(init) + update_conditions_(conditions, input_name, is_variadic, True, arr.shape, arr) conditions[f"{input_name}_is_none"] = False # Add type_vars info for initializers @@ -1102,6 +1222,7 @@ def __init__( model_proto: onnx.ModelProto, ep_name: str, device_type: str, + model_path: str | Path | None = None, dynamic_axis_strict_mode: bool = False, ) -> None: """Initialize runtime checker query. @@ -1110,11 +1231,13 @@ def __init__( model_proto: ONNX model proto ep_name: Execution provider name device_type: Device type (e.g., "CPU", "GPU", "NPU") + model_path: Path to the source ONNX model for resolving external data. dynamic_axis_strict_mode: If False (default), maps any dynamic axes to (0,) for matching against first_axis test data. If True, preserves exact dynamic axis indices. """ self.dynamic_axis_strict_mode = dynamic_axis_strict_mode + self.model_path = Path(model_path) if model_path is not None else None # Try shape inference: standard ONNX first, then symbolic (onnxruntime) try: # Standard ONNX shape inference — uses temp file for models @@ -1356,7 +1479,23 @@ def _build_single_node_model( if not inp_name: continue if inp_name in self.initializers: - graph_initializers.append(self.initializers[inp_name]) + init = self.initializers[inp_name] + # Skip external data initializers without loaded data; + # treat them as graph inputs instead so the model stays valid. + if init.data_location == onnx.TensorProto.EXTERNAL and not init.raw_data: + vi = self.valueinfo.get(inp_name) + if vi is not None: + graph_inputs.append(vi) + else: + # init.dims is a protobuf repeated field (always a sequence, never None). + # For external-data initializers the dims metadata is always present. + graph_inputs.append( + onnx.helper.make_tensor_value_info( + inp_name, init.data_type, list(init.dims) + ) + ) + else: + graph_initializers.append(init) elif inp_name in self.constants: # Convert Constant node output to initializer graph_initializers.append(self.constants[inp_name]) @@ -1433,18 +1572,31 @@ def _generate_node_inputs(self, node: onnx.NodeProto) -> dict[str, np.ndarray]: for inp_name in node.input: if not inp_name: continue - # Skip initializers and constants - they are embedded in the model - if inp_name in self.initializers or inp_name in self.constants: + # Skip initializers and constants - they are embedded in the model. + # Exception: external-data initializers without loaded raw_data are + # converted to graph inputs by _build_single_node_model, so we must + # generate dummy data for them. + if inp_name in self.initializers: + init = self.initializers[inp_name] + if not (init.data_location == onnx.TensorProto.EXTERNAL and not init.raw_data): + continue + elif inp_name in self.constants: continue vi = self.valueinfo.get(inp_name) if vi is None: - raise ValueError( - f"Input '{inp_name}' for node '{node.name}' ({node.op_type}) " - f"not found in valueinfo" - ) + init = self.initializers.get(inp_name) + if init is not None and init.data_location == onnx.TensorProto.EXTERNAL: + shape = tuple(init.dims) + dtype_str = dtype_from_tensorproto_enum(init.data_type) + else: + raise ValueError( + f"Input '{inp_name}' for node '{node.name}' ({node.op_type}) " + f"not found in valueinfo" + ) + else: + shape, dtype_str = shape_and_dtype_from_valueinfo(vi) - shape, dtype_str = shape_and_dtype_from_valueinfo(vi) if dtype_str is None: raise ValueError( f"Input '{inp_name}' for node '{node.name}' ({node.op_type}) " @@ -1932,6 +2084,7 @@ def get_pattern_id(is_qdq): op_domain, self.input_to_dq_type, self.output_to_q_type, + model_path=self.model_path, dynamic_axis_strict_mode=self.dynamic_axis_strict_mode, ) except ( diff --git a/tests/unit/analyze/core/test_runtime_checker.py b/tests/unit/analyze/core/test_runtime_checker.py index 35cb6de4d..c4901a5bf 100644 --- a/tests/unit/analyze/core/test_runtime_checker.py +++ b/tests/unit/analyze/core/test_runtime_checker.py @@ -13,11 +13,15 @@ """ import time +from pathlib import Path +import numpy as np +import onnx import pytest -from onnx import TensorProto, helper from winml.modelkit.analyze import ONNXModel, RuntimeChecker, RuntimeTestResult +from winml.modelkit.analyze.core import runtime_checker_query as runtime_checker_query_module +from winml.modelkit.analyze.core.runtime_checker_query import RuntimeCheckerQuery from winml.modelkit.analyze.models.runtime_checks import ( # Testing internal implementation AlternativeType, PatternAlternative, @@ -31,6 +35,10 @@ ) +TensorProto = onnx.TensorProto +helper = onnx.helper + + @pytest.fixture def simple_onnx_model() -> ONNXModel: """Create a simple ONNX model for testing.""" @@ -288,6 +296,95 @@ def test_full_workflow_with_model(self, simple_onnx_model: ONNXModel): assert "op_runtime_check_result" in summary assert len(summary["op_runtime_check_result"]) == len(op_results) + def test_op_support_handles_graph_only_external_initializer( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Graph-only external-data initializers survive the analyzer runtime-check path.""" + + weight = onnx.numpy_helper.from_array(np.zeros((2,), dtype=np.float32), name="weight") + weight.data_location = onnx.TensorProto.EXTERNAL + weight.ClearField("raw_data") + weight.external_data.add(key="location", value="weight.bin") + + input_value_info = helper.make_tensor_value_info("input", TensorProto.FLOAT, [2]) + output_value_info = helper.make_tensor_value_info("output", TensorProto.FLOAT, [2]) + add_node = helper.make_node("Add", ["weight", "input"], ["output"], name="add_node") + graph = helper.make_graph( + [add_node], + "external_initializer_graph", + [input_value_info], + [output_value_info], + initializer=[weight], + ) + model = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + model_path = tmp_path / "external_initializer.onnx" + onnx.save(model, model_path) + graph_only_model = onnx.load(str(model_path), load_external_data=False) + onnx_model = ONNXModel.from_onnx_model(graph_only_model, str(model_path)) + + checker = RuntimeChecker( + ep="CPUExecutionProvider", + device="CPU", + model=onnx_model, + ) + query = checker._get_query() + + captured_calls: list[tuple[str, bytes, dict[str, np.ndarray]]] = [] + + class FakeRunner: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + return False + + def run(self, fn, *args): + return {"result": fn(*args), "stdout": "", "stderr": ""} + + class FakeEPChecker: + def check_compile(self, model_bytes, input_feed): + captured_calls.append(("compile", model_bytes, input_feed)) + return {"success": True} + + def check_run(self, model_bytes, input_feed): + captured_calls.append(("run", model_bytes, input_feed)) + return {"success": True} + + monkeypatch.setattr(runtime_checker_query_module, "ResilientRunner", FakeRunner) + + query.ep_neg_rules = {} + query.df_tables = {} + monkeypatch.setattr(RuntimeCheckerQuery, "_is_ep_available_locally", lambda self: True) + monkeypatch.setattr( + RuntimeCheckerQuery, + "_get_ep_checker", + lambda self: FakeEPChecker(), + ) + + results = checker.op_support(run_unknown_op=True) + + assert len(results) == 1 + assert results[0].pattern_id == "OP/ai.onnx/Add" + assert results[0].result.compile is True + assert results[0].result.run is True + assert results[0].result.no_data is False + assert [phase for phase, _, _ in captured_calls] == ["compile", "run"] + + for _, model_bytes, input_feed in captured_calls: + assert set(input_feed) == {"weight", "input"} + assert input_feed["weight"].shape == (2,) + assert input_feed["weight"].dtype == np.float32 + assert input_feed["input"].shape == (2,) + + single_node_model = onnx.ModelProto() + single_node_model.ParseFromString(model_bytes) + assert {vi.name for vi in single_node_model.graph.input} == {"weight", "input"} + assert {init.name for init in single_node_model.graph.initializer} == set() + def test_full_workflow_with_patterns( self, sample_pattern_match: PatternMatchResult, simple_onnx_model: ONNXModel ): diff --git a/tests/unit/analyze/core/test_runtime_checker_query_helpers.py b/tests/unit/analyze/core/test_runtime_checker_query_helpers.py index 1490ccbd6..9f2c57610 100644 --- a/tests/unit/analyze/core/test_runtime_checker_query_helpers.py +++ b/tests/unit/analyze/core/test_runtime_checker_query_helpers.py @@ -5,10 +5,24 @@ """Unit tests for runtime checker query helper functions.""" +from pathlib import Path + +import numpy as np +import onnx import pytest +from onnx import TensorProto, helper -from winml.modelkit.analyze.core.runtime_checker_query import _build_table_filter_conditions +from winml.modelkit.analyze.core import runtime_checker_query as runtime_checker_query_module +from winml.modelkit.analyze.core.runtime_checker_query import ( + RuntimeCheckerQuery, + _build_table_filter_conditions, + _try_load_external_initializer_array, + get_query_conditions_for_node, + node_to_pattern_match, +) from winml.modelkit.analyze.exceptions import OpOptionalInputSupportError +from winml.modelkit.analyze.utils.model_utils import DUMMY_FLOAT +from winml.modelkit.onnx import ONNXDomain class TestBuildTableFilterConditions: @@ -79,3 +93,219 @@ def test_returns_empty_dict_for_empty_column_names(self): ) assert result == {} + + +class TestGetQueryConditionsForNode: + """Test condition extraction for runtime rule lookups.""" + + def test_external_initializer_without_payload_is_not_marked_constant(self): + """External-data initializers without loaded values keep shape but not constant status.""" + node = helper.make_node("Add", ["weight", "input"], ["output"], name="add_node") + input_value_info = helper.make_tensor_value_info("input", TensorProto.FLOAT, [2]) + + external_initializer = onnx.TensorProto() + external_initializer.name = "weight" + external_initializer.data_type = TensorProto.FLOAT + external_initializer.dims.extend([2]) + external_initializer.data_location = TensorProto.EXTERNAL + external_initializer.external_data.add(key="location", value="weight.bin") + + conditions, infinite_properties, is_qdq = get_query_conditions_for_node( + node=node, + opset_version=17, + valueinfo={"input": input_value_info}, + initializers={"weight": external_initializer}, + constants={}, + domain=ONNXDomain.AI_ONNX, + input_to_dq={}, + output_to_q={}, + ) + + assert conditions["A_is_constant"] is False + assert conditions["A_shape"] == (2,) + assert conditions["A_value"] is None + assert conditions["A_is_fixed_shape"] is True + assert conditions["A_dynamic_axes"] == () + assert conditions["A_is_none"] is False + assert infinite_properties == ["A_shape", "B_shape"] + assert is_qdq is False + + def test_external_initializer_sidecar_is_loaded_when_model_path_is_available( + self, + tmp_path: Path, + ) -> None: + """Small external-data initializers can be resolved from the model sidecar.""" + weight = onnx.numpy_helper.from_array( + np.array([1.5, -2.0], dtype=np.float32), + name="weight", + ) + node = helper.make_node("Add", ["weight", "input"], ["output"], name="add_node") + input_value_info = helper.make_tensor_value_info("input", TensorProto.FLOAT, [2]) + output_value_info = helper.make_tensor_value_info("output", TensorProto.FLOAT, [2]) + graph = helper.make_graph( + [node], + "external_initializer_graph", + [input_value_info], + [output_value_info], + initializer=[weight], + ) + model = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + model_path = tmp_path / "external_initializer.onnx" + onnx.save_model( + model, + model_path, + save_as_external_data=True, + all_tensors_to_one_file=True, + location="weights.bin", + size_threshold=0, + ) + graph_only_model = onnx.load(str(model_path), load_external_data=False) + + conditions, infinite_properties, is_qdq = get_query_conditions_for_node( + node=node, + opset_version=17, + valueinfo={"input": input_value_info}, + initializers={"weight": graph_only_model.graph.initializer[0]}, + constants={}, + domain=ONNXDomain.AI_ONNX, + input_to_dq={}, + output_to_q={}, + model_path=model_path, + ) + + assert conditions["A_is_constant"] is True + assert conditions["A_shape"] == (2,) + assert conditions["A_value"] == (DUMMY_FLOAT, DUMMY_FLOAT) + assert conditions["A_is_none"] is False + assert infinite_properties == ["A_shape", "B_shape"] + assert is_qdq is False + + def test_try_load_external_initializer_array_returns_plain_ndarray( + self, + tmp_path: Path, + ) -> None: + """Loaded sidecar tensors are copied into memory and do not retain file handles.""" + weight = onnx.numpy_helper.from_array( + np.array([[1.5, -2.0]], dtype=np.float32), + name="weight", + ) + node = helper.make_node("Identity", ["weight"], ["output"], name="identity_node") + output_value_info = helper.make_tensor_value_info("output", TensorProto.FLOAT, [1, 2]) + graph = helper.make_graph( + [node], + "external_initializer_graph", + [], + [output_value_info], + initializer=[weight], + ) + model = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + model_path = tmp_path / "external_initializer.onnx" + onnx.save_model( + model, + model_path, + save_as_external_data=True, + all_tensors_to_one_file=True, + location="weights.bin", + size_threshold=0, + ) + graph_only_model = onnx.load(str(model_path), load_external_data=False) + + loaded = _try_load_external_initializer_array( + graph_only_model.graph.initializer[0], + model_path, + ) + + assert loaded is not None + assert isinstance(loaded, np.ndarray) + assert not isinstance(loaded, np.memmap) + assert not isinstance(getattr(loaded, "base", None), np.memmap) + assert np.array_equal(loaded, np.array([[1.5, -2.0]], dtype=np.float32)) + + sidecar_path = tmp_path / "weights.bin" + renamed_sidecar_path = tmp_path / "weights-renamed.bin" + sidecar_path.rename(renamed_sidecar_path) + assert renamed_sidecar_path.exists() + + +class TestLocalEPFallback: + """Test local EP fallback helpers for single-node execution.""" + + def test_local_ep_check_feeds_promoted_external_initializer(self, monkeypatch): + """Promoted external-data initializers are included in the local EP input feed.""" + node = helper.make_node("Add", ["weight", "input"], ["output"], name="add_node") + input_value_info = helper.make_tensor_value_info("input", TensorProto.FLOAT, [2]) + output_value_info = helper.make_tensor_value_info("output", TensorProto.FLOAT, [2]) + + external_initializer = onnx.TensorProto() + external_initializer.name = "weight" + external_initializer.data_type = TensorProto.FLOAT + external_initializer.dims.extend([2]) + external_initializer.data_location = TensorProto.EXTERNAL + external_initializer.external_data.add(key="location", value="weight.bin") + + graph = helper.make_graph( + [node], + "external_initializer_graph", + [input_value_info], + [output_value_info], + initializer=[external_initializer], + ) + model = helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + query = RuntimeCheckerQuery(model, ep_name="CPUExecutionProvider", device_type="CPU") + + captured_calls = [] + + class FakeRunner: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + return False + + def run(self, fn, *args): + return {"result": fn(*args), "stdout": "", "stderr": ""} + + class FakeEPChecker: + def check_compile(self, model_bytes, input_feed): + captured_calls.append(("compile", model_bytes, input_feed)) + return {"success": True} + + def check_run(self, model_bytes, input_feed): + captured_calls.append(("run", model_bytes, input_feed)) + return {"success": True} + + monkeypatch.setattr(runtime_checker_query_module, "ResilientRunner", FakeRunner) + monkeypatch.setattr(RuntimeCheckerQuery, "_is_ep_available_locally", lambda self: True) + monkeypatch.setattr( + RuntimeCheckerQuery, + "_get_ep_checker", + lambda self: FakeEPChecker(), + ) + + result = query._try_local_ep_check( + node=node, + op_domain=ONNXDomain.AI_ONNX, + opset_version=17, + pattern_match=node_to_pattern_match(node), + node_tags=[], + fallback_reason="rules_not_found", + ) + + assert result is not None + assert result.result.compile is True + assert result.result.run is True + assert [phase for phase, _, _ in captured_calls] == ["compile", "run"] + + for _, model_bytes, input_feed in captured_calls: + assert set(input_feed) == {"weight", "input"} + assert input_feed["weight"].shape == (2,) + assert input_feed["weight"].dtype == np.float32 + assert input_feed["input"].shape == (2,) + + single_node_model = onnx.ModelProto() + single_node_model.ParseFromString(model_bytes) + assert {vi.name for vi in single_node_model.graph.input} == {"weight", "input"} + assert {init.name for init in single_node_model.graph.initializer} == set()