You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
279 lines
9.0 KiB
Python
279 lines
9.0 KiB
Python
"""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))
|