eric: comparison DebugClients/Python/coverage/data.py

-:04e5dfbd3f3d
+:3586ebd9fac8
 # For details: https://bitbucket.org/ned/coveragepy/src/default/NOTICE.txt
 """Coverage data for coverage.py."""
 import glob
+import itertools
 import json
 import optparse
 import os
 import os.path
 import random
 import re
 import socket
-import sys
 from coverage import env
 from coverage.backward import iitems, string_class
 from coverage.debug import _TEST_NAME_FILE
 from coverage.files import PathAliases
-from coverage.misc import CoverageException, file_be_gone
+from coverage.misc import CoverageException, file_be_gone, isolate_module
+os = isolate_module(os)
 class CoverageData(object):
 """Manages collected coverage data, including file storage.
 If the file was executed, returns a list of integers, the line numbers
 executed in the file. The list is in no particular order.
 """
 if self._arcs is not None:
-if filename in self._arcs:
+arcs = self._arcs.get(filename)
-return [s for s, __ in self._arcs[filename] if s > 0]
+if arcs is not None:
+all_lines = itertools.chain.from_iterable(arcs)
+return list(set(l for l in all_lines if l > 0))
 elif self._lines is not None:
-if filename in self._lines:
+return self._lines.get(filename)
-return self._lines[filename]
 return None
 def arcs(self, filename):
 """Get the list of arcs executed for a file.
 data = self._read_raw_data(file_obj)
 self._lines = self._arcs = None
 if 'lines' in data:
-self._lines = dict(
+self._lines = data['lines']
-(fname.encode(sys.getfilesystemencoding()), linenos)
-for fname, linenos in iitems(data['lines'])
-)
 if 'arcs' in data:
 self._arcs = dict(
-(fname.encode(sys.getfilesystemencoding()),
+(fname, [tuple(pair) for pair in arcs])
-[tuple(pair) for pair in arcs])
 for fname, arcs in iitems(data['arcs'])
 )
 self._file_tracers = data.get('file_tracers', {})
 self._runs = data.get('runs', [])
 def read_file(self, filename):
 """Read the coverage data from `filename` into this object."""
 if self._debug and self._debug.should('dataio'):
 self._debug.write("Reading data from %r" % (filename,))
-with self._open_for_reading(filename) as f:
+try:
-self.read_fileobj(f)
+with self._open_for_reading(filename) as f:
+self.read_fileobj(f)
+except Exception as exc:
+raise CoverageException(
+"Couldn't read data from '%s': %s: %s" % (
+filename, exc.__class__.__name__, exc,
+)
+)
 _GO_AWAY = "!coverage.py: This is a private format, don't read it directly!"
 @classmethod
 def _open_for_reading(cls, filename):
 # Create the file data.
 file_data = {}
 if self._has_arcs():
-file_data['arcs'] = dict(
+file_data['arcs'] = self._arcs
-(fname.decode(sys.getfilesystemencoding()),
-[tuple(pair) for pair in self._arcs])
-for fname, arcs in iitems(data['arcs'])
-)
 if self._has_lines():
-file_data['lines'] = dict(
+file_data['lines'] = self._lines
-(fname.decode(sys.getfilesystemencoding()), linenos)
-for fname, linenos in iitems(self._lines)
-)
 if self._file_tracers:
 file_data['file_tracers'] = self._file_tracers
 if self._runs:
 assert isinstance(key, string_class), "Key in _runs shouldn't be %r" % (key,)
 def add_to_hash(self, filename, hasher):
 """Contribute `filename`'s data to the `hasher`.
-`hasher` is a :class:`coverage.misc.Hasher` instance to be updated with
+`hasher` is a `coverage.misc.Hasher` instance to be updated with
 the file's data.  It should only get the results data, not the run
 data.
 """
 if self._has_arcs():
 class CoverageDataFiles(object):
 """Manage the use of coverage data files."""
-def __init__(self, basename=None):
+def __init__(self, basename=None, warn=None):
 """Create a CoverageDataFiles to manage data files.
+`warn` is the warning function to use.
 `basename` is the name of the file to use for storing data.
 """
+self.warn = warn
 # Construct the file name that will be used for data storage.
 self.filename = os.path.abspath(basename or ".coverage")
 def erase(self, parallel=False):
 """Erase the data from the file storage.
 `self.filename` plus dot as a prefix, and those files are combined.
 If `data_paths` is not provided, then the directory portion of
 `self.filename` is used as the directory to search for data files.
-Every data file found and combined is then deleted from disk.
+Every data file found and combined is then deleted from disk. If a file
+cannot be read, a warning will be issued, and the file will not be
+deleted.
 """
 # Because of the os.path.abspath in the constructor, data_dir will
 # never be an empty string.
 data_dir, local = os.path.split(self.filename)
 else:
 raise CoverageException("Couldn't combine from non-existent path '%s'" % (p,))
 for f in files_to_combine:
 new_data = CoverageData()
-new_data.read_file(f)
+try:
-data.update(new_data, aliases=aliases)
+new_data.read_file(f)
-file_be_gone(f)
+except CoverageException as exc:
+if self.warn:
+# The CoverageException has the file name in it, so just
+# use the message as the warning.
+self.warn(str(exc))
+else:
+data.update(new_data, aliases=aliases)
+file_be_gone(f)
 def canonicalize_json_data(data):
 """Canonicalize our JSON data so it can be compared."""
 for fname, lines in iitems(data.get('lines', {})):
 canonicalize_json_data(data)
 print(pretty_data(data))
 if __name__ == '__main__':
+import sys
 debug_main(sys.argv[1:])
 #
 # eflag: FileType = Python2

Mercurial Repositories > eric / file comparison

comparison: DebugClients/Python/coverage/data.py

DebugClients/Python/coverage/data.py