overhaul broken tests system, now uses yaml, supports regex

Author: strtgbb

ci/defs/job_configs.py

@@ -39,7 +39,7 @@
             "./tests/config",
             "./tests/*.txt",
             "./ci/docker/stateless-test",
-            "./tests/broken_tests.json",
+            "./tests/broken_tests.yaml",
         ],
     ),
     result_name_for_cidb="Tests",
@@ -686,7 +686,7 @@ class JobConfigs:
                 "./ci/jobs/scripts/integration_tests_runner.py",
                 "./tests/integration/",
                 "./ci/docker/integration",
-                "./tests/broken_tests.json",
+                "./tests/broken_tests.yaml",
             ],
         ),
     ).parametrize(
@@ -710,7 +710,7 @@ class JobConfigs:
                 "./ci/jobs/scripts/integration_tests_runner.py",
                 "./tests/integration/",
                 "./ci/docker/integration",
-                "./tests/broken_tests.json",
+                "./tests/broken_tests.yaml",
             ],
         ),
     ).parametrize(
@@ -752,7 +752,7 @@ class JobConfigs:
                 "./ci/jobs/scripts/integration_tests_runner.py",
                 "./tests/integration/",
                 "./ci/docker/integration",
-                "./tests/broken_tests.json",
+                "./tests/broken_tests.yaml",
             ],
         ),
         allow_merge_on_failure=True,
@@ -777,7 +777,7 @@ class JobConfigs:
                 "./ci/jobs/scripts/integration_tests_runner.py",
                 "./tests/integration/",
                 "./ci/docker/integration",
-                "./tests/broken_tests.json",
+                "./tests/broken_tests.yaml",
             ],
         ),
         requires=[ArtifactNames.CH_AMD_ASAN],

ci/jobs/scripts/functional_tests_results.py

@@ -3,6 +3,9 @@
 import os
 import traceback
 from typing import List
+import re
+
+import yaml
 
 from praktika.result import Result
 
@@ -30,14 +33,81 @@
 #         out.writerow(status)
 
 
-def get_broken_tests_list() -> dict:
-    file_path = "tests/broken_tests.json"
-    if not os.path.isfile(file_path) or os.path.getsize(file_path) == 0:
-        return {}
+def get_broken_tests_rules() -> dict:
+    broken_tests_file_path = "tests/broken_tests.yaml"
+    if (
+        not os.path.isfile(broken_tests_file_path)
+        or os.path.getsize(broken_tests_file_path) == 0
+    ):
+        raise ValueError(
+            "There is something wrong with getting broken tests rules: "
+            f"file '{broken_tests_file_path}' is empty or does not exist."
+        )
+
+    with open(broken_tests_file_path, "r", encoding="utf-8") as broken_tests_file:
+        broken_tests = yaml.safe_load(broken_tests_file)
+
+    compiled_rules = {"exact": {}, "pattern": {}}
+
+    for test in broken_tests:
+        regex = test.get("regex") is True
+        rule = {
+            "reason": test["reason"],
+            "message": re.compile(test["message"]) if regex else test["message"],
+        }
+
+        if test.get("not_message"):
+            rule["not_message"] = (
+                re.compile(test["not_message"]) if regex else test["not_message"]
+            )
+        if test.get("check_types"):
+            rule["check_types"] = test["check_types"]
+
+        if regex:
+            rule["regex"] = True
+            compiled_rules["pattern"][re.compile(test["name"])] = rule
+        else:
+            compiled_rules["exact"][test["name"]] = rule
+
+    return compiled_rules
+
+
+def test_is_known_fail(test_name, test_logs, known_broken_tests, test_options_string):
+    matching_rules = []
+
+    exact_rule = known_broken_tests["exact"].get(test_name)
+    if exact_rule:
+        matching_rules.append(exact_rule)
+
+    for name_re, data in known_broken_tests["pattern"].items():
+        if name_re.fullmatch(test_name):
+            matching_rules.append(data)
+
+    if not matching_rules:
+        return False
 
-    with open(file_path, "r", encoding="utf-8") as skip_list_file:
-        skip_list_tests = json.load(skip_list_file)
-    return skip_list_tests
+    def matches_field(field, log, is_regex):
+        if field is None:
+            return True
+        if is_regex:
+            return bool(field.search(log))
+        return field in log
+
+    for rule_data in matching_rules:
+        if rule_data.get("check_types") and not any(
+            ct in test_options_string for ct in rule_data["check_types"]
+        ):
+            continue  # check_types didn't match → skip rule
+
+        is_regex = rule_data.get("regex", False)
+        if matches_field(rule_data.get("not_message"), test_logs, is_regex):
+            continue  # not_message matched → skip rule
+        if not matches_field(rule_data.get("message"), test_logs, is_regex):
+            continue  # message didn't match → skip rule
+
+        return rule_data["reason"]
+
+    return False
 
 
 class FTResultsProcessor:
