#!/usr/bin/env python """ CmpRuns - A simple tool for comparing two static analyzer runs to determine which reports have been added, removed, or changed. This is designed to support automated testing using the static analyzer, from two perspectives: 1. To monitor changes in the static analyzer's reports on real code bases, for regression testing. 2. For use by end users who want to integrate regular static analyzer testing into a buildbot like environment. Usage: # Load the results of both runs, to obtain lists of the corresponding # AnalysisDiagnostic objects. # resultsA = load_results_from_single_run(singleRunInfoA, delete_empty) resultsB = load_results_from_single_run(singleRunInfoB, delete_empty) # Generate a relation from diagnostics in run A to diagnostics in run B # to obtain a list of triples (a, b, confidence). diff = compare_results(resultsA, resultsB) """ import json import os import plistlib import re import sys from math import log from collections import defaultdict from copy import copy from enum import Enum from typing import (Any, DefaultDict, Dict, List, NamedTuple, Optional, Sequence, Set, TextIO, TypeVar, Tuple, Union) Number = Union[int, float] Stats = Dict[str, Dict[str, Number]] Plist = Dict[str, Any] JSON = Dict[str, Any] # Diff in a form: field -> (before, after) JSONDiff = Dict[str, Tuple[str, str]] # Type for generics T = TypeVar('T') STATS_REGEXP = re.compile(r"Statistics: (\{.+\})", re.MULTILINE | re.DOTALL) class Colors: """ Color for terminal highlight. """ RED = '\x1b[2;30;41m' GREEN = '\x1b[6;30;42m' CLEAR = '\x1b[0m' class HistogramType(str, Enum): RELATIVE = "relative" LOG_RELATIVE = "log-relative" ABSOLUTE = "absolute" class ResultsDirectory(NamedTuple): path: str root: str = "" class SingleRunInfo: """ Information about analysis run: path - the analysis output directory root - the name of the root directory, which will be disregarded when determining the source file name """ def __init__(self, results: ResultsDirectory, verbose_log: Optional[str] = None): self.path = results.path self.root = results.root.rstrip("/\\") self.verbose_log = verbose_log class AnalysisDiagnostic: def __init__(self, data: Plist, report: "AnalysisReport", html_report: Optional[str]): self._data = data self._loc = self._data['location'] self._report = report self._html_report = html_report self._report_size = len(self._data['path']) def get_file_name(self) -> str: root = self._report.run.root file_name = self._report.files[self._loc['file']] if file_name.startswith(root) and len(root) > 0: return file_name[len(root) + 1:] return file_name def get_root_file_name(self) -> str: path = self._data['path'] if not path: return self.get_file_name() p = path[0] if 'location' in p: file_index = p['location']['file'] else: # control edge file_index = path[0]['edges'][0]['start'][0]['file'] out = self._report.files[file_index] root = self._report.run.root if out.startswith(root): return out[len(root):] return out def get_line(self) -> int: return self._loc['line'] def get_column(self) -> int: return self._loc['col'] def get_path_length(self) -> int: return self._report_size def get_category(self) -> str: return self._data['category'] def get_description(self) -> str: return self._data['description'] def get_location(self) -> str: return f"{self.get_file_name()}:{self.get_line()}:{self.get_column()}" def get_issue_identifier(self) -> str: id = self.get_file_name() + "+" if "issue_context" in self._data: id += self._data["issue_context"] + "+" if "issue_hash_content_of_line_in_context" in self._data: id += str(self._data["issue_hash_content_of_line_in_context"]) return id def get_html_report(self) -> str: if self._html_report is None: return " " return os.path.join(self._report.run.path, self._html_report) def get_readable_name(self) -> str: if "issue_context" in self._data: funcname_postfix = "#" + self._data["issue_context"] else: funcname_postfix = "" root_filename = self.get_root_file_name() file_name = self.get_file_name() if root_filename != file_name: file_prefix = f"[{root_filename}] {file_name}" else: file_prefix = root_filename line = self.get_line() col = self.get_column() return f"{file_prefix}{funcname_postfix}:{line}:{col}" \ f", {self.get_category()}: {self.get_description()}" KEY_FIELDS = ["check_name", "category", "description"] def is_similar_to(self, other: "AnalysisDiagnostic") -> bool: # We consider two diagnostics similar only if at least one # of the key fields is the same in both diagnostics. return len(self.get_diffs(other)) != len(self.KEY_FIELDS) def get_diffs(self, other: "AnalysisDiagnostic") -> JSONDiff: return {field: (self._data[field], other._data[field]) for field in self.KEY_FIELDS if self._data[field] != other._data[field]} # Note, the data format is not an API and may change from one analyzer # version to another. def get_raw_data(self) -> Plist: return self._data def __eq__(self, other: object) -> bool: return hash(self) == hash(other) def __ne__(self, other: object) -> bool: return hash(self) != hash(other) def __hash__(self) -> int: return hash(self.get_issue_identifier()) class AnalysisRun: def __init__(self, info: SingleRunInfo): self.path = info.path self.root = info.root self.info = info self.reports: List[AnalysisReport] = [] # Cumulative list of all diagnostics from all the reports. self.diagnostics: List[AnalysisDiagnostic] = [] self.clang_version: Optional[str] = None self.raw_stats: List[JSON] = [] def get_clang_version(self) -> Optional[str]: return self.clang_version def read_single_file(self, path: str, delete_empty: bool): with open(path, "rb") as plist_file: data = plistlib.load(plist_file) if 'statistics' in data: self.raw_stats.append(json.loads(data['statistics'])) data.pop('statistics') # We want to retrieve the clang version even if there are no # reports. Assume that all reports were created using the same # clang version (this is always true and is more efficient). if 'clang_version' in data: if self.clang_version is None: self.clang_version = data.pop('clang_version') else: data.pop('clang_version') # Ignore/delete empty reports. if not data['files']: if delete_empty: os.remove(path) return # Extract the HTML reports, if they exists. if 'HTMLDiagnostics_files' in data['diagnostics'][0]: htmlFiles = [] for d in data['diagnostics']: # FIXME: Why is this named files, when does it have multiple # files? assert len(d['HTMLDiagnostics_files']) == 1 htmlFiles.append(d.pop('HTMLDiagnostics_files')[0]) else: htmlFiles = [None] * len(data['diagnostics']) report = AnalysisReport(self, data.pop('files')) diagnostics = [AnalysisDiagnostic(d, report, h) for d, h in zip(data.pop('diagnostics'), htmlFiles)] assert not data report.diagnostics.extend(diagnostics) self.reports.append(report) self.diagnostics.extend(diagnostics) class AnalysisReport: def __init__(self, run: AnalysisRun, files: List[str]): self.run = run self.files = files self.diagnostics: List[AnalysisDiagnostic] = [] def load_results(results: ResultsDirectory, delete_empty: bool = True, verbose_log: Optional[str] = None) -> AnalysisRun: """ Backwards compatibility API. """ return load_results_from_single_run(SingleRunInfo(results, verbose_log), delete_empty) def load_results_from_single_run(info: SingleRunInfo, delete_empty: bool = True) -> AnalysisRun: """ # Load results of the analyzes from a given output folder. # - info is the SingleRunInfo object # - delete_empty specifies if the empty plist files should be deleted """ path = info.path run = AnalysisRun(info) if os.path.isfile(path): run.read_single_file(path, delete_empty) else: for dirpath, dirnames, filenames in os.walk(path): for f in filenames: if not f.endswith('plist'): continue p = os.path.join(dirpath, f) run.read_single_file(p, delete_empty) return run def cmp_analysis_diagnostic(d): return d.get_issue_identifier() AnalysisDiagnosticPair = Tuple[AnalysisDiagnostic, AnalysisDiagnostic] class ComparisonResult: def __init__(self): self.present_in_both: List[AnalysisDiagnostic] = [] self.present_only_in_old: List[AnalysisDiagnostic] = [] self.present_only_in_new: List[AnalysisDiagnostic] = [] self.changed_between_new_and_old: List[AnalysisDiagnosticPair] = [] def add_common(self, issue: AnalysisDiagnostic): self.present_in_both.append(issue) def add_removed(self, issue: AnalysisDiagnostic): self.present_only_in_old.append(issue) def add_added(self, issue: AnalysisDiagnostic): self.present_only_in_new.append(issue) def add_changed(self, old_issue: AnalysisDiagnostic, new_issue: AnalysisDiagnostic): self.changed_between_new_and_old.append((old_issue, new_issue)) GroupedDiagnostics = DefaultDict[str, List[AnalysisDiagnostic]] def get_grouped_diagnostics(diagnostics: List[AnalysisDiagnostic] ) -> GroupedDiagnostics: result: GroupedDiagnostics = defaultdict(list) for diagnostic in diagnostics: result[diagnostic.get_location()].append(diagnostic) return result def compare_results(results_old: AnalysisRun, results_new: AnalysisRun, histogram: Optional[HistogramType] = None ) -> ComparisonResult: """ compare_results - Generate a relation from diagnostics in run A to diagnostics in run B. The result is the relation as a list of triples (a, b) where each element {a,b} is None or a matching element from the respective run """ res = ComparisonResult() # Map size_before -> size_after path_difference_data: List[float] = [] diags_old = get_grouped_diagnostics(results_old.diagnostics) diags_new = get_grouped_diagnostics(results_new.diagnostics) locations_old = set(diags_old.keys()) locations_new = set(diags_new.keys()) common_locations = locations_old & locations_new for location in common_locations: old = diags_old[location] new = diags_new[location] # Quadratic algorithms in this part are fine because 'old' and 'new' # are most commonly of size 1. common: Set[AnalysisDiagnostic] = set() for a in old: for b in new: if a.get_issue_identifier() == b.get_issue_identifier(): a_path_len = a.get_path_length() b_path_len = b.get_path_length() if a_path_len != b_path_len: if histogram == HistogramType.RELATIVE: path_difference_data.append( float(a_path_len) / b_path_len) elif histogram == HistogramType.LOG_RELATIVE: path_difference_data.append( log(float(a_path_len) / b_path_len)) elif histogram == HistogramType.ABSOLUTE: path_difference_data.append( a_path_len - b_path_len) res.add_common(b) common.add(a) old = filter_issues(old, common) new = filter_issues(new, common) common = set() for a in old: for b in new: if a.is_similar_to(b): res.add_changed(a, b) common.add(a) common.add(b) old = filter_issues(old, common) new = filter_issues(new, common) # Whatever is left in 'old' doesn't have a corresponding diagnostic # in 'new', so we need to mark it as 'removed'. for a in old: res.add_removed(a) # Whatever is left in 'new' doesn't have a corresponding diagnostic # in 'old', so we need to mark it as 'added'. for b in new: res.add_added(b) only_old_locations = locations_old - common_locations for location in only_old_locations: for a in diags_old[location]: # These locations have been found only in the old build, so we # need to mark all of therm as 'removed' res.add_removed(a) only_new_locations = locations_new - common_locations for location in only_new_locations: for b in diags_new[location]: # These locations have been found only in the new build, so we # need to mark all of therm as 'added' res.add_added(b) # FIXME: Add fuzzy matching. One simple and possible effective idea would # be to bin the diagnostics, print them in a normalized form (based solely # on the structure of the diagnostic), compute the diff, then use that as # the basis for matching. This has the nice property that we don't depend # in any way on the diagnostic format. if histogram: from matplotlib import pyplot pyplot.hist(path_difference_data, bins=100) pyplot.show() return res def filter_issues(origin: List[AnalysisDiagnostic], to_remove: Set[AnalysisDiagnostic]) \ -> List[AnalysisDiagnostic]: return [diag for diag in origin if diag not in to_remove] def compute_percentile(values: Sequence[T], percentile: float) -> T: """ Return computed percentile. """ return sorted(values)[int(round(percentile * len(values) + 0.5)) - 1] def derive_stats(results: AnalysisRun) -> Stats: # Assume all keys are the same in each statistics bucket. combined_data = defaultdict(list) # Collect data on paths length. for report in results.reports: for diagnostic in report.diagnostics: combined_data['PathsLength'].append(diagnostic.get_path_length()) for stat in results.raw_stats: for key, value in stat.items(): combined_data[str(key)].append(value) combined_stats: Stats = {} for key, values in combined_data.items(): combined_stats[key] = { "max": max(values), "min": min(values), "mean": sum(values) / len(values), "90th %tile": compute_percentile(values, 0.9), "95th %tile": compute_percentile(values, 0.95), "median": sorted(values)[len(values) // 2], "total": sum(values) } return combined_stats # TODO: compare_results decouples comparison from the output, we should # do it here as well def compare_stats(results_old: AnalysisRun, results_new: AnalysisRun, out: TextIO = sys.stdout): stats_old = derive_stats(results_old) stats_new = derive_stats(results_new) old_keys = set(stats_old.keys()) new_keys = set(stats_new.keys()) keys = sorted(old_keys & new_keys) for key in keys: out.write(f"{key}\n") nested_keys = sorted(set(stats_old[key]) & set(stats_new[key])) for nested_key in nested_keys: val_old = float(stats_old[key][nested_key]) val_new = float(stats_new[key][nested_key]) report = f"{val_old:.3f} -> {val_new:.3f}" # Only apply highlighting when writing to TTY and it's not Windows if out.isatty() and os.name != 'nt': if val_new != 0: ratio = (val_new - val_old) / val_new if ratio < -0.2: report = Colors.GREEN + report + Colors.CLEAR elif ratio > 0.2: report = Colors.RED + report + Colors.CLEAR out.write(f"\t {nested_key} {report}\n") removed_keys = old_keys - new_keys if removed_keys: out.write(f"REMOVED statistics: {removed_keys}\n") added_keys = new_keys - old_keys if added_keys: out.write(f"ADDED statistics: {added_keys}\n") out.write("\n") def dump_scan_build_results_diff(dir_old: ResultsDirectory, dir_new: ResultsDirectory, delete_empty: bool = True, out: TextIO = sys.stdout, show_stats: bool = False, stats_only: bool = False, histogram: Optional[HistogramType] = None, verbose_log: Optional[str] = None): """ Compare directories with analysis results and dump results. :param delete_empty: delete empty plist files :param out: buffer to dump comparison results to. :param show_stats: compare execution stats as well. :param stats_only: compare ONLY execution stats. :param histogram: optional histogram type to plot path differences. :param verbose_log: optional path to an additional log file. """ results_old = load_results(dir_old, delete_empty, verbose_log) results_new = load_results(dir_new, delete_empty, verbose_log) if show_stats or stats_only: compare_stats(results_old, results_new) if stats_only: return # Open the verbose log, if given. if verbose_log: aux_log: Optional[TextIO] = open(verbose_log, "w") else: aux_log = None diff = compare_results(results_old, results_new, histogram) found_diffs = 0 total_added = 0 total_removed = 0 total_modified = 0 for new in diff.present_only_in_new: out.write(f"ADDED: {new.get_readable_name()}\n\n") found_diffs += 1 total_added += 1 if aux_log: aux_log.write(f"('ADDED', {new.get_readable_name()}, " f"{new.get_html_report()})\n") for old in diff.present_only_in_old: out.write(f"REMOVED: {old.get_readable_name()}\n\n") found_diffs += 1 total_removed += 1 if aux_log: aux_log.write(f"('REMOVED', {old.get_readable_name()}, " f"{old.get_html_report()})\n") for old, new in diff.changed_between_new_and_old: out.write(f"MODIFIED: {old.get_readable_name()}\n") found_diffs += 1 total_modified += 1 diffs = old.get_diffs(new) str_diffs = [f" '{key}' changed: " f"'{old_value}' -> '{new_value}'" for key, (old_value, new_value) in diffs.items()] out.write(",\n".join(str_diffs) + "\n\n") if aux_log: aux_log.write(f"('MODIFIED', {old.get_readable_name()}, " f"{old.get_html_report()})\n") total_reports = len(results_new.diagnostics) out.write(f"TOTAL REPORTS: {total_reports}\n") out.write(f"TOTAL ADDED: {total_added}\n") out.write(f"TOTAL REMOVED: {total_removed}\n") out.write(f"TOTAL MODIFIED: {total_modified}\n") if aux_log: aux_log.write(f"('TOTAL NEW REPORTS', {total_reports})\n") aux_log.write(f"('TOTAL DIFFERENCES', {found_diffs})\n") aux_log.close() # TODO: change to NamedTuple return found_diffs, len(results_old.diagnostics), \ len(results_new.diagnostics) if __name__ == "__main__": print("CmpRuns.py should not be used on its own.") print("Please use 'SATest.py compare' instead") sys.exit(1)