eric: comparison eric7/DebugClients/Python/coverage/collector.py

-:d728227e8ebb
+:0802ae193343
 import os
 import sys
 from coverage import env
-from coverage.backward import litems, range     # pylint: disable=redefined-builtin
 from coverage.debug import short_stack
 from coverage.disposition import FileDisposition
-from coverage.misc import CoverageException, isolate_module
+from coverage.exceptions import CoverageException
+from coverage.misc import human_sorted, isolate_module
 from coverage.pytracer import PyTracer
 os = isolate_module(os)
 try:
 # Use the C extension code when we can, for speed.
 from coverage.tracer import CTracer, CFileDisposition
 except ImportError:
 # Couldn't import the C extension, maybe it isn't built.
-if os.getenv('COVERAGE_TEST_TRACER') == 'c':
+if os.getenv('COVERAGE_TEST_TRACER') == 'c':        # pragma: part covered
 # During testing, we use the COVERAGE_TEST_TRACER environment variable
 # to indicate that we've fiddled with the environment to test this
 # fallback code.  If we thought we had a C tracer, but couldn't import
 # it, then exit quickly and clearly instead of dribbling confusing
 # errors. I'm using sys.exit here instead of an exception because an
 sys.stderr.write("*** COVERAGE_TEST_TRACER is 'c' but can't import CTracer!\n")
 sys.exit(1)
 CTracer = None
