Skip to content
Open
Show file tree
Hide file tree
Changes from all 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
174 changes: 174 additions & 0 deletions tests/openvino/transformations/test_transformations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
import os
import sys
import unittest



# 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],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
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 key in applied_norm:
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()