"""Coverage data for Coverage.""" import os from coverage.backward import iitems, pickle, sorted # pylint: disable=W0622 from coverage.files import PathAliases from coverage.misc import file_be_gone class CoverageData(object): """Manages collected coverage data, including file storage. The data file format is a pickled dict, with these keys: * collector: a string identifying the collecting software * lines: a dict mapping filenames to sorted lists of line numbers executed: { 'file1': [17,23,45], 'file2': [1,2,3], ... } * arcs: a dict mapping filenames to sorted lists of line number pairs: { 'file1': [(17,23), (17,25), (25,26)], ... } """ def __init__(self, basename=None, collector=None, debug=None): """Create a CoverageData. `basename` is the name of the file to use for storing data. `collector` is a string describing the coverage measurement software. `debug` is a `DebugControl` object for writing debug messages. """ self.collector = collector or 'unknown' self.debug = debug self.use_file = True # Construct the filename that will be used for data file storage, if we # ever do any file storage. self.filename = basename or ".coverage" self.filename = os.path.abspath(self.filename) # A map from canonical Python source file name to a dictionary in # which there's an entry for each line number that has been # executed: # # { # 'filename1.py': { 12: None, 47: None, ... }, # ... # } # self.lines = {} # A map from canonical Python source file name to a dictionary with an # entry for each pair of line numbers forming an arc: # # { # 'filename1.py': { (12,14): None, (47,48): None, ... }, # ... # } # self.arcs = {} def usefile(self, use_file=True): """Set whether or not to use a disk file for data.""" self.use_file = use_file def read(self): """Read coverage data from the coverage data file (if it exists).""" if self.use_file: self.lines, self.arcs = self._read_file(self.filename) else: self.lines, self.arcs = {}, {} def write(self, suffix=None): """Write the collected coverage data to a file. `suffix` is a suffix to append to the base file name. This can be used for multiple or parallel execution, so that many coverage data files can exist simultaneously. A dot will be used to join the base name and the suffix. """ if self.use_file: filename = self.filename if suffix: filename += "." + suffix self.write_file(filename) def erase(self): """Erase the data, both in this object, and from its file storage.""" if self.use_file: if self.filename: file_be_gone(self.filename) self.lines = {} self.arcs = {} def line_data(self): """Return the map from filenames to lists of line numbers executed.""" return dict( [(f, sorted(lmap.keys())) for f, lmap in iitems(self.lines)] ) def arc_data(self): """Return the map from filenames to lists of line number pairs.""" return dict( [(f, sorted(amap.keys())) for f, amap in iitems(self.arcs)] ) def write_file(self, filename): """Write the coverage data to `filename`.""" # Create the file data. data = {} data['lines'] = self.line_data() arcs = self.arc_data() if arcs: data['arcs'] = arcs if self.collector: data['collector'] = self.collector if self.debug and self.debug.should('dataio'): self.debug.write("Writing data to %r" % (filename,)) # Write the pickle to the file. fdata = open(filename, 'wb') try: pickle.dump(data, fdata, 2) finally: fdata.close() def read_file(self, filename): """Read the coverage data from `filename`.""" self.lines, self.arcs = self._read_file(filename) def raw_data(self, filename): """Return the raw pickled data from `filename`.""" if self.debug and self.debug.should('dataio'): self.debug.write("Reading data from %r" % (filename,)) fdata = open(filename, 'rb') try: data = pickle.load(fdata) finally: fdata.close() return data def _read_file(self, filename): """Return the stored coverage data from the given file. Returns two values, suitable for assigning to `self.lines` and `self.arcs`. """ lines = {} arcs = {} try: data = self.raw_data(filename) if isinstance(data, dict): # Unpack the 'lines' item. lines = dict([ (f, dict.fromkeys(linenos, None)) for f, linenos in iitems(data.get('lines', {})) ]) # Unpack the 'arcs' item. arcs = dict([ (f, dict.fromkeys(arcpairs, None)) for f, arcpairs in iitems(data.get('arcs', {})) ]) except Exception: pass return lines, arcs def combine_parallel_data(self, aliases=None): """Combine a number of data files together. Treat `self.filename` as a file prefix, and combine the data from all of the data files starting with that prefix plus a dot. If `aliases` is provided, it's a `PathAliases` object that is used to re-map paths to match the local machine's. """ aliases = aliases or PathAliases() data_dir, local = os.path.split(self.filename) localdot = local + '.' for f in os.listdir(data_dir or '.'): if f.startswith(localdot): full_path = os.path.join(data_dir, f) new_lines, new_arcs = self._read_file(full_path) for filename, file_data in iitems(new_lines): filename = aliases.map(filename) self.lines.setdefault(filename, {}).update(file_data) for filename, file_data in iitems(new_arcs): filename = aliases.map(filename) self.arcs.setdefault(filename, {}).update(file_data) if f != local: os.remove(full_path) def add_line_data(self, line_data): """Add executed line data. `line_data` is { filename: { lineno: None, ... }, ...} """ for filename, linenos in iitems(line_data): self.lines.setdefault(filename, {}).update(linenos) def add_arc_data(self, arc_data): """Add measured arc data. `arc_data` is { filename: { (l1,l2): None, ... }, ...} """ for filename, arcs in iitems(arc_data): self.arcs.setdefault(filename, {}).update(arcs) def touch_file(self, filename): """Ensure that `filename` appears in the data, empty if needed.""" self.lines.setdefault(filename, {}) def measured_files(self): """A list of all files that had been measured.""" return list(self.lines.keys()) def executed_lines(self, filename): """A map containing all the line numbers executed in `filename`. If `filename` hasn't been collected at all (because it wasn't executed) then return an empty map. """ return self.lines.get(filename) or {} def executed_arcs(self, filename): """A map containing all the arcs executed in `filename`.""" return self.arcs.get(filename) or {} def add_to_hash(self, filename, hasher): """Contribute `filename`'s data to the Md5Hash `hasher`.""" hasher.update(self.executed_lines(filename)) hasher.update(self.executed_arcs(filename)) def summary(self, fullpath=False): """Return a dict summarizing the coverage data. Keys are based on the filenames, and values are the number of executed lines. If `fullpath` is true, then the keys are the full pathnames of the files, otherwise they are the basenames of the files. """ summ = {} if fullpath: filename_fn = lambda f: f else: filename_fn = os.path.basename for filename, lines in iitems(self.lines): summ[filename_fn(filename)] = len(lines) return summ def has_arcs(self): """Does this data have arcs?""" return bool(self.arcs) if __name__ == '__main__': # Ad-hoc: show the raw data in a data file. import pprint, sys covdata = CoverageData() if sys.argv[1:]: fname = sys.argv[1] else: fname = covdata.filename pprint.pprint(covdata.raw_data(fname))