Skip to content
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions tests/openvino/transformations/expected_transformations.txt
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@

# Format: arch_name: Transformation1, Transformation2


afmoe: MoEMatMulsFusion,ConvertToCPUSpecificOpset
gpt2: ConvertToCPUSpecificOpset,MatMulToFCFusion
173 changes: 173 additions & 0 deletions tests/openvino/transformations/test_transformations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,173 @@
import os
import sys
import unittest

os.environ["OV_ENABLE_PROFILE_PASS"] = "1"
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think setting OV_ENABLE_PROFILE_PASS at the top level might unintentionally leak into the global test environment . This variable is already being set inside the subprocess - line 49 . we can safely remove this line .


# we are adding this , so the parent directory (tests/openvino/) is in the python search path for utils_test.py to be imported
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))


import subprocess
import textwrap
import re
from difflib import get_close_matches

from parameterized import parameterized
from utils_tests import MODEL_NAMES, OPENVINO_DEVICE, REMOTE_CODE_MODELS





# Maps architecture name -> list of transformation needed to be applied , as per expected_transformations.txt
def _load_expected_transformations(path):
result = {}
with open(path) as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
arch, _, transforms_str = line.partition(":")
result[arch.strip()] = [
t.strip() for t in transforms_str.split(",") if t.strip()
]
return result


_CONFIG_PATH = os.path.join(
os.path.dirname(__file__), "expected_transformations.txt"
)
ARCH_TO_EXPECTED_TRANSFORMATIONS = _load_expected_transformations(_CONFIG_PATH)


def _capture_stderr_during(model_id, OPENVINO_DEVICE, trust_remote_code):
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the subprocess code here may be quite expensive since it invokes OVModelForCausalLM.from_pretrained(...) inside the string, which can trigger model export on every test run.

# Runs model loading in a subprocess to reliably capture OpenVINO C++ logs.

code = textwrap.dedent(f"""
import os
os.environ["OV_ENABLE_PROFILE_PASS"] = "1"

from optimum.intel import OVModelForCausalLM

OVModelForCausalLM.from_pretrained(
"{model_id}",
export=True,
compile=True,
device="{OPENVINO_DEVICE}",
trust_remote_code={trust_remote_code},
)
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can add this param

cache_dir="./ov_cache"

we could pass a cache_dir so that models are reused across runs instead of being re-exported each time.

This should help make the test more scalable.

""")

result = subprocess.run(
[sys.executable, "-c", code],
capture_output=True,
text=True,
)

return result.stdout
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should use stderr instead of stdout since OpenVINO logs are emitted to stderr
even function name says capture stderr

- return result.stdout
+ return result.stderr



# Remove separators and lowercase for fuzzy comparison.
def normalize(name: str) -> str:
return re.sub(r'[\s_\-]', '', name).lower()


# Extract transformation name — always last token before NUMBERms +/-
def extract_transform_name(line: str) -> str | None:
match = re.search(
r'([A-Za-z][A-Za-z0-9_]*)\s+\d+ms\s*[+-]\s*$',
line.strip()
)
return match.group(1) if match else None


# Algo to identify tranformations present with '+' in the log.
def check_failed_transformations(log: str, words: list[str]) -> dict:
applied_raw = []
applied_norm = []

for line in log.splitlines():
stripped = line.strip()

if not stripped:
continue

if not stripped.endswith('+'): # neglect '-' because those transformations are not applied
continue

name = extract_transform_name(stripped)
if name:
applied_raw.append(name)
applied_norm.append(normalize(name))

remaining = {normalize(w): w for w in words}

for key in list(remaining.keys()):
if any(key in a for a in applied_norm):
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think substring matching here could cause false positives like MatMul matching MatMulFusionExtra
Using exact matching like if key in applied_norm: would be safer.

Copy link
Copy Markdown
Author

@alien-cyber alien-cyber Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @MissLostCodes
Thanks for the analysis.

  • Now the string matching function is Fixed.
  • stderr and stdout are captured in the same place.
  • unnecessary OV_ENABLE_PROFILE_PASS is removed

Regarding cache_dir="./ov_cache", I believe it is not necessary in this context, as it mainly changes the default directory used for caching model files. The default Hugging Face cache should already handle reuse of downloaded model configuration.

del remaining[key]

hints = {}
for key, original in remaining.items():
matches = get_close_matches(key, applied_norm, n=2, cutoff=0.8)

if matches:
readable = [
applied_raw[applied_norm.index(m)]
for m in matches
]
hints[original] = readable

return {
"not_found": list(remaining.values()),
"hints": hints
}


class OVTransformationTest(unittest.TestCase):

@parameterized.expand(
list(ARCH_TO_EXPECTED_TRANSFORMATIONS.items())
)
def test_transformations_applied(
self,
model_arch,
expected_transforms
):
model_id = MODEL_NAMES[model_arch]
trust_remote_code = model_arch in REMOTE_CODE_MODELS

log_output = _capture_stderr_during(
model_id,
OPENVINO_DEVICE,
trust_remote_code
)

result = check_failed_transformations(
log_output,
expected_transforms
)

if result["not_found"]:
not_found = ", ".join(result["not_found"])
hints = result["hints"]

hint_lines = ""

if hints:
hint_lines = (
"\nPossible matches in log:\n"
+ "\n".join(
f" '{wrong}' → did you mean '{', '.join(suggestions)}'?"
for wrong, suggestions in hints.items()
)
)

raise AssertionError(
f"The following transformations were not applied for '{model_arch}' architecture: "
f"{not_found}{hint_lines}"
)


if __name__ == "__main__":
unittest.main()
Loading