1#!/usr/bin/env python3 2 3import argparse 4from collections import defaultdict 5import difflib 6import os 7import re 8from glcollate import Collate 9from termcolor import colored 10from urllib.parse import urlparse 11 12 13def get_canonical_name(job_name): 14 return re.split(r" \d+/\d+", job_name)[0] 15 16 17def get_xfails_file_path(job_name, suffix): 18 canonical_name = get_canonical_name(job_name) 19 name = canonical_name.replace(":", "-") 20 script_dir = os.path.dirname(os.path.abspath(__file__)) 21 return os.path.join(script_dir, f"{name}-{suffix}.txt") 22 23 24def get_unit_test_name_and_results(unit_test): 25 if "Artifact results/failures.csv not found" in unit_test or '' == unit_test: 26 return None, None 27 unit_test_name, unit_test_result = unit_test.strip().split(",") 28 return unit_test_name, unit_test_result 29 30 31def read_file(file_path): 32 try: 33 with open(file_path, "r") as file: 34 f = file.readlines() 35 if len(f): 36 f[-1] = f[-1].strip() + "\n" 37 return f 38 except FileNotFoundError: 39 return [] 40 41 42def save_file(content, file_path): 43 # delete file is content is empty 44 if not content or not any(content): 45 if os.path.exists(file_path): 46 os.remove(file_path) 47 return 48 49 with open(file_path, "w") as file: 50 file.writelines(content) 51 52 53def is_test_present_on_file(file_content, unit_test_name): 54 return any(unit_test_name in line for line in file_content) 55 56 57def is_unit_test_present_in_other_jobs(unit_test, job_ids): 58 return all(unit_test in job_ids[job_id] for job_id in job_ids) 59 60 61def remove_unit_test_if_present(lines, unit_test_name): 62 if not is_test_present_on_file(lines, unit_test_name): 63 return 64 lines[:] = [line for line in lines if unit_test_name not in line] 65 66 67def add_unit_test_if_not_present(lines, unit_test_name, file_name): 68 # core_getversion is mandatory 69 if "core_getversion" in unit_test_name: 70 print("WARNING: core_getversion should pass, not adding it to", os.path.basename(file_name)) 71 elif all(unit_test_name not in line for line in lines): 72 lines.append(unit_test_name + "\n") 73 74 75def update_unit_test_result_in_fails_txt(fails_txt, unit_test): 76 unit_test_name, unit_test_result = get_unit_test_name_and_results(unit_test) 77 for i, line in enumerate(fails_txt): 78 if unit_test_name in line: 79 _, current_result = get_unit_test_name_and_results(line) 80 fails_txt[i] = unit_test + "\n" 81 return 82 83 84def add_unit_test_or_update_result_to_fails_if_present(fails_txt, unit_test, fails_txt_path): 85 unit_test_name, _ = get_unit_test_name_and_results(unit_test) 86 if not is_test_present_on_file(fails_txt, unit_test_name): 87 add_unit_test_if_not_present(fails_txt, unit_test, fails_txt_path) 88 # if it is present but not with the same result 89 elif not is_test_present_on_file(fails_txt, unit_test): 90 update_unit_test_result_in_fails_txt(fails_txt, unit_test) 91 92 93def split_unit_test_from_collate(xfails): 94 for job_name in xfails.keys(): 95 for job_id in xfails[job_name].copy().keys(): 96 if "not found" in xfails[job_name][job_id]: 97 del xfails[job_name][job_id] 98 continue 99 xfails[job_name][job_id] = xfails[job_name][job_id].strip().split("\n") 100 101 102def get_xfails_from_pipeline_url(pipeline_url): 103 parsed_url = urlparse(pipeline_url) 104 path_components = parsed_url.path.strip("/").split("/") 105 106 namespace = path_components[0] 107 project = path_components[1] 108 pipeline_id = path_components[-1] 109 110 print("Collating from:", namespace, project, pipeline_id) 111 xfails = ( 112 Collate(namespace=namespace, project=project) 113 .from_pipeline(pipeline_id) 114 .get_artifact("results/failures.csv") 115 ) 116 117 split_unit_test_from_collate(xfails) 118 return xfails 119 120 121def get_xfails_from_pipeline_urls(pipelines_urls): 122 xfails = defaultdict(dict) 123 124 for url in pipelines_urls: 125 new_xfails = get_xfails_from_pipeline_url(url) 126 for key in new_xfails: 127 xfails[key].update(new_xfails[key]) 128 129 return xfails 130 131 132def print_diff(old_content, new_content, file_name): 133 diff = difflib.unified_diff(old_content, new_content, lineterm="", fromfile=file_name, tofile=file_name) 134 diff = [colored(line, "green") if line.startswith("+") else 135 colored(line, "red") if line.startswith("-") else line for line in diff] 136 print("\n".join(diff[:3])) 137 print("".join(diff[3:])) 138 139 140def main(pipelines_urls, only_flakes): 141 xfails = get_xfails_from_pipeline_urls(pipelines_urls) 142 143 for job_name in xfails.keys(): 144 fails_txt_path = get_xfails_file_path(job_name, "fails") 145 flakes_txt_path = get_xfails_file_path(job_name, "flakes") 146 147 fails_txt = read_file(fails_txt_path) 148 flakes_txt = read_file(flakes_txt_path) 149 150 fails_txt_original = fails_txt.copy() 151 flakes_txt_original = flakes_txt.copy() 152 153 for job_id in xfails[job_name].keys(): 154 for unit_test in xfails[job_name][job_id]: 155 unit_test_name, unit_test_result = get_unit_test_name_and_results(unit_test) 156 157 if not unit_test_name: 158 continue 159 160 if only_flakes: 161 remove_unit_test_if_present(fails_txt, unit_test_name) 162 add_unit_test_if_not_present(flakes_txt, unit_test_name, flakes_txt_path) 163 continue 164 165 # drop it from flakes if it is present to analyze it again 166 remove_unit_test_if_present(flakes_txt, unit_test_name) 167 168 if unit_test_result == "UnexpectedPass": 169 remove_unit_test_if_present(fails_txt, unit_test_name) 170 # flake result 171 if not is_unit_test_present_in_other_jobs(unit_test, xfails[job_name]): 172 add_unit_test_if_not_present(flakes_txt, unit_test_name, flakes_txt_path) 173 continue 174 175 # flake result 176 if not is_unit_test_present_in_other_jobs(unit_test, xfails[job_name]): 177 remove_unit_test_if_present(fails_txt, unit_test_name) 178 add_unit_test_if_not_present(flakes_txt, unit_test_name, flakes_txt_path) 179 continue 180 181 # consistent result 182 add_unit_test_or_update_result_to_fails_if_present(fails_txt, unit_test, 183 fails_txt_path) 184 185 fails_txt.sort() 186 flakes_txt.sort() 187 188 if fails_txt != fails_txt_original: 189 save_file(fails_txt, fails_txt_path) 190 print_diff(fails_txt_original, fails_txt, os.path.basename(fails_txt_path)) 191 if flakes_txt != flakes_txt_original: 192 save_file(flakes_txt, flakes_txt_path) 193 print_diff(flakes_txt_original, flakes_txt, os.path.basename(flakes_txt_path)) 194 195 196if __name__ == "__main__": 197 parser = argparse.ArgumentParser(description="Update xfails from a given pipeline.") 198 parser.add_argument("pipeline_urls", nargs="+", type=str, help="URLs to the pipelines to analyze the failures.") 199 parser.add_argument("--only-flakes", action="store_true", help="Treat every detected failure as a flake, edit *-flakes.txt only.") 200 201 args = parser.parse_args() 202 203 main(args.pipeline_urls, args.only_flakes) 204 print("Done.") 205