File size: 4,306 Bytes
ed3aeeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | from __future__ import annotations
import hashlib
import xml.etree.ElementTree as ET
from pathlib import Path
import pytest
from scripts import build_mlir_ir_graphs as builder
from scripts import mlir_graph_common as common
def digest(path: Path) -> str:
value = hashlib.sha256()
value.update(path.read_bytes())
return value.hexdigest()
def write_mlir(tmp_path: Path, text: str, name: str = "fixture.mlir") -> Path:
path = tmp_path / name
path.write_text(text.strip() + "\n", encoding="utf-8")
return path
def test_onnx_ssa_dependency_and_static_order(tmp_path: Path) -> None:
path = write_mlir(
tmp_path,
"""
module {
func.func @main(%arg0: tensor<1xf32>) -> tensor<1xf32> {
%relu = "onnx.Relu"(%arg0) {onnx_node_name = "Relu_0"} : (tensor<1xf32>) -> tensor<1xf32>
return %relu : tensor<1xf32>
}
}
""",
)
parsed = common.parse_mlir(path, "TEST:fp32:ONNX")
relu = next(row for row in parsed.operations if row["operation"] == "onnx.Relu")
returned = next(row for row in parsed.operations if row["operation"] == "func.return")
assert relu["results"] == ["%relu"]
assert returned["operands"] == ["%relu"]
assert parsed.unresolved_use_count == 0
assert parsed.duplicate_definition_count == 0
assert parsed.producer_after_consumer_count == 0
svg, metadata = common.render_execution_dependency_svg(
parsed,
title="ONNX primary fixture",
graph_id="TEST:fp32:ONNX",
stage="ONNX",
)
ET.fromstring(svg)
assert metadata["rendered_operation_nodes"] == len(parsed.operations)
assert "STATIC_MLIR_PROGRAM_ORDER" in svg
def test_grouped_results_expand_and_resolve(tmp_path: Path) -> None:
path = write_mlir(
tmp_path,
"""
module {
func.func @main(%arg0: tensor<4xf32>) {
%packed : 2 = "onnx.Split"(%arg0) : (tensor<4xf32>) -> (tensor<2xf32>, tensor<2xf32>)
"onnx.Add"(%packed#0, %packed#1) : (tensor<2xf32>, tensor<2xf32>) -> ()
return
}
}
""",
)
parsed = common.parse_mlir(path, "GROUPED:public_quantized:ONNX")
split = next(row for row in parsed.operations if row["operation"] == "onnx.Split")
assert split["results"] == ["%packed#0", "%packed#1"]
assert parsed.unresolved_use_count == 0
def test_primary_only_selection_is_exactly_21_pairs_42_graphs(tmp_path: Path) -> None:
model_ids = [f"M{index:02d}" for index in range(21)]
rows: list[dict[str, str]] = []
for model_id in model_ids:
for variant in common.VARIANTS:
directory = tmp_path / "models" / model_id / variant
directory.mkdir(parents=True)
onnx_path = directory / "onnx.mlir"
onnx_path.write_text("module {}\n", encoding="utf-8")
rows.append(
{
"model_id": model_id,
"task": "fixture",
"variant": variant,
"onnx_status": "PASS",
"onnx_artifact": str(onnx_path.relative_to(tmp_path)),
"onnx_sha256": digest(onnx_path),
"affine_scf_memref_status": "FAIL",
"affine_scf_memref_artifact": "NONE",
"affine_scf_memref_sha256": "NONE",
"krnl_status": "FAIL",
"krnl_artifact": "NONE",
"krnl_sha256": "NONE",
"llvm_status": "FAIL",
"llvm_artifact": "NONE",
"llvm_sha256": "NONE",
"last_fully_successful_ir": "ONNX",
}
)
by_key = builder.require_matrix_rows(rows)
specs, pairs = builder.build_specs(by_key, tmp_path, primary_only=True)
assert len(specs) == 42
assert len({row["graph_id"] for row in specs}) == 42
assert {row["stage"] for row in specs} == {"ONNX"}
assert {row["graph_role"] for row in specs} == {"PRIMARY_ONNX_DIALECT"}
assert len(pairs) == 21
assert {row["pair_common_stage"] for row in pairs} == {"ONNX"}
assert {row["supplemental_lower_graphs"] for row in pairs} == {
"0 (primary ONNX reused as common view)"
}
with pytest.raises(ValueError, match="duplicate IR coverage row"):
builder.require_matrix_rows([*rows, dict(rows[0])])
|