[Bug][CPU] QDQ MatMul fusion accepts INT32 activations unsupported by MatMulIntegerToFloat #33437

import copy
import hashlib
import json
import tempfile
from pathlib import Path
import numpy as np
import onnx
import onnxruntime as ort
from onnx import helper as h, numpy_helper as nh

def run(model, feeds, optimize=False, disabled=(), export=None):
    options = ort.SessionOptions()
    options.graph_optimization_level = (ort.GraphOptimizationLevel.ORT_ENABLE_ALL
        if optimize else ort.GraphOptimizationLevel.ORT_DISABLE_ALL)
    options.intra_op_num_threads = options.inter_op_num_threads = 1
    options.log_severity_level = 4
    if export is not None:
        options.optimized_model_filepath = str(export)
    session = ort.InferenceSession(model.SerializeToString(), options,
        providers=["CPUExecutionProvider"], disabled_optimizers=set(disabled))
    assert [v.name for v in session.get_outputs()] == [v.name for v in model.graph.output]
    return session.run(None, feeds)

def exact(left, right):
    return len(left) == len(right) and all(isinstance(a, np.ndarray) and isinstance(b, np.ndarray)
        and a.shape == b.shape and a.dtype == b.dtype and np.array_equal(a, b)
        for a, b in zip(left, right))

def describe(values):
    return [None if a is None else {"shape": list(a.shape), "dtype": str(a.dtype),
        "sample": a.reshape(-1)[:8].tolist()} for a in values]

def make_source():
    nodes = [
        h.make_node('DequantizeLinear', ['a', 's', 'az'], ['af']),
        h.make_node('DequantizeLinear', ['b', 's', 'bz'], ['bf']),
        h.make_node('MatMul', ['af', 'bf'], ['y']),
    ]
    initializers = [
        nh.from_array(np.array(1.0, dtype='float32'), 's'),
        nh.from_array(np.array(0, dtype='int32'), 'az'),
        nh.from_array(np.array([[3], [4]], dtype='int8'), 'b'),
        nh.from_array(np.array(0, dtype='int8'), 'bz'),
    ]
    return h.make_model(h.make_graph(nodes, "repro", [h.make_tensor_value_info('a', 6, [1, 2])],
        [h.make_tensor_value_info('y', 1, [1, 1])], initializers),
        opset_imports=[h.make_opsetid('', 13)], ir_version=10)

source = make_source()
onnx.checker.check_model(source, full_check=True)
onnx.shape_inference.infer_shapes(source, strict_mode=True)
feeds = {'a': np.array([[1, 2]], dtype='int32')}
expected = [np.array([[11.0]], dtype='float32')]
before = run(source, feeds)
assert exact(before, expected), (describe(before), describe(expected))
guarded = run(source, feeds, optimize=True, disabled=('QDQSelectorActionTransformer',))
assert exact(before, guarded)
try:
    run(source, feeds, optimize=True)
except Exception as error:
    assert all(token in str(error) for token in ['MatMulIntegerToFloat', 'int32']), str(error)
    print(type(error).__name__, str(error).split("Status Message:")[0])
else:
    raise AssertionError("The reported optimization failure did not reproduce")
print("source_sha256", hashlib.sha256(source.SerializeToString()).hexdigest())
print("ort", ort.__version__, ort.get_build_info())
print("REPRODUCED")