@@ -75,7 +145,7 @@ def _process_test_output(self):
         test_results = []
         test_end = True
 
-        known_broken_tests = get_broken_tests_list()
+        known_broken_tests = get_broken_tests_rules()
 
         with open(self.tests_output_file, "r", encoding="utf-8") as test_file:
             for line in test_file:
@@ -161,34 +231,16 @@ def _process_test_output(self):
                 )
 
                 if test[1] == "FAIL":
-                    broken_message = None
-                    if test[0] in known_broken_tests.keys():
-                        message = known_broken_tests[test[0]].get("message")
-                        check_types = known_broken_tests[test[0]].get("check_types")
-                        if check_types and not any(
-                            check_type in test_options_string
-                            for check_type in check_types
-                        ):
-                            broken_message = None
-                        elif message:
-                            if message in test_results_[-1].info:
-                                broken_message = (
-                                    f"\nMarked as broken, matched message: '{message}'"
-                                )
-                        else:
-                            broken_message = f"\nMarked as broken, no message specified"
-
-                        if broken_message and check_types:
-                            broken_message += (
-                                f", matched one or more check types {check_types}"
-                            )
+                    broken_message = test_is_known_fail(
+                        test[0], test[3], known_broken_tests, test_options_string
+                    )
 
                     if broken_message:
                         broken += 1
                         failed -= 1
                         test_results_[-1].set_status(Result.StatusExtended.BROKEN)
                         test_results_[-1].set_label(Result.Label.BROKEN)
-                        test_results_[-1].info += broken_message
+                        test_results_[-1].info += "\nMarked as broken: " + broken_message
 
             except Exception as e:
                 print(f"ERROR: Failed to parse test results: {test}")

ci/jobs/scripts/integration_tests_runner.py

@@ -406,17 +406,43 @@ def _get_parallel_tests_skip_list(repo_path):
         return list(sorted(skip_list_tests))
 
     @staticmethod
-    def _get_broken_tests_list(repo_path: str) -> dict:
-        skip_list_file_path = f"{repo_path}/tests/broken_tests.json"
+    def _get_broken_tests_rules(repo_path: str) -> dict:
+        broken_tests_file_path = f"{repo_path}/tests/broken_tests.yaml"
         if (
-            not os.path.isfile(skip_list_file_path)
-            or os.path.getsize(skip_list_file_path) == 0
+            not os.path.isfile(broken_tests_file_path)
+            or os.path.getsize(broken_tests_file_path) == 0
         ):
-            return {}
+            raise ValueError(
+                "There is something wrong with getting broken tests rules: "
+                f"file '{broken_tests_file_path}' is empty or does not exist."
+            )
 
-        with open(skip_list_file_path, "r", encoding="utf-8") as skip_list_file:
-            skip_list_tests = json.load(skip_list_file)
-        return skip_list_tests
+        with open(broken_tests_file_path, "r", encoding="utf-8") as broken_tests_file:
+            broken_tests = yaml.safe_load(broken_tests_file)
+
+        compiled_rules = {"exact": {}, "pattern": {}}
+
+        for test in broken_tests:
+            regex = test.get("regex") is True
+            rule = {
+                "reason": test["reason"],
+                "message": re.compile(test["message"]) if regex else test["message"],
+            }
+
+            if test.get("not_message"):
+                rule["not_message"] = (
+                    re.compile(test["not_message"]) if regex else test["not_message"]
+                )
+            if test.get("check_types"):
+                rule["check_types"] = test["check_types"]
+
+            if regex:
+                rule["regex"] = True
+                compiled_rules["pattern"][re.compile(test["name"])] = rule
+            else:
+                compiled_rules["exact"][test["name"]] = rule
+
+        return compiled_rules
 
     @staticmethod
     def group_test_by_file(tests):
@@ -469,6 +495,43 @@ def get_log_paths(test_name):
 
         broken_tests_log = os.path.join(self.result_path, "broken_tests_handler.log")
 
