[CONFORMANCE] Extend conformance runner to use in GA with expected_failures filelist (#19285)

* [CONFORMANCE] Extend conformance runner to use in GA with expected_failures filelist

* fix

* exclude failed tests from run in case without update

* Small refactoring

src/tests/test_utils/functional_test_utils/layer_tests_summary/merge_xmls.py

@@ -63,7 +63,7 @@ def aggregate_test_results(aggregated_results: SubElement, xml_reports: list,
         try:
             xml_root = ET.parse(xml).getroot()
         except ET.ParseError:
-            logger.error(f' {xml} is corrupted and skipped')
+            # logger.error(f' {xml} is corrupted and skipped')
             continue
         xml_results = xml_root.find("results")
         xml_timestamp = xml_root.get("timestamp")

src/tests/test_utils/functional_test_utils/layer_tests_summary/run_conformance.py

@@ -13,6 +13,7 @@ import defusedxml.ElementTree as ET
 from urllib.parse import urlparse
 
 import os
+import csv
 import urllib.request as ur
 from utils import constants
 from utils.conformance_utils import get_logger
@@ -40,61 +41,88 @@ def get_default_working_dir():
 def parse_arguments():
     parser = ArgumentParser()
 
-    models_path_help = "Path to the directory/ies containing models to dump subgraph (the default way is to download conformance IR). It may be directory, archieve file, .lst file with model to download or http link to download something . If --s=0, specify the Conformance IRs directory"
-    device_help = " Specify the target device. The default value is CPU"
-    ov_help = "OV repo path. The default way is try to find the absolute path of OV repo (by using script path)"
-    working_dir_help = "Specify a working directory to save all artifacts, such as reports, models, conformance_irs, etc."
+    models_path_help = "Path to the directory/ies containing models to dump subgraph (the default way is to download conformance IR). It may be directory, archieve file, .lst file with models to download by a link, model file paths. If --s=0, specify the Conformance IRs directory. NOTE: Applicable only for Opset Conformance."
+    dump_graph_help = "Set '1' to create Conformance IRs from models using subgraphsDumper tool. The default value is '0'. NOTE: Applicable only for Opset Conformance."
+    device_help = "Specify a target device. The default value is `CPU`"
+    ov_help = "OV binary path. The default way is to find the absolute path of latest bin in the repo (by using script path)"
+    working_dir_help = "Specify a working directory to save a run artifacts"
     type_help = "Specify conformance type: `OP` or `API`. The default value is `OP`"
-    workers_help = "Specify number of workers to run in parallel. The default value is CPU count - 1"
-    gtest_filter_helper = "Specify gtest filter to apply when running test. E.g. *Add*:*BinaryConv*. The default value is None"
-    ov_config_path_helper = "Specify path to file contains plugin config"
-    dump_conformance_help = "Set '1' if you want to create Conformance IRs from custom/downloaded models. In other cases, set 0. The default value is '1'"
-    shape_mode_help = "Specify shape mode for conformance. Default value is ``. Possible values: `static`, `dynamic`, ``"
-    parallel_help = "Parallel over HW devices. For example run tests over GPU.0, GPU.1 and etc"
+    workers_help = "Specify number of workers to run in parallel. The default value is `CPU_count-1`"
+    gtest_filter_helper = "Specify gtest filter to apply for a test run. E.g. *Add*:*BinaryConv*. The default value is None"
+    ov_config_path_helper = "Specify path to a plugin config file as `.lst` file. Default value is ``"
+    special_mode_help = "Specify shape mode (`static`, `dynamic` or ``) for Opset conformance or API scope type (`mandatory` or ``). Default value is ``"
+    parallel_help = "Parallel over HW devices. For example run tests over GPU.0 and GPU.1 in case when device are the same"
+    expected_failures_help = "Excepted failures list file path as csv"
+    cache_path_help = "Path to the cache file with test_name list sorted by execution time as `.lst` file!"
+    expected_failures_update_help = "Overwrite expected failures list in case same failures were fixed"
 
-    parser.add_argument("-m", "--models_path", help=models_path_help, type=str, required=False, default=NO_MODEL_CONSTANT)
     parser.add_argument("-d", "--device", help= device_help, type=str, required=False, default="CPU")
-    parser.add_argument("-ov", "--ov_path", help=ov_help, type=str, required=False, default=file_utils.get_ov_path(SCRIPT_DIR_PATH))
-    parser.add_argument("-w", "--working_dir", help=working_dir_help, type=str, required=False, default=get_default_working_dir())
     parser.add_argument("-t", "--type", help=type_help, type=str, required=False, default=constants.OP_CONFORMANCE)
-    parser.add_argument("-j", "--workers", help=workers_help, type=int, required=False, default=os.cpu_count()-1)
     parser.add_argument("--gtest_filter", help=gtest_filter_helper, type=str, required=False, default="*")
+    parser.add_argument("-w", "--working_dir", help=working_dir_help, type=str, required=False, default=get_default_working_dir())
+    parser.add_argument("-m", "--models_path", help=models_path_help, type=str, required=False, default=NO_MODEL_CONSTANT)
+    parser.add_argument("-ov", "--ov_path", help=ov_help, type=str, required=False, default="")
+    parser.add_argument("-j", "--workers", help=workers_help, type=int, required=False, default=os.cpu_count()-1)
     parser.add_argument("-c", "--ov_config_path", help=ov_config_path_helper, type=str, required=False, default="")
-    parser.add_argument("-s", "--dump_conformance", help=dump_conformance_help, type=int, required=False, default=0)
-    parser.add_argument("-sm", "--shape_mode", help=shape_mode_help, type=str, required=False, default="")
+    parser.add_argument("-s", "--dump_graph", help=dump_graph_help, type=int, required=False, default=0)
+    parser.add_argument("-sm", "--special_mode", help=special_mode_help, type=str, required=False, default="")
     parser.add_argument("-p", "--parallel_devices", help=parallel_help, type=bool, required=False, default=False)
+    parser.add_argument("-f", "--expected_failures", help=expected_failures_help, type=str, required=False, default="")
+    parser.add_argument("-u", "--expected_failures_update", help=expected_failures_update_help, type=bool, required=False, default=False)
+    parser.add_argument("--cache_path", help=cache_path_help, type=str, required=False, default="")
 
     return parser.parse_args()
 
 class Conformance:
     def __init__(self, device:str, model_path:os.path, ov_path:os.path, type:str, workers:int,
-                 gtest_filter:str, working_dir:os.path, ov_config_path:os.path, shape_mode:str,
-                 parallel_devices:bool):
+                 gtest_filter:str, working_dir:os.path, ov_config_path:os.path, special_mode:str,
+                 cache_path:str, parallel_devices:bool, expected_failures_file: str,
+                 expected_failures_update: bool):
         self._device = device
         self._model_path = model_path
-        self._ov_path = ov_path
-        self._ov_bin_path = file_utils.get_ov_path(SCRIPT_DIR_PATH, self._ov_path, True)
+        if os.path.isdir(ov_path):
+            self._ov_path = ov_path
+        else:
+            self._ov_path = file_utils.get_ov_path(SCRIPT_DIR_PATH, None, True)
         self._working_dir = working_dir
         if os.path.exists(self._working_dir):
             logger.info(f"Working dir {self._working_dir} is cleaned up")
             rmtree(self._working_dir)
         os.mkdir(self._working_dir)
-        if not (type == constants.OP_CONFORMANCE or type == constants.API_CONFORMANCE):
-            logger.error(f"Incorrect conformance type: {type}. Please use 'OP' or 'API'")
+        self._cache_path = cache_path if os.path.isfile(cache_path) else ""
+        if type == constants.OP_CONFORMANCE:
+            if special_mode == "static" or special_mode == "dynamic" or special_mode == "":
+                self._special_mode = special_mode
+            else:
+                logger.error(f'Incorrect value to set shape mode: {special_mode}. Please check to get possible values')
+                exit(-1)
+            self._gtest_filter = gtest_filter
+        elif type == constants.API_CONFORMANCE:
+            self._special_mode = ""
+            if special_mode == "mandatory":
+                self._gtest_filter = f"*mandatory*{gtest_filter}*:*{gtest_filter}*mandatory*"
+            elif special_mode == "":
+                self._gtest_filter = gtest_filter
+            else:
+                logger.error(f'Incorrect value to set API scope: {special_mode}. Please check to get possible values')
+                exit(-1)
+        else:
+            logger.error(f"Incorrect conformance type: {type}. Please use '{constants.OP_CONFORMANCE}' or '{constants.API_CONFORMANCE}'")
             exit(-1)
         self._type = type
         self._workers = workers
-        self._gtest_filter = gtest_filter
         if not os.path.exists(ov_config_path) and ov_config_path != "":
             logger.error(f"Specified config file does not exist: {ov_config_path}.")
             exit(-1)
         self._ov_config_path = ov_config_path
-        if shape_mode == "static" or shape_mode == "dynamic" or shape_mode == "":
-            self._shape_mode = shape_mode
-        else:
-            logger.error(f'Incorrect value to set shape mode: {shape_mode}. Please check to get possible values')
-            exit(-1)
         self._is_parallel_over_devices = parallel_devices
+        self._expected_failures = set()
+        self._expected_failures_file = expected_failures_file
+        if os.path.isfile(expected_failures_file):
+            self._expected_failures = self.__get_failed_test_from_csv(expected_failures_file)
+        else:
+            logger.warning(f"Expected failures testlist `{self._expected_failures_file}` does not exist!")
+        self._expected_failures_update = expected_failures_update
 
     def __download_models(self, url_to_download, path_to_save):
         _, file_name = os.path.split(urlparse(url_to_download).path)
@@ -116,7 +144,7 @@ class Conformance:
         
 
     def __dump_subgraph(self):
-        subgraph_dumper_path = os.path.join(self._ov_bin_path, f'{SUBGRAPH_DUMPER_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
+        subgraph_dumper_path = os.path.join(self._ov_path, f'{SUBGRAPH_DUMPER_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
         if not os.path.isfile(subgraph_dumper_path):
             logger.error(f"{subgraph_dumper_path} is not exist!")
             exit(-1)
@@ -142,16 +170,46 @@ class Conformance:
         if has_python_api:
             op_rel_weight = create_hash(Path(self._model_path))
             save_rel_weights(Path(self._model_path), op_rel_weight)
-            logger.info(f"All conformance IRs in {self._ov_bin_path} were renamed based on hash")
+            logger.info(f"All conformance IRs in {self._model_path} were renamed based on hash")
         else:
             logger.warning("The OV Python was not built or Environment was not updated to requirments. Skip the step to rename Conformance IR based on a hash")
 
+
+    @staticmethod
+    def __get_failed_test_from_csv(csv_file:str):
+        failures = set()
+        with open(csv_file, "r") as failures_file:
+            for row in csv.reader(failures_file, delimiter=','):
+                if row[0] == "Test Name":
+                    continue
+                failures.add(row[0])
+            failures_file.close()
+        return failures
+
+    def __check_expected_failures(self):
+        this_failures_file = os.path.join(self._working_dir, f"{self._device}_logs", "logs", "fix_priority.csv")
+        if not os.path.isfile(this_failures_file):
+            return
+        this_run_failures = self.__get_failed_test_from_csv(this_failures_file)
+
+        diff = this_run_failures.difference(self._expected_failures)
+        if len(diff) > 0:
+            logger.error(f"Unexpected failures: {diff}")
+            exit(-1)
+        
+        intersection = self._expected_failures.intersection(this_run_failures)
+        if this_run_failures != self._expected_failures and self._expected_failures_update:
+            logger.info(f"Expected failures file {self._expected_failures} will be updated!!!")
+            os.remove(self._expected_failures_file)
+            this_failures_file = Path(this_failures_file)
+            this_failures_file.rename(self._expected_failures_file)
+
     def __run_conformance(self):
         conformance_path = None
         if self._type == constants.OP_CONFORMANCE:
-            conformance_path = os.path.join(self._ov_bin_path, f'{OP_CONFORMANCE_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
+            conformance_path = os.path.join(self._ov_path, f'{OP_CONFORMANCE_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
         else:
-            conformance_path = os.path.join(self._ov_bin_path, f'{API_CONFORMANCE_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
+            conformance_path = os.path.join(self._ov_path, f'{API_CONFORMANCE_BIN_NAME}{constants.OS_BIN_FILE_EXT}')
 
         if not os.path.isfile(conformance_path):
             logger.error(f"{conformance_path} is not exist!")
@@ -168,14 +226,24 @@ class Conformance:
         if not os.path.isdir(logs_dir):
             os.mkdir(logs_dir)
         
-        command_line_args = [f"--device={self._device}", f'--input_folders="{self._model_path}"',
+        command_line_args = [f"--device={self._device}", 
+                             f'--input_folders="{self._model_path}"' if self._type == constants.OP_CONFORMANCE else '',
                              f"--report_unique_name", f'--output_folder="{parallel_report_dir}"',
-                             f'--gtest_filter={self._gtest_filter}', f'--config_path="{self._ov_config_path}"',
-                             f'--shape_mode={self._shape_mode}']
-        conformance = TestParallelRunner(f"{conformance_path}", command_line_args, self._workers, logs_dir, "", self._is_parallel_over_devices)
+                             f'--gtest_filter=\"{self._gtest_filter}\"', f'--config_path="{self._ov_config_path}"',
+                             f'--shape_mode={self._special_mode}']
+        conformance = TestParallelRunner(f"{conformance_path}",
+                                         command_line_args,
+                                         self._workers,
+                                         logs_dir,
+                                         self._cache_path,
+                                         self._is_parallel_over_devices,
+                                         self._expected_failures if not self._expected_failures_update else set())
         conformance.run()
         conformance.postprocess_logs()
 
+        if os.path.isfile(self._expected_failures_file):
+            self.__check_expected_failures()
+
         final_report_name = f'report_{self._type.lower()}'
         merge_xml([parallel_report_dir], report_dir, final_report_name, self._type, True)
 
@@ -185,7 +253,7 @@ class Conformance:
     def __summarize(self, xml_report_path:os.path, report_dir: os.path):
         if self._type == constants.OP_CONFORMANCE:
             summary_root = ET.parse(xml_report_path).getroot()
-            rel_weights_path = os.path.join(self._model_path, constants.REL_WEIGHTS_FILENAME.replace(constants.REL_WEIGHTS_REPLACE_STR, self._shape_mode))
+            rel_weights_path = os.path.join(self._model_path, constants.REL_WEIGHTS_FILENAME.replace(constants.REL_WEIGHTS_REPLACE_STR, self._special_mode))
             create_summary(summary_root, report_dir, [], "", "", True, True, rel_weights_path)
         else:
             create_api_summary([xml_report_path], report_dir, [], "", "")
@@ -205,35 +273,39 @@ class Conformance:
             logger.error("Impossible to install requirements!")
             exit(-1)
         logger.info(f"[ARGUMENTS] --device = {self._device}")
-        logger.info(f"[ARGUMENTS] --ov_path = {self._ov_path}")
-        logger.info(f"[ARGUMENTS] --models_path = {self._model_path}")
-        logger.info(f"[ARGUMENTS] --working_dir = {self._working_dir}")
         logger.info(f"[ARGUMENTS] --type = {self._type}")
+        logger.info(f"[ARGUMENTS] --working_dir = {self._working_dir}")
+        logger.info(f"[ARGUMENTS] --ov_path = {self._ov_path}")
+        logger.info(f"[ARGUMENTS] --ov_config_path = {self._ov_config_path}")
         logger.info(f"[ARGUMENTS] --workers = {self._workers}")
         logger.info(f"[ARGUMENTS] --gtest_filter = {self._gtest_filter}")
-        logger.info(f"[ARGUMENTS] --ov_config_path = {self._ov_config_path}")
-        logger.info(f"[ARGUMENTS] --dump_conformance = {dump_models}")
-        logger.info(f"[ARGUMENTS] --shape_mode = {self._shape_mode}")
+        logger.info(f"[ARGUMENTS] --models_path = {self._model_path}")
+        logger.info(f"[ARGUMENTS] --dump_graph = {dump_models}")
+        logger.info(f"[ARGUMENTS] --shape_mode = {self._special_mode}")
         logger.info(f"[ARGUMENTS] --parallel_devices = {self._is_parallel_over_devices}")
+        logger.info(f"[ARGUMENTS] --cache_path = {self._cache_path}")
+        logger.info(f"[ARGUMENTS] --expected_failures = {self._expected_failures_file}")
+        logger.info(f"[ARGUMENTS] --expected_failures_update = {self._expected_failures_update}")
 
-        if file_utils.is_url(self._model_path):
-            self._model_path = self.__download_models(self._model_path, self._working_dir)
-        if self._model_path == NO_MODEL_CONSTANT or os.path.splitext(self._model_path)[1] == ".lst":
-            with open(self._model_path, "r") as model_list_file:
-                model_dir = os.path.join(self._working_dir, "models")
-                if not os.path.isdir(model_dir):
-                    os.mkdir(model_dir)
-                for model in model_list_file.readlines():
-                    self.__download_models(model, model_dir)
-                self._model_path = model_dir
-        if dump_models:
-            self.__dump_subgraph()
-        if not os.path.exists(self._model_path):
-            logger.error(f"The model direstory {self._model_path} does not exist!")
-            exit(-1)
-        if not os.path.exists(self._model_path):
-            logger.error(f"Directory {self._model_path} does not exist")
-            exit(-1)
+        if self._type == constants.OP_CONFORMANCE:
+            if file_utils.is_url(self._model_path):
+                self._model_path = self.__download_models(self._model_path, self._working_dir)
+            if self._model_path == NO_MODEL_CONSTANT or os.path.splitext(self._model_path)[1] == ".lst":
+                with open(self._model_path, "r") as model_list_file:
+                    model_dir = os.path.join(self._working_dir, "models")
+                    if not os.path.isdir(model_dir):
+                        os.mkdir(model_dir)
+                    for model in model_list_file.readlines():
+                        self.__download_models(model, model_dir)
+                    self._model_path = model_dir
+            if dump_models:
+                self.__dump_subgraph()
+            if not os.path.exists(self._model_path):
+                logger.error(f"The model direstory {self._model_path} does not exist!")
+                exit(-1)
+            if not os.path.exists(self._model_path):
+                logger.error(f"Directory {self._model_path} does not exist")
+                exit(-1)
         xml_report, report_dir = self.__run_conformance()
         self.__summarize(xml_report, report_dir)
 
@@ -243,5 +315,8 @@ if __name__ == "__main__":
                               args.ov_path, args.type,
                               args.workers, args.gtest_filter,
                               args.working_dir, args.ov_config_path,
-                              args.shape_mode, args.parallel_devices)
-    conformance.run(args.dump_conformance)
+                              args.special_mode, args.cache_path,
+                              args.parallel_devices, args.expected_failures,
+                              args.expected_failures_update)
+    conformance.run(args.dump_graph)
+    

src/tests/test_utils/functional_test_utils/layer_tests_summary/run_parallel.py

@@ -204,7 +204,9 @@ class TaskManager:
         return self._idx
 
 class TestParallelRunner:
-    def __init__(self, exec_file_path: os.path, test_command_line: list, worker_num: int, working_dir: os.path, cache_path: os.path, is_parallel_devices=False):
+    def __init__(self, exec_file_path: os.path, test_command_line: list,
+                 worker_num: int, working_dir: os.path, cache_path: os.path,
+                 is_parallel_devices=False, excluded_tests=set()):
         self._exec_file_path = exec_file_path
         self._working_dir = working_dir
         self._conformance_ir_filelists = list()
@@ -225,6 +227,10 @@ class TestParallelRunner:
         self._available_devices = [self._device] if not self._device is None else []
         if has_python_api and is_parallel_devices:
             self._available_devices = get_available_devices(self._device)
+        self._excluded_tests_re = set()
+        self._orig_excluded_tests = excluded_tests
+        for test in excluded_tests:
+            self._excluded_tests_re.add(f'"{self.__replace_restricted_symbols(test)}":')
             
 
     def __init_basic_command_line_for_exec_file(self, test_command_line: list):
@@ -320,11 +326,11 @@ class TestParallelRunner:
         cached_test_list_names = list()
 
         for test in test_list_cache:
-            if test._name in test_list_runtime:
+            if test._name in test_list_runtime and not test in self._excluded_tests_re:
                 cached_test_list.append(test)
                 cached_test_list_names.append(test._name)
         for test in test_list_runtime:
-            if not test in cached_test_list_names:
+            if not test in cached_test_list_names and not test in self._excluded_tests_re:
                 runtime_test_test.append(test)
 
         if len(runtime_test_test) > 0:
@@ -469,7 +475,9 @@ class TestParallelRunner:
                     interapted_tests.append(f'"{test_name}":')
                 log_file.close()
         test_list_runtime = set(self.__get_test_list_by_runtime())
-        return list(test_list_runtime.difference(test_names)), interapted_tests
+        not_runned_tests = test_list_runtime.difference(test_names).difference(self._excluded_tests_re)
+        interapted_tests = set(interapted_tests).difference(self._excluded_tests_re)
+        return list(not_runned_tests), list(interapted_tests)
 
     def run(self):
         if TaskManager.process_timeout == -1:
@@ -723,7 +731,7 @@ class TestParallelRunner:
         not_run_tests_path = os.path.join(logs_dir, "not_run_tests.log")
         with open(not_run_tests_path, "w") as not_run_tests_path_file:
             test_list_runtime = self.__get_test_list_by_runtime()
-            diff_set = set(test_list_runtime).difference(set(saved_tests))
+            diff_set = set(test_list_runtime).difference(set(saved_tests)).difference(self._orig_excluded_tests)
             diff_list = list()
             for item in diff_set:
                 diff_list.append(f"{item}\n")

src/tests/test_utils/functional_test_utils/layer_tests_summary/utils/file_utils.py

@@ -71,8 +71,7 @@ def unzip_archieve(zip_path: os.path, dst_path: os.path):
 def find_latest_dir(in_dir: Path, pattern_list = list()):
     get_latest_dir = lambda path: sorted(Path(path).iterdir(), key=os.path.getmtime)
     entities = get_latest_dir(in_dir)
-    if not constants.IS_MACOS:
-        entities.reverse()
+    entities.reverse()
 
     for entity in entities:
         if entity.is_dir():