--- a/eric7/DebugClients/Python/coverage/collector.py Fri Nov 19 19:28:47 2021 +0100
+++ b/eric7/DebugClients/Python/coverage/collector.py Sat Nov 20 16:47:38 2021 +0100
@@ -7,10 +7,10 @@
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)
@@ -21,7 +21,7 @@
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
@@ -33,7 +33,7 @@
CTracer = None
-class Collector(object):
+class Collector:
"""Collects trace data.
Creates a Tracer object for each thread, since they track stack
@@ -116,7 +116,7 @@
# 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:
@@ -136,13 +136,13 @@
import threading
self.threading = threading
else:
- raise CoverageException("Don't understand concurrency=%s" % concurrency)
- except ImportError:
+ raise CoverageException(f"Don't understand concurrency={concurrency}")
+ 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()
@@ -157,12 +157,14 @@
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."""
@@ -244,7 +246,7 @@
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(),
)
)
@@ -318,8 +320,8 @@
(frame, event, arg), lineno = args
try:
fn(frame, event, arg, lineno=lineno)
- except TypeError:
- raise Exception("fullcoverage must be run with the C trace function.")
+ except TypeError as ex:
+ raise Exception("fullcoverage must be run with the C trace function.") from ex
# Install our installation tracer in threading, to jump-start other
# threads.
@@ -332,9 +334,9 @@
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()
@@ -352,8 +354,8 @@
stats = tracer.get_stats()
if stats:
print("\nCoverage.py tracer stats:")
- for k in sorted(stats.keys()):
- print("%20s: %s" % (k, stats[k]))
+ for k in human_sorted(stats.keys()):
+ print(f"{k:>20}: {stats[k]}")
if self.threading:
self.threading.settrace(None)
@@ -390,7 +392,7 @@
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
@@ -404,22 +406,22 @@
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)
- except RuntimeError as ex:
+ items = list(d.items())
+ 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."""
@@ -437,7 +439,25 @@
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))
