diff --git a/tests/memory_tests/tools/run_tests.py b/tests/memory_tests/tools/run_tests.py index c65902385f6c..572b1dcc1baa 100644 --- a/tests/memory_tests/tools/run_tests.py +++ b/tests/memory_tests/tools/run_tests.py @@ -211,10 +211,25 @@ def modelid_assume_info(modelid): return modelname, framework, precision +@dataclass +class TestCase: + model_id: str + model_path: Path + device: str + + # metadata + 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.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 @@ -226,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: @@ -246,12 +261,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 +275,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": {"originalSource": test.original_model} }) test_report.append(sample_report) response = attempt(self.api, "v2/memory/push-2-db-facade", {"data": test_report}) @@ -315,33 +331,56 @@ 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" + description_text = None + try: + 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): - def _with_filesize(paths): - for (modelid, path) in paths: - weights_path, _ = os.path.splitext(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 - 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): + for (model_id, model_path) in self.scan_directory(ir_cache_dir): + 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 + for device in self.devices: + yield TestCase( + model_id=model_id, + model_path=model_path, + device=device, + 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, model_path, device]) + return run_test_executable_extract_result([ + 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} - 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 +393,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,16 +414,15 @@ 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__": 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 '