+        def test_is_known_fail(test_name, test_logs):
+            matching_rules = []
+
+            exact_rule = known_broken_tests["exact"].get(test_name)
+            if exact_rule:
+                matching_rules.append(exact_rule)
+
+            for name_re, data in known_broken_tests["pattern"].items():
+                if name_re.fullmatch(test_name):
+                    matching_rules.append(data)
+
+            if not matching_rules:
+                return False
+
+            def matches_field(field, log, is_regex):
+                if field is None:
+                    return True
+                if is_regex:
+                    return bool(field.search(log))
+                return field in log
+
+            for rule_data in matching_rules:
+                if rule_data.get("check_types") and not any(
+                    ct in context_name for ct in rule_data["check_types"]
+                ):
+                    continue  # check_types didn't match → skip rule
+
+                is_regex = rule_data.get("regex", False)
+                if matches_field(rule_data.get("not_message"), test_logs, is_regex):
+                    continue  # not_message matched → skip rule
+                if not matches_field(rule_data.get("message"), test_logs, is_regex):
+                    continue  # message didn't match → skip rule
+
+                return rule_data["reason"]
+
+            return False
+
         with open(broken_tests_log, "a") as log_file:
             log_file.write(f"{len(known_broken_tests)} Known broken tests\n")
             for status, tests in counters.items():
@@ -479,45 +542,21 @@ def get_log_paths(test_name):
                     log_file.write(
                         f"Checking test {failed_test} (status: {fail_status})\n"
                     )
-                    if failed_test not in known_broken_tests.keys():
-                        log_file.write(
-                            f"Test {failed_test} is not in known broken tests\n"
-                        )
-                        continue
-
-                    check_types = known_broken_tests[failed_test].get("check_types")
-                    fail_message = known_broken_tests[failed_test].get("message")
 
-                    if check_types and not any(
-                        check_type in context_name for check_type in check_types
-                    ):
+                    # Should only care about the most recent log file
+                    log_path = get_log_paths(failed_test)[0]
+                    test_log = extract_fail_logs(log_path).get(
+                        failed_test.split("::")[-1]
+                    )
+                    known_fail_reason = test_is_known_fail(failed_test, test_log)
+                    if known_fail_reason is not False:
                         log_file.write(
-                            f"Test {context_name} {failed_test} is only known to be broken for check types {check_types}\n"
+                            f"Test {failed_test} is known to fail: {known_fail_reason}\n"
                         )
-                        mark_as_broken = False
-                    elif not fail_message:
-                        log_file.write("No fail message specified, marking as broken\n")
-                        mark_as_broken = True
-                    else:
-                        log_file.write(f"Looking for fail message: {fail_message}\n")
-                        mark_as_broken = False
-                        for log_path in get_log_paths(failed_test):
-                            if log_path.endswith(".log"):
-                                log_file.write(f"Checking log file: {log_path}\n")
-                                fail_logs = extract_fail_logs(log_path).get(
-                                    failed_test.split("::")[-1]
-                                )
-                                if fail_logs and fail_message in fail_logs:
-                                    log_file.write("Found fail message in logs\n")
-                                    mark_as_broken = True
-                                    break
-
-                    if mark_as_broken:
-                        log_file.write(f"Moving test to BROKEN state\n")
                         counters[fail_status].remove(failed_test)
                         counters["BROKEN"].append(failed_test)
                     else:
-                        log_file.write("Test not marked as broken\n")
+                        log_file.write(f"Test {failed_test} is not known to fail\n")
 
             for status, tests in counters.items():
                 log_file.write(f"Total tests in {status} state: {len(tests)}\n")
@@ -796,7 +835,7 @@ def run_flaky_check(self, should_fail=False):
         }  # type: Dict
         tests_times = defaultdict(float)  # type: Dict
         tests_log_paths = defaultdict(list)
-        known_broken_tests = self._get_broken_tests_list(self.repo_path)
+        known_broken_tests = self._get_broken_tests_rules(self.repo_path)
         id_counter = 0
         for test_to_run in tests_to_run:
             tries_num = 1 if should_fail else FLAKY_TRIES_COUNT
@@ -1234,7 +1273,7 @@ def run_normal_check(self):
         tests_times = defaultdict(float)
         tests_log_paths = defaultdict(list)
         items_to_run = list(grouped_tests.items())
-        known_broken_tests = self._get_broken_tests_list(self.repo_path)
+        known_broken_tests = self._get_broken_tests_rules(self.repo_path)
         logging.info("Total test groups %s", len(items_to_run))
         if self.shuffle_test_groups():
             logging.info("Shuffling test groups")