# type: ignore """ report.py - Utilities for reporting statistics about benchmark results """ import copy import os import random import re import unittest from numpy import array from scipy.stats import gmean, mannwhitneyu class BenchmarkColor(object): def __init__(self, name, code): self.name = name self.code = code def __repr__(self): return "%s%r" % (self.__class__.__name__, (self.name, self.code)) def __format__(self, format): return self.code # Benchmark Colors Enumeration BC_NONE = BenchmarkColor("NONE", "") BC_MAGENTA = BenchmarkColor("MAGENTA", "\033[95m") BC_CYAN = BenchmarkColor("CYAN", "\033[96m") BC_OKBLUE = BenchmarkColor("OKBLUE", "\033[94m") BC_OKGREEN = BenchmarkColor("OKGREEN", "\033[32m") BC_HEADER = BenchmarkColor("HEADER", "\033[92m") BC_WARNING = BenchmarkColor("WARNING", "\033[93m") BC_WHITE = BenchmarkColor("WHITE", "\033[97m") BC_FAIL = BenchmarkColor("FAIL", "\033[91m") BC_ENDC = BenchmarkColor("ENDC", "\033[0m") BC_BOLD = BenchmarkColor("BOLD", "\033[1m") BC_UNDERLINE = BenchmarkColor("UNDERLINE", "\033[4m") UTEST_MIN_REPETITIONS = 2 UTEST_OPTIMAL_REPETITIONS = 9 # Lowest reasonable number, More is better. UTEST_COL_NAME = "_pvalue" _TIME_UNIT_TO_SECONDS_MULTIPLIER = { "s": 1.0, "ms": 1e-3, "us": 1e-6, "ns": 1e-9, } def color_format(use_color, fmt_str, *args, **kwargs): """ Return the result of 'fmt_str.format(*args, **kwargs)' after transforming 'args' and 'kwargs' according to the value of 'use_color'. If 'use_color' is False then all color codes in 'args' and 'kwargs' are replaced with the empty string. """ assert use_color is True or use_color is False if not use_color: args = [ arg if not isinstance(arg, BenchmarkColor) else BC_NONE for arg in args ] kwargs = { key: arg if not isinstance(arg, BenchmarkColor) else BC_NONE for key, arg in kwargs.items() } return fmt_str.format(*args, **kwargs) def find_longest_name(benchmark_list): """ Return the length of the longest benchmark name in a given list of benchmark JSON objects """ longest_name = 1 for bc in benchmark_list: if len(bc["name"]) > longest_name: longest_name = len(bc["name"]) return longest_name def calculate_change(old_val, new_val): """ Return a float representing the decimal change between old_val and new_val. """ if old_val == 0 and new_val == 0: return 0.0 if old_val == 0: return float(new_val - old_val) / (float(old_val + new_val) / 2) return float(new_val - old_val) / abs(old_val) def filter_benchmark(json_orig, family, replacement=""): """ Apply a filter to the json, and only leave the 'family' of benchmarks. """ regex = re.compile(family) filtered = {} filtered["benchmarks"] = [] for be in json_orig["benchmarks"]: if not regex.search(be["name"]): continue filteredbench = copy.deepcopy(be) # Do NOT modify the old name! filteredbench["name"] = regex.sub(replacement, filteredbench["name"]) filtered["benchmarks"].append(filteredbench) return filtered def get_unique_benchmark_names(json): """ While *keeping* the order, give all the unique 'names' used for benchmarks. """ seen = set() uniqued = [ x["name"] for x in json["benchmarks"] if x["name"] not in seen and (seen.add(x["name"]) or True) ] return uniqued def intersect(list1, list2): """ Given two lists, get a new list consisting of the elements only contained in *both of the input lists*, while preserving the ordering. """ return [x for x in list1 if x in list2] def is_potentially_comparable_benchmark(x): return "time_unit" in x and "real_time" in x and "cpu_time" in x def partition_benchmarks(json1, json2): """ While preserving the ordering, find benchmarks with the same names in both of the inputs, and group them. (i.e. partition/filter into groups with common name) """ json1_unique_names = get_unique_benchmark_names(json1) json2_unique_names = get_unique_benchmark_names(json2) names = intersect(json1_unique_names, json2_unique_names) partitions = [] for name in names: time_unit = None # Pick the time unit from the first entry of the lhs benchmark. # We should be careful not to crash with unexpected input. for x in json1["benchmarks"]: if x["name"] == name and is_potentially_comparable_benchmark(x): time_unit = x["time_unit"] break if time_unit is None: continue # Filter by name and time unit. # All the repetitions are assumed to be comparable. lhs = [ x for x in json1["benchmarks"] if x["name"] == name and x["time_unit"] == time_unit ] rhs = [ x for x in json2["benchmarks"] if x["name"] == name and x["time_unit"] == time_unit ] partitions.append([lhs, rhs]) return partitions def get_timedelta_field_as_seconds(benchmark, field_name): """ Get value of field_name field of benchmark, which is time with time unit time_unit, as time in seconds. """ timedelta = benchmark[field_name] time_unit = benchmark.get("time_unit", "s") return timedelta * _TIME_UNIT_TO_SECONDS_MULTIPLIER.get(time_unit) def calculate_geomean(json): """ Extract all real/cpu times from all the benchmarks as seconds, and calculate their geomean. """ times = [] for benchmark in json["benchmarks"]: if "run_type" in benchmark and benchmark["run_type"] == "aggregate": continue times.append( [ get_timedelta_field_as_seconds(benchmark, "real_time"), get_timedelta_field_as_seconds(benchmark, "cpu_time"), ] ) return gmean(times) if times else array([]) def extract_field(partition, field_name): # The count of elements may be different. We want *all* of them. lhs = [x[field_name] for x in partition[0]] rhs = [x[field_name] for x in partition[1]] return [lhs, rhs] def calc_utest(timings_cpu, timings_time): min_rep_cnt = min( len(timings_time[0]), len(timings_time[1]), len(timings_cpu[0]), len(timings_cpu[1]), ) # Does *everything* has at least UTEST_MIN_REPETITIONS repetitions? if min_rep_cnt < UTEST_MIN_REPETITIONS: return False, None, None time_pvalue = mannwhitneyu( timings_time[0], timings_time[1], alternative="two-sided" ).pvalue cpu_pvalue = mannwhitneyu( timings_cpu[0], timings_cpu[1], alternative="two-sided" ).pvalue return (min_rep_cnt >= UTEST_OPTIMAL_REPETITIONS), cpu_pvalue, time_pvalue def print_utest(bc_name, utest, utest_alpha, first_col_width, use_color=True): def get_utest_color(pval): return BC_FAIL if pval >= utest_alpha else BC_OKGREEN # Check if we failed miserably with minimum required repetitions for utest if ( not utest["have_optimal_repetitions"] and utest["cpu_pvalue"] is None and utest["time_pvalue"] is None ): return [] dsc = "U Test, Repetitions: {} vs {}".format( utest["nr_of_repetitions"], utest["nr_of_repetitions_other"] ) dsc_color = BC_OKGREEN # We still got some results to show but issue a warning about it. if not utest["have_optimal_repetitions"]: dsc_color = BC_WARNING dsc += ". WARNING: Results unreliable! {}+ repetitions recommended.".format( UTEST_OPTIMAL_REPETITIONS ) special_str = "{}{:<{}s}{endc}{}{:16.4f}{endc}{}{:16.4f}{endc}{} {}" return [ color_format( use_color, special_str, BC_HEADER, "{}{}".format(bc_name, UTEST_COL_NAME), first_col_width, get_utest_color(utest["time_pvalue"]), utest["time_pvalue"], get_utest_color(utest["cpu_pvalue"]), utest["cpu_pvalue"], dsc_color, dsc, endc=BC_ENDC, ) ] def get_difference_report(json1, json2, utest=False): """ Calculate and report the difference between each test of two benchmarks runs specified as 'json1' and 'json2'. Output is another json containing relevant details for each test run. """ assert utest is True or utest is False diff_report = [] partitions = partition_benchmarks(json1, json2) for partition in partitions: benchmark_name = partition[0][0]["name"] label = partition[0][0]["label"] if "label" in partition[0][0] else "" time_unit = partition[0][0]["time_unit"] measurements = [] utest_results = {} # Careful, we may have different repetition count. for i in range(min(len(partition[0]), len(partition[1]))): bn = partition[0][i] other_bench = partition[1][i] measurements.append( { "real_time": bn["real_time"], "cpu_time": bn["cpu_time"], "real_time_other": other_bench["real_time"], "cpu_time_other": other_bench["cpu_time"], "time": calculate_change( bn["real_time"], other_bench["real_time"] ), "cpu": calculate_change( bn["cpu_time"], other_bench["cpu_time"] ), } ) # After processing the whole partition, if requested, do the U test. if utest: timings_cpu = extract_field(partition, "cpu_time") timings_time = extract_field(partition, "real_time") have_optimal_repetitions, cpu_pvalue, time_pvalue = calc_utest( timings_cpu, timings_time ) if cpu_pvalue is not None and time_pvalue is not None: utest_results = { "have_optimal_repetitions": have_optimal_repetitions, "cpu_pvalue": cpu_pvalue, "time_pvalue": time_pvalue, "nr_of_repetitions": len(timings_cpu[0]), "nr_of_repetitions_other": len(timings_cpu[1]), } # Store only if we had any measurements for given benchmark. # E.g. partition_benchmarks will filter out the benchmarks having # time units which are not compatible with other time units in the # benchmark suite. if measurements: run_type = ( partition[0][0]["run_type"] if "run_type" in partition[0][0] else "" ) aggregate_name = ( partition[0][0]["aggregate_name"] if run_type == "aggregate" and "aggregate_name" in partition[0][0] else "" ) diff_report.append( { "name": benchmark_name, "label": label, "measurements": measurements, "time_unit": time_unit, "run_type": run_type, "aggregate_name": aggregate_name, "utest": utest_results, } ) lhs_gmean = calculate_geomean(json1) rhs_gmean = calculate_geomean(json2) if lhs_gmean.any() and rhs_gmean.any(): diff_report.append( { "name": "OVERALL_GEOMEAN", "label": "", "measurements": [ { "real_time": lhs_gmean[0], "cpu_time": lhs_gmean[1], "real_time_other": rhs_gmean[0], "cpu_time_other": rhs_gmean[1], "time": calculate_change(lhs_gmean[0], rhs_gmean[0]), "cpu": calculate_change(lhs_gmean[1], rhs_gmean[1]), } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, } ) return diff_report def print_difference_report( json_diff_report, include_aggregates_only=False, utest=False, utest_alpha=0.05, use_color=True, ): """ Calculate and report the difference between each test of two benchmarks runs specified as 'json1' and 'json2'. """ assert utest is True or utest is False def get_color(res): if res > 0.05: return BC_FAIL elif res > -0.07: return BC_WHITE else: return BC_CYAN first_col_width = find_longest_name(json_diff_report) first_col_width = max(first_col_width, len("Benchmark")) first_col_width += len(UTEST_COL_NAME) first_line = "{:<{}s}Time CPU Time Old Time New CPU Old CPU New".format( "Benchmark", 12 + first_col_width ) output_strs = [first_line, "-" * len(first_line)] fmt_str = "{}{:<{}s}{endc}{}{:+16.4f}{endc}{}{:+16.4f}{endc}{:14.0f}{:14.0f}{endc}{:14.0f}{:14.0f}" for benchmark in json_diff_report: # *If* we were asked to only include aggregates, # and if it is non-aggregate, then don't print it. if ( not include_aggregates_only or "run_type" not in benchmark or benchmark["run_type"] == "aggregate" ): for measurement in benchmark["measurements"]: output_strs += [ color_format( use_color, fmt_str, BC_HEADER, benchmark["name"], first_col_width, get_color(measurement["time"]), measurement["time"], get_color(measurement["cpu"]), measurement["cpu"], measurement["real_time"], measurement["real_time_other"], measurement["cpu_time"], measurement["cpu_time_other"], endc=BC_ENDC, ) ] # After processing the measurements, if requested and # if applicable (e.g. u-test exists for given benchmark), # print the U test. if utest and benchmark["utest"]: output_strs += print_utest( benchmark["name"], benchmark["utest"], utest_alpha=utest_alpha, first_col_width=first_col_width, use_color=use_color, ) return output_strs ############################################################################### # Unit tests class TestGetUniqueBenchmarkNames(unittest.TestCase): def load_results(self): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput = os.path.join(testInputs, "test3_run0.json") with open(testOutput, "r") as f: json = json.load(f) return json def test_basic(self): expect_lines = [ "BM_One", "BM_Two", "short", # These two are not sorted "medium", # These two are not sorted ] json = self.load_results() output_lines = get_unique_benchmark_names(json) print("\n") print("\n".join(output_lines)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): self.assertEqual(expect_lines[i], output_lines[i]) class TestReportDifference(unittest.TestCase): @classmethod def setUpClass(cls): def load_results(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput1 = os.path.join(testInputs, "test1_run1.json") testOutput2 = os.path.join(testInputs, "test1_run2.json") with open(testOutput1, "r") as f: json1 = json.load(f) with open(testOutput2, "r") as f: json2 = json.load(f) return json1, json2 json1, json2 = load_results() cls.json_diff_report = get_difference_report(json1, json2) def test_json_diff_report_pretty_printing(self): expect_lines = [ ["BM_SameTimes", "+0.0000", "+0.0000", "10", "10", "10", "10"], ["BM_2xFaster", "-0.5000", "-0.5000", "50", "25", "50", "25"], ["BM_2xSlower", "+1.0000", "+1.0000", "50", "100", "50", "100"], [ "BM_1PercentFaster", "-0.0100", "-0.0100", "100", "99", "100", "99", ], [ "BM_1PercentSlower", "+0.0100", "+0.0100", "100", "101", "100", "101", ], [ "BM_10PercentFaster", "-0.1000", "-0.1000", "100", "90", "100", "90", ], [ "BM_10PercentSlower", "+0.1000", "+0.1000", "100", "110", "100", "110", ], [ "BM_100xSlower", "+99.0000", "+99.0000", "100", "10000", "100", "10000", ], [ "BM_100xFaster", "-0.9900", "-0.9900", "10000", "100", "10000", "100", ], [ "BM_10PercentCPUToTime", "+0.1000", "-0.1000", "100", "110", "100", "90", ], ["BM_ThirdFaster", "-0.3333", "-0.3334", "100", "67", "100", "67"], ["BM_NotBadTimeUnit", "-0.9000", "+0.2000", "0", "0", "0", "1"], ["BM_hasLabel", "+0.0000", "+0.0000", "1", "1", "1", "1"], ["OVERALL_GEOMEAN", "-0.8113", "-0.7779", "0", "0", "0", "0"], ] output_lines_with_header = print_difference_report( self.json_diff_report, use_color=False ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(len(parts), 7) self.assertEqual(expect_lines[i], parts) def test_json_diff_report_output(self): expected_output = [ { "name": "BM_SameTimes", "label": "", "measurements": [ { "time": 0.0000, "cpu": 0.0000, "real_time": 10, "real_time_other": 10, "cpu_time": 10, "cpu_time_other": 10, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_2xFaster", "label": "", "measurements": [ { "time": -0.5000, "cpu": -0.5000, "real_time": 50, "real_time_other": 25, "cpu_time": 50, "cpu_time_other": 25, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_2xSlower", "label": "", "measurements": [ { "time": 1.0000, "cpu": 1.0000, "real_time": 50, "real_time_other": 100, "cpu_time": 50, "cpu_time_other": 100, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_1PercentFaster", "label": "", "measurements": [ { "time": -0.0100, "cpu": -0.0100, "real_time": 100, "real_time_other": 98.9999999, "cpu_time": 100, "cpu_time_other": 98.9999999, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_1PercentSlower", "label": "", "measurements": [ { "time": 0.0100, "cpu": 0.0100, "real_time": 100, "real_time_other": 101, "cpu_time": 100, "cpu_time_other": 101, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_10PercentFaster", "label": "", "measurements": [ { "time": -0.1000, "cpu": -0.1000, "real_time": 100, "real_time_other": 90, "cpu_time": 100, "cpu_time_other": 90, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_10PercentSlower", "label": "", "measurements": [ { "time": 0.1000, "cpu": 0.1000, "real_time": 100, "real_time_other": 110, "cpu_time": 100, "cpu_time_other": 110, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_100xSlower", "label": "", "measurements": [ { "time": 99.0000, "cpu": 99.0000, "real_time": 100, "real_time_other": 10000, "cpu_time": 100, "cpu_time_other": 10000, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_100xFaster", "label": "", "measurements": [ { "time": -0.9900, "cpu": -0.9900, "real_time": 10000, "real_time_other": 100, "cpu_time": 10000, "cpu_time_other": 100, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_10PercentCPUToTime", "label": "", "measurements": [ { "time": 0.1000, "cpu": -0.1000, "real_time": 100, "real_time_other": 110, "cpu_time": 100, "cpu_time_other": 90, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_ThirdFaster", "label": "", "measurements": [ { "time": -0.3333, "cpu": -0.3334, "real_time": 100, "real_time_other": 67, "cpu_time": 100, "cpu_time_other": 67, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_NotBadTimeUnit", "label": "", "measurements": [ { "time": -0.9000, "cpu": 0.2000, "real_time": 0.4, "real_time_other": 0.04, "cpu_time": 0.5, "cpu_time_other": 0.6, } ], "time_unit": "s", "utest": {}, }, { "name": "BM_hasLabel", "label": "a label", "measurements": [ { "time": 0.0000, "cpu": 0.0000, "real_time": 1, "real_time_other": 1, "cpu_time": 1, "cpu_time_other": 1, } ], "time_unit": "s", "utest": {}, }, { "name": "OVERALL_GEOMEAN", "label": "", "measurements": [ { "real_time": 3.1622776601683826e-06, "cpu_time": 3.2130844755623912e-06, "real_time_other": 1.9768988699420897e-07, "cpu_time_other": 2.397447755209533e-07, "time": -0.8112976497120911, "cpu": -0.7778551721181174, } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, }, ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["label"], expected["label"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) assert_measurements(self, out, expected) class TestReportDifferenceBetweenFamilies(unittest.TestCase): @classmethod def setUpClass(cls): def load_result(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput = os.path.join(testInputs, "test2_run.json") with open(testOutput, "r") as f: json = json.load(f) return json json = load_result() json1 = filter_benchmark(json, "BM_Z.ro", ".") json2 = filter_benchmark(json, "BM_O.e", ".") cls.json_diff_report = get_difference_report(json1, json2) def test_json_diff_report_pretty_printing(self): expect_lines = [ [".", "-0.5000", "-0.5000", "10", "5", "10", "5"], ["./4", "-0.5000", "-0.5000", "40", "20", "40", "20"], ["Prefix/.", "-0.5000", "-0.5000", "20", "10", "20", "10"], ["Prefix/./3", "-0.5000", "-0.5000", "30", "15", "30", "15"], ["OVERALL_GEOMEAN", "-0.5000", "-0.5000", "0", "0", "0", "0"], ] output_lines_with_header = print_difference_report( self.json_diff_report, use_color=False ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(len(parts), 7) self.assertEqual(expect_lines[i], parts) def test_json_diff_report(self): expected_output = [ { "name": ".", "measurements": [ { "time": -0.5, "cpu": -0.5, "real_time": 10, "real_time_other": 5, "cpu_time": 10, "cpu_time_other": 5, } ], "time_unit": "ns", "utest": {}, }, { "name": "./4", "measurements": [ { "time": -0.5, "cpu": -0.5, "real_time": 40, "real_time_other": 20, "cpu_time": 40, "cpu_time_other": 20, } ], "time_unit": "ns", "utest": {}, }, { "name": "Prefix/.", "measurements": [ { "time": -0.5, "cpu": -0.5, "real_time": 20, "real_time_other": 10, "cpu_time": 20, "cpu_time_other": 10, } ], "time_unit": "ns", "utest": {}, }, { "name": "Prefix/./3", "measurements": [ { "time": -0.5, "cpu": -0.5, "real_time": 30, "real_time_other": 15, "cpu_time": 30, "cpu_time_other": 15, } ], "time_unit": "ns", "utest": {}, }, { "name": "OVERALL_GEOMEAN", "measurements": [ { "real_time": 2.213363839400641e-08, "cpu_time": 2.213363839400641e-08, "real_time_other": 1.1066819197003185e-08, "cpu_time_other": 1.1066819197003185e-08, "time": -0.5000000000000009, "cpu": -0.5000000000000009, } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, }, ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) assert_measurements(self, out, expected) class TestReportDifferenceWithUTest(unittest.TestCase): @classmethod def setUpClass(cls): def load_results(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput1 = os.path.join(testInputs, "test3_run0.json") testOutput2 = os.path.join(testInputs, "test3_run1.json") with open(testOutput1, "r") as f: json1 = json.load(f) with open(testOutput2, "r") as f: json2 = json.load(f) return json1, json2 json1, json2 = load_results() cls.json_diff_report = get_difference_report(json1, json2, utest=True) def test_json_diff_report_pretty_printing(self): expect_lines = [ ["BM_One", "-0.1000", "+0.1000", "10", "9", "100", "110"], ["BM_Two", "+0.1111", "-0.0111", "9", "10", "90", "89"], ["BM_Two", "-0.1250", "-0.1628", "8", "7", "86", "72"], [ "BM_Two_pvalue", "1.0000", "0.6667", "U", "Test,", "Repetitions:", "2", "vs", "2.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["short", "-0.1250", "-0.0625", "8", "7", "80", "75"], ["short", "-0.4325", "-0.1351", "8", "5", "77", "67"], [ "short_pvalue", "0.7671", "0.2000", "U", "Test,", "Repetitions:", "2", "vs", "3.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["medium", "-0.3750", "-0.3375", "8", "5", "80", "53"], ["OVERALL_GEOMEAN", "+1.6405", "-0.6985", "0", "0", "0", "0"], ] output_lines_with_header = print_difference_report( self.json_diff_report, utest=True, utest_alpha=0.05, use_color=False ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(expect_lines[i], parts) def test_json_diff_report_pretty_printing_aggregates_only(self): expect_lines = [ ["BM_One", "-0.1000", "+0.1000", "10", "9", "100", "110"], [ "BM_Two_pvalue", "1.0000", "0.6667", "U", "Test,", "Repetitions:", "2", "vs", "2.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["short", "-0.1250", "-0.0625", "8", "7", "80", "75"], ["short", "-0.4325", "-0.1351", "8", "5", "77", "67"], [ "short_pvalue", "0.7671", "0.2000", "U", "Test,", "Repetitions:", "2", "vs", "3.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["OVERALL_GEOMEAN", "+1.6405", "-0.6985", "0", "0", "0", "0"], ] output_lines_with_header = print_difference_report( self.json_diff_report, include_aggregates_only=True, utest=True, utest_alpha=0.05, use_color=False, ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(expect_lines[i], parts) def test_json_diff_report(self): expected_output = [ { "name": "BM_One", "measurements": [ { "time": -0.1, "cpu": 0.1, "real_time": 10, "real_time_other": 9, "cpu_time": 100, "cpu_time_other": 110, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_Two", "measurements": [ { "time": 0.1111111111111111, "cpu": -0.011111111111111112, "real_time": 9, "real_time_other": 10, "cpu_time": 90, "cpu_time_other": 89, }, { "time": -0.125, "cpu": -0.16279069767441862, "real_time": 8, "real_time_other": 7, "cpu_time": 86, "cpu_time_other": 72, }, ], "time_unit": "ns", "utest": { "have_optimal_repetitions": False, "cpu_pvalue": 0.6666666666666666, "time_pvalue": 1.0, }, }, { "name": "short", "measurements": [ { "time": -0.125, "cpu": -0.0625, "real_time": 8, "real_time_other": 7, "cpu_time": 80, "cpu_time_other": 75, }, { "time": -0.4325, "cpu": -0.13506493506493514, "real_time": 8, "real_time_other": 4.54, "cpu_time": 77, "cpu_time_other": 66.6, }, ], "time_unit": "ns", "utest": { "have_optimal_repetitions": False, "cpu_pvalue": 0.2, "time_pvalue": 0.7670968684102772, }, }, { "name": "medium", "measurements": [ { "time": -0.375, "cpu": -0.3375, "real_time": 8, "real_time_other": 5, "cpu_time": 80, "cpu_time_other": 53, } ], "time_unit": "ns", "utest": {}, }, { "name": "OVERALL_GEOMEAN", "measurements": [ { "real_time": 8.48528137423858e-09, "cpu_time": 8.441336246629233e-08, "real_time_other": 2.2405267593145244e-08, "cpu_time_other": 2.5453661413660466e-08, "time": 1.6404861082353634, "cpu": -0.6984640740519662, } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, }, ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) assert_measurements(self, out, expected) class TestReportDifferenceWithUTestWhileDisplayingAggregatesOnly( unittest.TestCase ): @classmethod def setUpClass(cls): def load_results(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput1 = os.path.join(testInputs, "test3_run0.json") testOutput2 = os.path.join(testInputs, "test3_run1.json") with open(testOutput1, "r") as f: json1 = json.load(f) with open(testOutput2, "r") as f: json2 = json.load(f) return json1, json2 json1, json2 = load_results() cls.json_diff_report = get_difference_report(json1, json2, utest=True) def test_json_diff_report_pretty_printing(self): expect_lines = [ ["BM_One", "-0.1000", "+0.1000", "10", "9", "100", "110"], ["BM_Two", "+0.1111", "-0.0111", "9", "10", "90", "89"], ["BM_Two", "-0.1250", "-0.1628", "8", "7", "86", "72"], [ "BM_Two_pvalue", "1.0000", "0.6667", "U", "Test,", "Repetitions:", "2", "vs", "2.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["short", "-0.1250", "-0.0625", "8", "7", "80", "75"], ["short", "-0.4325", "-0.1351", "8", "5", "77", "67"], [ "short_pvalue", "0.7671", "0.2000", "U", "Test,", "Repetitions:", "2", "vs", "3.", "WARNING:", "Results", "unreliable!", "9+", "repetitions", "recommended.", ], ["medium", "-0.3750", "-0.3375", "8", "5", "80", "53"], ["OVERALL_GEOMEAN", "+1.6405", "-0.6985", "0", "0", "0", "0"], ] output_lines_with_header = print_difference_report( self.json_diff_report, utest=True, utest_alpha=0.05, use_color=False ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(expect_lines[i], parts) def test_json_diff_report(self): expected_output = [ { "name": "BM_One", "measurements": [ { "time": -0.1, "cpu": 0.1, "real_time": 10, "real_time_other": 9, "cpu_time": 100, "cpu_time_other": 110, } ], "time_unit": "ns", "utest": {}, }, { "name": "BM_Two", "measurements": [ { "time": 0.1111111111111111, "cpu": -0.011111111111111112, "real_time": 9, "real_time_other": 10, "cpu_time": 90, "cpu_time_other": 89, }, { "time": -0.125, "cpu": -0.16279069767441862, "real_time": 8, "real_time_other": 7, "cpu_time": 86, "cpu_time_other": 72, }, ], "time_unit": "ns", "utest": { "have_optimal_repetitions": False, "cpu_pvalue": 0.6666666666666666, "time_pvalue": 1.0, }, }, { "name": "short", "measurements": [ { "time": -0.125, "cpu": -0.0625, "real_time": 8, "real_time_other": 7, "cpu_time": 80, "cpu_time_other": 75, }, { "time": -0.4325, "cpu": -0.13506493506493514, "real_time": 8, "real_time_other": 4.54, "cpu_time": 77, "cpu_time_other": 66.6, }, ], "time_unit": "ns", "utest": { "have_optimal_repetitions": False, "cpu_pvalue": 0.2, "time_pvalue": 0.7670968684102772, }, }, { "name": "medium", "measurements": [ { "real_time_other": 5, "cpu_time": 80, "time": -0.375, "real_time": 8, "cpu_time_other": 53, "cpu": -0.3375, } ], "utest": {}, "time_unit": "ns", "aggregate_name": "", }, { "name": "OVERALL_GEOMEAN", "measurements": [ { "real_time": 8.48528137423858e-09, "cpu_time": 8.441336246629233e-08, "real_time_other": 2.2405267593145244e-08, "cpu_time_other": 2.5453661413660466e-08, "time": 1.6404861082353634, "cpu": -0.6984640740519662, } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, }, ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) assert_measurements(self, out, expected) class TestReportDifferenceForPercentageAggregates(unittest.TestCase): @classmethod def setUpClass(cls): def load_results(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput1 = os.path.join(testInputs, "test4_run0.json") testOutput2 = os.path.join(testInputs, "test4_run1.json") with open(testOutput1, "r") as f: json1 = json.load(f) with open(testOutput2, "r") as f: json2 = json.load(f) return json1, json2 json1, json2 = load_results() cls.json_diff_report = get_difference_report(json1, json2, utest=True) def test_json_diff_report_pretty_printing(self): expect_lines = [["whocares", "-0.5000", "+0.5000", "0", "0", "0", "0"]] output_lines_with_header = print_difference_report( self.json_diff_report, utest=True, utest_alpha=0.05, use_color=False ) output_lines = output_lines_with_header[2:] print("\n") print("\n".join(output_lines_with_header)) self.assertEqual(len(output_lines), len(expect_lines)) for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] self.assertEqual(expect_lines[i], parts) def test_json_diff_report(self): expected_output = [ { "name": "whocares", "measurements": [ { "time": -0.5, "cpu": 0.5, "real_time": 0.01, "real_time_other": 0.005, "cpu_time": 0.10, "cpu_time_other": 0.15, } ], "time_unit": "ns", "utest": {}, } ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) assert_measurements(self, out, expected) class TestReportSorting(unittest.TestCase): @classmethod def setUpClass(cls): def load_result(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput = os.path.join(testInputs, "test4_run.json") with open(testOutput, "r") as f: json = json.load(f) return json cls.json = load_result() def test_json_diff_report_pretty_printing(self): import util expected_names = [ "99 family 0 instance 0 repetition 0", "98 family 0 instance 0 repetition 1", "97 family 0 instance 0 aggregate", "96 family 0 instance 1 repetition 0", "95 family 0 instance 1 repetition 1", "94 family 0 instance 1 aggregate", "93 family 1 instance 0 repetition 0", "92 family 1 instance 0 repetition 1", "91 family 1 instance 0 aggregate", "90 family 1 instance 1 repetition 0", "89 family 1 instance 1 repetition 1", "88 family 1 instance 1 aggregate", ] for n in range(len(self.json["benchmarks"]) ** 2): random.shuffle(self.json["benchmarks"]) sorted_benchmarks = util.sort_benchmark_results(self.json)[ "benchmarks" ] self.assertEqual(len(expected_names), len(sorted_benchmarks)) for out, expected in zip(sorted_benchmarks, expected_names): self.assertEqual(out["name"], expected) class TestReportDifferenceWithUTestWhileDisplayingAggregatesOnly2( unittest.TestCase ): @classmethod def setUpClass(cls): def load_results(): import json testInputs = os.path.join( os.path.dirname(os.path.realpath(__file__)), "Inputs" ) testOutput1 = os.path.join(testInputs, "test5_run0.json") testOutput2 = os.path.join(testInputs, "test5_run1.json") with open(testOutput1, "r") as f: json1 = json.load(f) json1["benchmarks"] = [ json1["benchmarks"][0] for i in range(1000) ] with open(testOutput2, "r") as f: json2 = json.load(f) json2["benchmarks"] = [ json2["benchmarks"][0] for i in range(1000) ] return json1, json2 json1, json2 = load_results() cls.json_diff_report = get_difference_report(json1, json2, utest=True) def test_json_diff_report_pretty_printing(self): expect_line = [ "BM_ManyRepetitions_pvalue", "0.0000", "0.0000", "U", "Test,", "Repetitions:", "1000", "vs", "1000", ] output_lines_with_header = print_difference_report( self.json_diff_report, utest=True, utest_alpha=0.05, use_color=False ) output_lines = output_lines_with_header[2:] found = False for i in range(0, len(output_lines)): parts = [x for x in output_lines[i].split(" ") if x] found = expect_line == parts if found: break self.assertTrue(found) def test_json_diff_report(self): expected_output = [ { "name": "BM_ManyRepetitions", "label": "", "time_unit": "s", "run_type": "", "aggregate_name": "", "utest": { "have_optimal_repetitions": True, "cpu_pvalue": 0.0, "time_pvalue": 0.0, "nr_of_repetitions": 1000, "nr_of_repetitions_other": 1000, }, }, { "name": "OVERALL_GEOMEAN", "label": "", "measurements": [ { "real_time": 1.0, "cpu_time": 1000.000000000069, "real_time_other": 1000.000000000069, "cpu_time_other": 1.0, "time": 999.000000000069, "cpu": -0.9990000000000001, } ], "time_unit": "s", "run_type": "aggregate", "aggregate_name": "geomean", "utest": {}, }, ] self.assertEqual(len(self.json_diff_report), len(expected_output)) for out, expected in zip(self.json_diff_report, expected_output): self.assertEqual(out["name"], expected["name"]) self.assertEqual(out["time_unit"], expected["time_unit"]) assert_utest(self, out, expected) def assert_utest(unittest_instance, lhs, rhs): if lhs["utest"]: unittest_instance.assertAlmostEqual( lhs["utest"]["cpu_pvalue"], rhs["utest"]["cpu_pvalue"] ) unittest_instance.assertAlmostEqual( lhs["utest"]["time_pvalue"], rhs["utest"]["time_pvalue"] ) unittest_instance.assertEqual( lhs["utest"]["have_optimal_repetitions"], rhs["utest"]["have_optimal_repetitions"], ) else: # lhs is empty. assert if rhs is not. unittest_instance.assertEqual(lhs["utest"], rhs["utest"]) def assert_measurements(unittest_instance, lhs, rhs): for m1, m2 in zip(lhs["measurements"], rhs["measurements"]): unittest_instance.assertEqual(m1["real_time"], m2["real_time"]) unittest_instance.assertEqual(m1["cpu_time"], m2["cpu_time"]) # m1['time'] and m1['cpu'] hold values which are being calculated, # and therefore we must use almost-equal pattern. unittest_instance.assertAlmostEqual(m1["time"], m2["time"], places=4) unittest_instance.assertAlmostEqual(m1["cpu"], m2["cpu"], places=4) if __name__ == "__main__": unittest.main() # vim: tabstop=4 expandtab shiftwidth=4 softtabstop=4 # kate: tab-width: 4; replace-tabs on; indent-width 4; tab-indents: off; # kate: indent-mode python; remove-trailing-spaces modified;