From 7211d1862dd62bee21e480e3f5bab506488f7e63 Mon Sep 17 00:00:00 2001 From: Egor Tiuvaev Date: Mon, 15 Jun 2026 13:02:43 +0200 Subject: [PATCH 1/3] memory_tests: update reporting --- tests/memory_tests/tools/run_tests.py | 89 ++++++++++++++++++--------- 1 file changed, 60 insertions(+), 29 deletions(-) diff --git a/tests/memory_tests/tools/run_tests.py b/tests/memory_tests/tools/run_tests.py index c65902385f6c..7acd3099e278 100644 --- a/tests/memory_tests/tools/run_tests.py +++ b/tests/memory_tests/tools/run_tests.py @@ -211,10 +211,21 @@ def modelid_assume_info(modelid): return modelname, framework, precision +@dataclass +class TestCase: + model_id: str + model_path: str + device: str + + # metadata + original_model_path: str + weights_size: int + + class TestSession: def __init__(self, executable, ir_cache_dirs, devices, api=None, report_reference=False): self.executable = executable - self.test_name = executable.rsplit("/", 1)[-1].removesuffix(".exe").removeprefix("test_") + self.test_name = executable.replace("\\", "/").rsplit("/", 1)[-1].removesuffix(".exe").removeprefix("test_") self.ir_cache_dirs = ir_cache_dirs self.devices = devices self.report_api = api @@ -246,12 +257,12 @@ def api(self, method, data=None, **kwargs): print(f"API Error: {response.text}") return response.json() - def api_push_test_result(self, model_path, modelid, weights_size, device, result): + def api_push_test_result(self, test: TestCase, result): if not self.report_metadata: print("No job metadata found, no report will be made.") return - model_assumptions = modelid_assume_info(modelid) - modelname, framework, precision = model_assumptions or (modelid, "unknown", "unknown") + model_assumptions = modelid_assume_info(test.model_id) + modelname, framework, precision = model_assumptions or (test.model_id, "unknown", "unknown") test_report = [] sample_names = result.get("samples", {}).keys() or self.test_info["samples"] for sname in sample_names: @@ -260,16 +271,17 @@ def api_push_test_result(self, model_path, modelid, weights_size, device, result sample_report.update({ "test_name": f"{result.get('test', self.test_name)}:{sname}", "status": "failed" if "error" in result else "passed", - "source": model_path, + "source": test.model_path, "log": result.get("stderr", ""), "model_name": modelname, - "model": modelid, - "device": result.get("device") or device, + "model": test.model_id, + "device": result.get("device") or test.device, "framework": framework, "precision": precision, "metrics": sample.as_dict(), "cpu_family": CPU_FAMILY, - "model_size": weights_size + "model_size": test.weights_size, + "ext": {"original_model_source": test.original_model_path} }) test_report.append(sample_report) response = attempt(self.api, "v2/memory/push-2-db-facade", {"data": test_report}) @@ -317,31 +329,51 @@ def scan_directory(self, directory: Path): new_files = sorted(new_files) yield from ((path.removeprefix(cache_dir).replace("\\", "/"), path) for path in new_files) + def get_description(self, model_path) -> dict[str, str] | None: + model_dir = os.path.dirname(model_path) + description_path = f"{model_dir}/description.txt" + try: + with open(description_path) as infile: + return dict([ + item.strip("\r\n").split(":", 1) + for item in infile.readlines() + if ":" in item + ]) + except (IOError, OSError): + return None + def generate_test_cases(self): - def _with_filesize(paths): - for (modelid, path) in paths: - weights_path, _ = os.path.splitext(path) + for ir_cache_dir in self.ir_cache_dirs: + for (model_id, model_path) in self.scan_directory(ir_cache_dir): + weights_path, _ = os.path.splitext(model_path) weights_path = f"{weights_path}.bin" if os.path.isfile(weights_path): weights_size = os.path.getsize(weights_path) else: - # weights file does not exist -> invalid test case + print(f"Warning: Test case {model_id} invalid: can't find weights file at {weights_path}") continue - yield modelid, path, weights_size - for ir_cache_dir in self.ir_cache_dirs: - yield from itertools.product( - _with_filesize(self.scan_directory(ir_cache_dir)), - self.devices - ) - - def run_test_case(self, model_path, device): + model_description = self.get_description(model_path) + original_model_path = "" + if model_description: + original_model_path = model_description.get("src_model_path", "") + for device in self.devices: + yield TestCase( + model_id=model_id, + model_path=model_path, + device=device, + original_model_path=original_model_path, + weights_size=weights_size, + ) + + def run_test_case(self, test_case: TestCase): try: - return run_test_executable_extract_result([self.executable, model_path, device]) + return run_test_executable_extract_result([ + self.executable, test_case.model_path, test_case.device]) except Exception as ex: print(f" When running test an unexpected error happened: {ex}") return {"error": "unexpected error", "exception": ex} - def handle_test_result(self, modelid, weights_size, device, result): + def handle_test_result(self, test: TestCase, result): base2_suffixes = ["bytes", "KiB", "MiB", "GiB", "TiB", "PiB"] def _base2_human_readable(number): @@ -354,8 +386,8 @@ def _base2_human_readable(number): return f"{number} {suffix}" status = "error" if "error" in result else "ok" - weights_size_human_read = _base2_human_readable(weights_size) - print(f"TEST {modelid} ({weights_size_human_read}) x {device}: {status}") + weights_size_human_read = _base2_human_readable(test.weights_size) + print(f"TEST {test.model_id} ({weights_size_human_read}) x {test.device}: {status}") if status == "error": error = result.get("error") stdout = result.get("stdout") @@ -375,11 +407,10 @@ def _base2_human_readable(number): sys.stderr.flush() def run(self): - for (modelid, model_path, weights_size), device in self.generate_test_cases(): - result = self.run_test_case(model_path, device) - test_name = result.get("test", self.test_name) - self.api_push_test_result(model_path, modelid, weights_size, device, result) - self.handle_test_result(modelid, weights_size, device, result) + for test in self.generate_test_cases(): + result = self.run_test_case(test) + self.api_push_test_result(test, result) + self.handle_test_result(test, result) if __name__ == "__main__": From adb193b0b620642e00ea81aa0232c31c8986c8a8 Mon Sep 17 00:00:00 2001 From: Egor Tiuvaev Date: Tue, 16 Jun 2026 17:45:57 +0200 Subject: [PATCH 2/3] fix description extraction --- tests/memory_tests/tools/run_tests.py | 47 ++++++++++++++------------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/tests/memory_tests/tools/run_tests.py b/tests/memory_tests/tools/run_tests.py index 7acd3099e278..79d9598bab3a 100644 --- a/tests/memory_tests/tools/run_tests.py +++ b/tests/memory_tests/tools/run_tests.py @@ -281,7 +281,7 @@ def api_push_test_result(self, test: TestCase, result): "metrics": sample.as_dict(), "cpu_family": CPU_FAMILY, "model_size": test.weights_size, - "ext": {"original_model_source": test.original_model_path} + "ext": {"originalSource": test.original_model_path} }) test_report.append(sample_report) response = attempt(self.api, "v2/memory/push-2-db-facade", {"data": test_report}) @@ -329,37 +329,40 @@ def scan_directory(self, directory: Path): new_files = sorted(new_files) yield from ((path.removeprefix(cache_dir).replace("\\", "/"), path) for path in new_files) - def get_description(self, model_path) -> dict[str, str] | None: - model_dir = os.path.dirname(model_path) - description_path = f"{model_dir}/description.txt" + def get_description(self, model_path: Path) -> dict[str, str] | None: + description_path = model_path.parent / "description.txt" + description_text = None try: - with open(description_path) as infile: - return dict([ - item.strip("\r\n").split(":", 1) - for item in infile.readlines() - if ":" in item - ]) - except (IOError, OSError): - return None + description_text = description_path.read_text() + except OSError: + pass + if not description_text: + description_path = model_path.parent.parent / "description.txt" + try: + description_text = description_path.read_text() + except OSError: + return None + return dict([ + line.split(":", 1) + for line in description_text.splitlines() + if ":" in line + ]) def generate_test_cases(self): for ir_cache_dir in self.ir_cache_dirs: for (model_id, model_path) in self.scan_directory(ir_cache_dir): - weights_path, _ = os.path.splitext(model_path) - weights_path = f"{weights_path}.bin" - if os.path.isfile(weights_path): - weights_size = os.path.getsize(weights_path) - else: + model_path = Path(model_path) + weights_path = model_path.with_suffix(".bin") + if not weights_path.is_file(): print(f"Warning: Test case {model_id} invalid: can't find weights file at {weights_path}") continue - model_description = self.get_description(model_path) - original_model_path = "" - if model_description: - original_model_path = model_description.get("src_model_path", "") + weights_size = weights_path.stat().st_size + model_description = self.get_description(model_path) or {} + original_model_path = model_description.get("src_model_path", "") for device in self.devices: yield TestCase( model_id=model_id, - model_path=model_path, + model_path=str(model_path), device=device, original_model_path=original_model_path, weights_size=weights_size, From a56152a528cb10fc2afd13f000b75cb1d28e3a64 Mon Sep 17 00:00:00 2001 From: Egor Tiuvaev Date: Fri, 19 Jun 2026 15:09:10 +0200 Subject: [PATCH 3/3] Remove unnecessary raw path manipulations --- tests/memory_tests/tools/run_tests.py | 32 +++++++++++++++------------ 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/tests/memory_tests/tools/run_tests.py b/tests/memory_tests/tools/run_tests.py index 79d9598bab3a..572b1dcc1baa 100644 --- a/tests/memory_tests/tools/run_tests.py +++ b/tests/memory_tests/tools/run_tests.py @@ -214,18 +214,22 @@ def modelid_assume_info(modelid): @dataclass class TestCase: model_id: str - model_path: str + model_path: Path device: str # metadata - original_model_path: str + description: dict[str, str] weights_size: int + @property + def original_model(self) -> str | None: + return self.description.get("src_model_path") + class TestSession: - def __init__(self, executable, ir_cache_dirs, devices, api=None, report_reference=False): + def __init__(self, executable: Path, ir_cache_dirs: list[Path], devices: list[str], api=None, report_reference=False): self.executable = executable - self.test_name = executable.replace("\\", "/").rsplit("/", 1)[-1].removesuffix(".exe").removeprefix("test_") + self.test_name = executable.stem.removeprefix("test_") self.ir_cache_dirs = ir_cache_dirs self.devices = devices self.report_api = api @@ -237,7 +241,7 @@ def __init__(self, executable, ir_cache_dirs, devices, api=None, report_referenc self.detect_report_metadata() def get_test_info(self): - result = run_test_executable_extract_result([self.executable, "--info"]) + result = run_test_executable_extract_result([str(self.executable), "--info"]) if "error" in result: raise Exception(f"Test executable does not behave correctly: {result}") if "samples" not in result: @@ -281,7 +285,7 @@ def api_push_test_result(self, test: TestCase, result): "metrics": sample.as_dict(), "cpu_family": CPU_FAMILY, "model_size": test.weights_size, - "ext": {"originalSource": test.original_model_path} + "ext": {"originalSource": test.original_model} }) test_report.append(sample_report) response = attempt(self.api, "v2/memory/push-2-db-facade", {"data": test_report}) @@ -327,7 +331,10 @@ def scan_directory(self, directory: Path): return found_models.update(new_files) new_files = sorted(new_files) - yield from ((path.removeprefix(cache_dir).replace("\\", "/"), path) for path in new_files) + yield from ( + (path.removeprefix(cache_dir).replace("\\", "/"), Path(path)) + for path in new_files + ) def get_description(self, model_path: Path) -> dict[str, str] | None: description_path = model_path.parent / "description.txt" @@ -351,27 +358,24 @@ def get_description(self, model_path: Path) -> dict[str, str] | None: def generate_test_cases(self): for ir_cache_dir in self.ir_cache_dirs: for (model_id, model_path) in self.scan_directory(ir_cache_dir): - model_path = Path(model_path) weights_path = model_path.with_suffix(".bin") if not weights_path.is_file(): print(f"Warning: Test case {model_id} invalid: can't find weights file at {weights_path}") continue weights_size = weights_path.stat().st_size - model_description = self.get_description(model_path) or {} - original_model_path = model_description.get("src_model_path", "") for device in self.devices: yield TestCase( model_id=model_id, - model_path=str(model_path), + model_path=model_path, device=device, - original_model_path=original_model_path, + description=self.get_description(model_path) or {}, weights_size=weights_size, ) def run_test_case(self, test_case: TestCase): try: return run_test_executable_extract_result([ - self.executable, test_case.model_path, test_case.device]) + str(self.executable), test_case.model_path, test_case.device]) except Exception as ex: print(f" When running test an unexpected error happened: {ex}") return {"error": "unexpected error", "exception": ex} @@ -418,7 +422,7 @@ def run(self): if __name__ == "__main__": parser = argparse.ArgumentParser() - parser.add_argument("test_executable") + parser.add_argument("test_executable", type=Path) parser.add_argument("--ir-cache", "--ir_cache", type=Path, nargs='*', required=True, help='Path to directory with *.xml model files; ' 'can be a wildcard expression, among all directories '