-class Collector(object):
+class Collector:
 """Collects trace data.
 Creates a Tracer object for each thread, since they track stack
 information.  Each Tracer points to the same shared data, contributing
 traced data points.
 self.mapped_file_cache = {}
 # We can handle a few concurrency options here, but only one at a time.
 these_concurrencies = self.SUPPORTED_CONCURRENCIES.intersection(concurrency)
 if len(these_concurrencies) > 1:
-raise CoverageException("Conflicting concurrency settings: %s" % concurrency)
+raise CoverageException(f"Conflicting concurrency settings: {concurrency}")
 self.concurrency = these_concurrencies.pop() if these_concurrencies else ''
 try:
 if self.concurrency == "greenlet":
 import greenlet
 # it's imported early, and the program being measured uses
 # gevent, then gevent's monkey-patching won't work properly.
 import threading
 self.threading = threading
 else:
-raise CoverageException("Don't understand concurrency=%s" % concurrency)
+raise CoverageException(f"Don't understand concurrency={concurrency}")
-except ImportError:
+except ImportError as ex:
 raise CoverageException(
-"Couldn't trace with concurrency=%s, the module isn't installed." % (
+"Couldn't trace with concurrency={}, the module isn't installed.".format(
 self.concurrency,
 )
-)
+) from ex
 self.reset()
 if timid:
 # Being timid: use the simple Python trace function.
 self._trace_class = CTracer or PyTracer
 if self._trace_class is CTracer:
 self.file_disposition_class = CFileDisposition
 self.supports_plugins = True
+self.packed_arcs = True
 else:
 self.file_disposition_class = FileDisposition
 self.supports_plugins = False
+self.packed_arcs = False
 def __repr__(self):
-return "<Collector at 0x%x: %s>" % (id(self), self.tracer_name())
+return f"<Collector at 0x{id(self):x}: {self.tracer_name()}>"
 def use_data(self, covdata, context):
 """Use `covdata` for recording data."""
 self.covdata = covdata
 self.static_context = context
 if hasattr(tracer, 'concur_id_func'):
 tracer.concur_id_func = self.concur_id_func
 elif self.concur_id_func:
 raise CoverageException(
-"Can't support concurrency=%s with %s, only threads are supported" % (
+"Can't support concurrency={} with {}, only threads are supported".format(
 self.concurrency, self.tracer_name(),
 )
 )
 if hasattr(tracer, 'file_tracers'):
 # Replay all the events from fullcoverage into the new trace function.
 for args in traces0:
 (frame, event, arg), lineno = args
 try:
 fn(frame, event, arg, lineno=lineno)
-except TypeError:
+except TypeError as ex:
-raise Exception("fullcoverage must be run with the C trace function.")
+raise Exception("fullcoverage must be run with the C trace function.") from ex
 # Install our installation tracer in threading, to jump-start other
 # threads.
 if self.threading:
 self.threading.settrace(self._installation_trace)
 """Stop collecting trace information."""
 assert self._collectors
 if self._collectors[-1] is not self:
 print("self._collectors:")
 for c in self._collectors:
-print("  {!r}\n{}".format(c, c.origin))
+print(f"  {c!r}\n{c.origin}")
 assert self._collectors[-1] is self, (
-"Expected current collector to be %r, but it's %r" % (self, self._collectors[-1])
+f"Expected current collector to be {self!r}, but it's {self._collectors[-1]!r}"
 )
 self.pause()
 # Remove this Collector from the stack, and resume the one underneath
 for tracer in self.tracers:
 tracer.stop()
 stats = tracer.get_stats()
 if stats:
 print("\nCoverage.py tracer stats:")
-for k in sorted(stats.keys()):
+for k in human_sorted(stats.keys()):
-print("%20s: %s" % (k, stats[k]))
+print(f"{k:>20}: {stats[k]}")
 if self.threading:
 self.threading.settrace(None)
 def resume(self):
 """Resume tracing after a `pause`."""
 def disable_plugin(self, disposition):
 """Disable the plugin mentioned in `disposition`."""
 file_tracer = disposition.file_tracer
 plugin = file_tracer._coverage_plugin
 plugin_name = plugin._coverage_plugin_name
-self.warn("Disabling plug-in {!r} due to previous exception".format(plugin_name))
+self.warn(f"Disabling plug-in {plugin_name!r} due to previous exception")
 plugin._coverage_enabled = False
 disposition.trace = False
 def cached_mapped_file(self, filename):
 """A locally cached version of file names mapped through file_mapper."""
 except KeyError:
 return self.mapped_file_cache.setdefault(key, self.file_mapper(filename))
 def mapped_file_dict(self, d):
 """Return a dict like d, but with keys modified by file_mapper."""
-# The call to litems() ensures that the GIL protects the dictionary
+# The call to list(items()) ensures that the GIL protects the dictionary
 # iterator against concurrent modifications by tracers running
 # in other threads. We try three times in case of concurrent
 # access, hoping to get a clean copy.
 runtime_err = None
-for _ in range(3):
+for _ in range(3):                      # pragma: part covered
 try:
-items = litems(d)
+items = list(d.items())
-except RuntimeError as ex:
+except RuntimeError as ex:          # pragma: cant happen
 runtime_err = ex
 else:
 break
 else:
-raise runtime_err
+raise runtime_err                   # pragma: cant happen
-return dict((self.cached_mapped_file(k), v) for k, v in items if v)
+return {self.cached_mapped_file(k): v for k, v in items if v}
 def plugin_was_disabled(self, plugin):
 """Record that `plugin` was disabled during the run."""
 self.disabled_plugins.add(plugin._coverage_plugin_name)
 """
 if not self._activity():
 return False
 if self.branch:
-self.covdata.add_arcs(self.mapped_file_dict(self.data))
+if self.packed_arcs:
+# Unpack the line number pairs packed into integers.  See
+# tracer.c:CTracer_record_pair for the C code that creates
+# these packed ints.
+data = {}
+for fname, packeds in self.data.items():
+tuples = []
+for packed in packeds:
+l1 = packed & 0xFFFFF
+l2 = (packed & (0xFFFFF << 20)) >> 20
+if packed & (1 << 40):
+l1 *= -1
+if packed & (1 << 41):
+l2 *= -1
+tuples.append((l1, l2))
+data[fname] = tuples
+else:
+data = self.data
+self.covdata.add_arcs(self.mapped_file_dict(data))
 else:
 self.covdata.add_lines(self.mapped_file_dict(self.data))
 file_tracers = {
 k: v for k, v in self.file_tracers.items()

Mercurial Repositories > eric / file comparison

comparison: eric7/DebugClients/Python/coverage/collector.py

eric7/DebugClients/Python/coverage/collector.py