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