1#!/usr/bin/python3 2 3import shutil 4import subprocess 5import argparse 6import os 7import glob 8import re 9import pandas as pd 10 11 12def highest_value(inp): 13 ret_value = False 14 for x in inp: 15 if x: 16 return True 17 else: 18 return False 19 20 21def generateTestCompletionTables(output_dir, completion_table): 22 data_table = pd.DataFrame(completion_table, columns=["Agent", "Domain", "Test", "With Asan", "With UBsan"]) 23 data_table.to_html(os.path.join(output_dir, 'completions_table.html')) 24 os.makedirs(os.path.join(output_dir, "post_process"), exist_ok=True) 25 26 pivot_by_agent = pd.pivot_table(data_table, index=["Agent", "Domain", "Test"]) 27 pivot_by_agent.to_html(os.path.join(output_dir, "post_process", 'completions_table_by_agent.html')) 28 pivot_by_test = pd.pivot_table(data_table, index=["Domain", "Test", "Agent"]) 29 pivot_by_test.to_html(os.path.join(output_dir, "post_process", 'completions_table_by_test.html')) 30 pivot_by_asan = pd.pivot_table(data_table, index=["Domain", "Test"], values=["With Asan"], aggfunc=highest_value) 31 pivot_by_asan.to_html(os.path.join(output_dir, "post_process", 'completions_table_by_asan.html')) 32 pivot_by_ubsan = pd.pivot_table(data_table, index=["Domain", "Test"], values=["With UBsan"], aggfunc=highest_value) 33 pivot_by_ubsan.to_html(os.path.join(output_dir, "post_process", 'completions_table_by_ubsan.html')) 34 35 36def generateCoverageReport(output_dir, repo_dir): 37 coveragePath = os.path.join(output_dir, '**', 'cov_total.info') 38 covfiles = [os.path.abspath(p) for p in glob.glob(coveragePath, recursive=True)] 39 for f in covfiles: 40 print(f) 41 if len(covfiles) == 0: 42 return 43 lcov_opts = [ 44 '--rc lcov_branch_coverage=1', 45 '--rc lcov_function_coverage=1', 46 '--rc genhtml_branch_coverage=1', 47 '--rc genhtml_function_coverage=1', 48 '--rc genhtml_legend=1', 49 '--rc geninfo_all_blocks=1', 50 ] 51 cov_total = os.path.abspath(os.path.join(output_dir, 'cov_total.info')) 52 coverage = os.path.join(output_dir, 'coverage') 53 lcov = 'lcov' + ' ' + ' '.join(lcov_opts) + ' -q -a ' + ' -a '.join(covfiles) + ' -o ' + cov_total 54 genhtml = 'genhtml' + ' ' + ' '.join(lcov_opts) + ' -q ' + cov_total + ' --legend' + ' -t "Combined" --show-details -o ' + coverage 55 try: 56 subprocess.check_call([lcov], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 57 except subprocess.CalledProcessError as e: 58 print("lcov failed") 59 print(e) 60 return 61 cov_total_file = open(cov_total, 'r') 62 replacement = "SF:" + repo_dir 63 file_contents = cov_total_file.readlines() 64 cov_total_file.close() 65 os.remove(cov_total) 66 with open(cov_total, 'w+') as file: 67 for Line in file_contents: 68 Line = re.sub("^SF:.*/repo", replacement, Line) 69 file.write(Line + '\n') 70 try: 71 subprocess.check_call([genhtml], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) 72 except subprocess.CalledProcessError as e: 73 print("genhtml failed") 74 print(e) 75 for f in covfiles: 76 os.remove(f) 77 78 79def collectOne(output_dir, dir_name): 80 dirs = glob.glob(os.path.join(output_dir, '*', dir_name)) 81 dirs.sort() 82 if len(dirs) == 0: 83 return 84 85 # Collect first instance of dir_name and move it to the top level 86 collect_dir = dirs.pop(0) 87 shutil.move(collect_dir, os.path.join(output_dir, dir_name)) 88 89 # Delete all other instances 90 for d in dirs: 91 shutil.rmtree(d) 92 93 94def getCompletions(completionFile, test_list, test_completion_table): 95 agent_name = os.path.basename(os.path.dirname(completionFile)) 96 with open(completionFile, 'r') as completionList: 97 completions = completionList.read() 98 99 asan_enabled = "asan" in completions 100 ubsan_enabled = "ubsan" in completions 101 102 for line in completions.splitlines(): 103 try: 104 domain, test_name = line.strip().split() 105 test_list[test_name] = (True, asan_enabled | test_list[test_name][1], ubsan_enabled | test_list[test_name][2]) 106 test_completion_table.append([agent_name, domain, test_name, asan_enabled, ubsan_enabled]) 107 try: 108 test_completion_table.remove(["None", "None", test_name, False, False]) 109 except ValueError: 110 continue 111 except KeyError: 112 continue 113 114 115def printList(header, test_list, index, condition): 116 print("\n\n-----%s------" % header) 117 executed_tests = [x for x in sorted(test_list) if test_list[x][index] is condition] 118 print(*executed_tests, sep="\n") 119 120 121def printListInformation(table_type, test_list): 122 printList("%s Executed in Build" % table_type, test_list, 0, True) 123 printList("%s Missing From Build" % table_type, test_list, 0, False) 124 printList("%s Missing ASAN" % table_type, test_list, 1, False) 125 printList("%s Missing UBSAN" % table_type, test_list, 2, False) 126 127 128def getSkippedTests(repo_dir): 129 skipped_test_file = os.path.join(repo_dir, "test", "common", "skipped_tests.txt") 130 if not os.path.exists(skipped_test_file): 131 return [] 132 else: 133 with open(skipped_test_file, "r") as skipped_test_data: 134 return [x.strip() for x in skipped_test_data.readlines() if "#" not in x and x.strip() != ''] 135 136 137def confirmPerPatchTests(test_list, skiplist): 138 missing_tests = [x for x in sorted(test_list) if test_list[x][0] is False 139 and x not in skiplist] 140 if len(missing_tests) > 0: 141 print("Not all tests were run. Failing the build.") 142 print(missing_tests) 143 exit(1) 144 145 146def aggregateCompletedTests(output_dir, repo_dir, skip_confirm=False): 147 test_list = {} 148 test_completion_table = [] 149 150 testFiles = glob.glob(os.path.join(output_dir, '**', 'all_tests.txt'), recursive=True) 151 completionFiles = glob.glob(os.path.join(output_dir, '**', 'test_completions.txt'), recursive=True) 152 153 if len(testFiles) == 0: 154 print("Unable to perform test completion aggregator. No input files.") 155 return 0 156 157 with open(testFiles[0], 'r') as raw_test_list: 158 for line in raw_test_list: 159 try: 160 test_name = line.strip() 161 except Exception: 162 print("Failed to parse a test type.") 163 return 1 164 165 test_list[test_name] = (False, False, False) 166 test_completion_table.append(["None", "None", test_name, False, False]) 167 168 for completionFile in completionFiles: 169 getCompletions(completionFile, test_list, test_completion_table) 170 171 printListInformation("Tests", test_list) 172 generateTestCompletionTables(output_dir, test_completion_table) 173 skipped_tests = getSkippedTests(repo_dir) 174 if not skip_confirm: 175 confirmPerPatchTests(test_list, skipped_tests) 176 177 178def main(output_dir, repo_dir, skip_confirm=False): 179 print("-----Begin Post Process Script------") 180 generateCoverageReport(output_dir, repo_dir) 181 collectOne(output_dir, 'doc') 182 collectOne(output_dir, 'ut_coverage') 183 aggregateCompletedTests(output_dir, repo_dir, skip_confirm) 184 185 186if __name__ == "__main__": 187 parser = argparse.ArgumentParser(description="SPDK Coverage Processor") 188 parser.add_argument("-d", "--directory_location", type=str, required=True, 189 help="The location of your build's output directory") 190 parser.add_argument("-r", "--repo_directory", type=str, required=True, 191 help="The location of your spdk repository") 192 parser.add_argument("-s", "--skip_confirm", required=False, action="store_true", 193 help="Do not check if all autotest.sh tests were executed.") 194 args = parser.parse_args() 195 main(args.directory_location, args.repo_directory, args.skip_confirm) 196