diff -r dc171b1d8261 -r 362cd1b6f81a eric6/DebugClients/Python/coverage/control.py
--- a/eric6/DebugClients/Python/coverage/control.py Wed Feb 19 19:38:36 2020 +0100
+++ b/eric6/DebugClients/Python/coverage/control.py Sat Feb 22 14:27:42 2020 +0100
@@ -1,36 +1,35 @@
# Licensed under the Apache License: http://www.apache.org/licenses/LICENSE-2.0
-# For details: https://bitbucket.org/ned/coveragepy/src/default/NOTICE.txt
+# For details: https://github.com/nedbat/coveragepy/blob/master/NOTICE.txt
"""Core control stuff for coverage.py."""
-
import atexit
-import inspect
-import itertools
+import contextlib
import os
+import os.path
import platform
-import re
import sys
import time
-import traceback
from coverage import env
from coverage.annotate import AnnotateReporter
from coverage.backward import string_class, iitems
-from coverage.collector import Collector
+from coverage.collector import Collector, CTracer
from coverage.config import read_coverage_config
-from coverage.data import CoverageData, CoverageDataFiles
-from coverage.debug import DebugControl, write_formatted_info
-from coverage.files import TreeMatcher, FnmatchMatcher
-from coverage.files import PathAliases, find_python_files, prep_patterns
-from coverage.files import canonical_filename, set_relative_directory
-from coverage.files import ModuleMatcher, abs_file
+from coverage.context import should_start_context_test_function, combine_context_switchers
+from coverage.data import CoverageData, combine_parallel_data
+from coverage.debug import DebugControl, short_stack, write_formatted_info
+from coverage.disposition import disposition_debug_msg
+from coverage.files import PathAliases, abs_file, relative_filename, set_relative_directory
from coverage.html import HtmlReporter
+from coverage.inorout import InOrOut
+from coverage.jsonreport import JsonReporter
from coverage.misc import CoverageException, bool_or_none, join_regex
-from coverage.misc import file_be_gone, isolate_module
+from coverage.misc import DefaultValue, ensure_dir_for_file, isolate_module
from coverage.plugin import FileReporter
from coverage.plugin_support import Plugins
-from coverage.python import PythonFileReporter, source_for_file
+from coverage.python import PythonFileReporter
+from coverage.report import render_report
from coverage.results import Analysis, Numbers
from coverage.summary import SummaryReporter
from coverage.xmlreport import XmlReporter
@@ -43,22 +42,23 @@
os = isolate_module(os)
-# Pypy has some unusual stuff in the "stdlib". Consider those locations
-# when deciding where the stdlib is. These modules are not used for anything,
-# they are modules importable from the pypy lib directories, so that we can
-# find those directories.
-_structseq = _pypy_irc_topic = None
-if env.PYPY:
+@contextlib.contextmanager
+def override_config(cov, **kwargs):
+ """Temporarily tweak the configuration of `cov`.
+
+ The arguments are applied to `cov.config` with the `from_args` method.
+ At the end of the with-statement, the old configuration is restored.
+ """
+ original_config = cov.config
+ cov.config = cov.config.copy()
try:
- import _structseq
- except ImportError:
- pass
+ cov.config.from_args(**kwargs)
+ yield
+ finally:
+ cov.config = original_config
- try:
- import _pypy_irc_topic
- except ImportError:
- pass
+_DEFAULT_DATAFILE = DefaultValue("MISSING")
class Coverage(object):
"""Programmatic access to coverage.py.
@@ -73,18 +73,45 @@
cov.stop()
cov.html_report(directory='covhtml')
+ Note: in keeping with Python custom, names starting with underscore are
+ not part of the public API. They might stop working at any point. Please
+ limit yourself to documented methods to avoid problems.
+
"""
+
+ # The stack of started Coverage instances.
+ _instances = []
+
+ @classmethod
+ def current(cls):
+ """Get the latest started `Coverage` instance, if any.
+
+ Returns: a `Coverage` instance, or None.
+
+ .. versionadded:: 5.0
+
+ """
+ if cls._instances:
+ return cls._instances[-1]
+ else:
+ return None
+
def __init__(
- self, data_file=None, data_suffix=None, cover_pylib=None,
+ self, data_file=_DEFAULT_DATAFILE, data_suffix=None, cover_pylib=None,
auto_data=False, timid=None, branch=None, config_file=True,
source=None, omit=None, include=None, debug=None,
- concurrency=None,
+ concurrency=None, check_preimported=False, context=None,
):
"""
- `data_file` is the base name of the data file to use, defaulting to
- ".coverage". `data_suffix` is appended (with a dot) to `data_file` to
- create the final file name. If `data_suffix` is simply True, then a
- suffix is created with the machine and process identity included.
+ Many of these arguments duplicate and override values that can be
+ provided in a configuration file. Parameters that are missing here
+ will use values from the config file.
+
+ `data_file` is the base name of the data file to use. The config value
+ defaults to ".coverage". None can be provided to prevent writing a data
+ file. `data_suffix` is appended (with a dot) to `data_file` to create
+ the final file name. If `data_suffix` is simply True, then a suffix is
+ created with the machine and process identity included.
`cover_pylib` is a boolean determining whether Python code installed
with the Python interpreter is measured. This includes the Python
@@ -132,58 +159,73 @@
"eventlet", "gevent", "multiprocessing", or "thread" (the default).
This can also be a list of these strings.
+ If `check_preimported` is true, then when coverage is started, the
+ already-imported files will be checked to see if they should be
+ measured by coverage. Importing measured files before coverage is
+ started can mean that code is missed.
+
+ `context` is a string to use as the :ref:`static context
+ <static_contexts>` label for collected data.
+
.. versionadded:: 4.0
The `concurrency` parameter.
.. versionadded:: 4.2
The `concurrency` parameter can now be a list of strings.
+ .. versionadded:: 5.0
+ The `check_preimported` and `context` parameters.
+
"""
+ # data_file=None means no disk file at all. data_file missing means
+ # use the value from the config file.
+ self._no_disk = data_file is None
+ if data_file is _DEFAULT_DATAFILE:
+ data_file = None
+
# Build our configuration from a number of sources.
- self.config_file, self.config = read_coverage_config(
+ self.config = read_coverage_config(
config_file=config_file,
data_file=data_file, cover_pylib=cover_pylib, timid=timid,
branch=branch, parallel=bool_or_none(data_suffix),
source=source, run_omit=omit, run_include=include, debug=debug,
report_omit=omit, report_include=include,
- concurrency=concurrency,
+ concurrency=concurrency, context=context,
)
# This is injectable by tests.
self._debug_file = None
self._auto_load = self._auto_save = auto_data
- self._data_suffix = data_suffix
-
- # The matchers for _should_trace.
- self.source_match = None
- self.source_pkgs_match = None
- self.pylib_match = self.cover_match = None
- self.include_match = self.omit_match = None
+ self._data_suffix_specified = data_suffix
# Is it ok for no data to be collected?
self._warn_no_data = True
self._warn_unimported_source = True
+ self._warn_preimported_source = check_preimported
+ self._no_warn_slugs = None
# A record of all the warnings that have been issued.
self._warnings = []
# Other instance attributes, set later.
- self.omit = self.include = self.source = None
- self.source_pkgs_unmatched = None
- self.source_pkgs = None
- self.data = self.data_files = self.collector = None
- self.plugins = None
- self.pylib_paths = self.cover_paths = None
- self.data_suffix = self.run_suffix = None
+ self._data = self._collector = None
+ self._plugins = None
+ self._inorout = None
+ self._inorout_class = InOrOut
+ self._data_suffix = self._run_suffix = None
self._exclude_re = None
- self.debug = None
+ self._debug = None
+ self._file_mapper = None
# State machine variables:
# Have we initialized everything?
self._inited = False
+ self._inited_for_start = False
# Have we started collecting and not stopped it?
self._started = False
+ # Should we write the debug output?
+ self._should_write_debug = True
# If we have sub-process measurement happening automatically, then we
# want any explicit creation of a Coverage object to mean, this process
@@ -209,378 +251,61 @@
# Create and configure the debugging controller. COVERAGE_DEBUG_FILE
# is an environment variable, the name of a file to append debug logs
# to.
- if self._debug_file is None:
- debug_file_name = os.environ.get("COVERAGE_DEBUG_FILE")
- if debug_file_name:
- self._debug_file = open(debug_file_name, "a")
- else:
- self._debug_file = sys.stderr
- self.debug = DebugControl(self.config.debug, self._debug_file)
+ self._debug = DebugControl(self.config.debug, self._debug_file)
+
+ if "multiprocessing" in (self.config.concurrency or ()):
+ # Multi-processing uses parallel for the subprocesses, so also use
+ # it for the main process.
+ self.config.parallel = True
# _exclude_re is a dict that maps exclusion list names to compiled regexes.
self._exclude_re = {}
set_relative_directory()
+ self._file_mapper = relative_filename if self.config.relative_files else abs_file
# Load plugins
- self.plugins = Plugins.load_plugins(self.config.plugins, self.config, self.debug)
+ self._plugins = Plugins.load_plugins(self.config.plugins, self.config, self._debug)
# Run configuring plugins.
- for plugin in self.plugins.configurers:
+ for plugin in self._plugins.configurers:
# We need an object with set_option and get_option. Either self or
# self.config will do. Choosing randomly stops people from doing
# other things with those objects, against the public API. Yes,
# this is a bit childish. :)
plugin.configure([self, self.config][int(time.time()) % 2])
- # The source argument can be directories or package names.
- self.source = []
- self.source_pkgs = []
- for src in self.config.source or []:
- if os.path.isdir(src):
- self.source.append(canonical_filename(src))
- else:
- self.source_pkgs.append(src)
- self.source_pkgs_unmatched = self.source_pkgs[:]
-
- self.omit = prep_patterns(self.config.run_omit)
- self.include = prep_patterns(self.config.run_include)
-
- concurrency = self.config.concurrency or []
- if "multiprocessing" in concurrency:
- if not patch_multiprocessing:
- raise CoverageException( # pragma: only jython
- "multiprocessing is not supported on this Python"
- )
- patch_multiprocessing(rcfile=self.config_file)
- # Multi-processing uses parallel for the subprocesses, so also use
- # it for the main process.
- self.config.parallel = True
-
- self.collector = Collector(
- should_trace=self._should_trace,
- check_include=self._check_include_omit_etc,
- timid=self.config.timid,
- branch=self.config.branch,
- warn=self._warn,
- concurrency=concurrency,
- )
-
- # Early warning if we aren't going to be able to support plugins.
- if self.plugins.file_tracers and not self.collector.supports_plugins:
- self._warn(
- "Plugin file tracers (%s) aren't supported with %s" % (
- ", ".join(
- plugin._coverage_plugin_name
- for plugin in self.plugins.file_tracers
- ),
- self.collector.tracer_name(),
- )
- )
- for plugin in self.plugins.file_tracers:
- plugin._coverage_enabled = False
-
- # Suffixes are a bit tricky. We want to use the data suffix only when
- # collecting data, not when combining data. So we save it as
- # `self.run_suffix` now, and promote it to `self.data_suffix` if we
- # find that we are collecting data later.
- if self._data_suffix or self.config.parallel:
- if not isinstance(self._data_suffix, string_class):
- # if data_suffix=True, use .machinename.pid.random
- self._data_suffix = True
- else:
- self._data_suffix = None
- self.data_suffix = None
- self.run_suffix = self._data_suffix
+ def _post_init(self):
+ """Stuff to do after everything is initialized."""
+ if self._should_write_debug:
+ self._should_write_debug = False
+ self._write_startup_debug()
- # Create the data file. We do this at construction time so that the
- # data file will be written into the directory where the process
- # started rather than wherever the process eventually chdir'd to.
- self.data = CoverageData(debug=self.debug)
- self.data_files = CoverageDataFiles(
- basename=self.config.data_file, warn=self._warn, debug=self.debug,
- )
-
- # The directories for files considered "installed with the interpreter".
- self.pylib_paths = set()
- if not self.config.cover_pylib:
- # Look at where some standard modules are located. That's the
- # indication for "installed with the interpreter". In some
- # environments (virtualenv, for example), these modules may be
- # spread across a few locations. Look at all the candidate modules
- # we've imported, and take all the different ones.
- for m in (atexit, inspect, os, platform, _pypy_irc_topic, re, _structseq, traceback):
- if m is not None and hasattr(m, "__file__"):
- self.pylib_paths.add(self._canonical_path(m, directory=True))
-
- if _structseq and not hasattr(_structseq, '__file__'):
- # PyPy 2.4 has no __file__ in the builtin modules, but the code
- # objects still have the file names. So dig into one to find
- # the path to exclude.
- structseq_new = _structseq.structseq_new
- try:
- structseq_file = structseq_new.func_code.co_filename
- except AttributeError:
- structseq_file = structseq_new.__code__.co_filename
- self.pylib_paths.add(self._canonical_path(structseq_file))
-
- # To avoid tracing the coverage.py code itself, we skip anything
- # located where we are.
- self.cover_paths = [self._canonical_path(__file__, directory=True)]
- if env.TESTING:
- # Don't include our own test code.
- self.cover_paths.append(os.path.join(self.cover_paths[0], "tests"))
-
- # When testing, we use PyContracts, which should be considered
- # part of coverage.py, and it uses six. Exclude those directories
- # just as we exclude ourselves.
- import contracts
- import six
- for mod in [contracts, six]:
- self.cover_paths.append(self._canonical_path(mod))
-
- # Set the reporting precision.
- Numbers.set_precision(self.config.precision)
-
- atexit.register(self._atexit)
-
- # Create the matchers we need for _should_trace
- if self.source or self.source_pkgs:
- self.source_match = TreeMatcher(self.source)
- self.source_pkgs_match = ModuleMatcher(self.source_pkgs)
- else:
- if self.cover_paths:
- self.cover_match = TreeMatcher(self.cover_paths)
- if self.pylib_paths:
- self.pylib_match = TreeMatcher(self.pylib_paths)
- if self.include:
- self.include_match = FnmatchMatcher(self.include)
- if self.omit:
- self.omit_match = FnmatchMatcher(self.omit)
-
- # The user may want to debug things, show info if desired.
- self._write_startup_debug()
+ # '[run] _crash' will raise an exception if the value is close by in
+ # the call stack, for testing error handling.
+ if self.config._crash and self.config._crash in short_stack(limit=4):
+ raise Exception("Crashing because called by {}".format(self.config._crash))
def _write_startup_debug(self):
"""Write out debug info at startup if needed."""
wrote_any = False
- with self.debug.without_callers():
- if self.debug.should('config'):
+ with self._debug.without_callers():
+ if self._debug.should('config'):
config_info = sorted(self.config.__dict__.items())
- write_formatted_info(self.debug, "config", config_info)
+ config_info = [(k, v) for k, v in config_info if not k.startswith('_')]
+ write_formatted_info(self._debug, "config", config_info)
wrote_any = True
- if self.debug.should('sys'):
- write_formatted_info(self.debug, "sys", self.sys_info())
- for plugin in self.plugins:
+ if self._debug.should('sys'):
+ write_formatted_info(self._debug, "sys", self.sys_info())
+ for plugin in self._plugins:
header = "sys: " + plugin._coverage_plugin_name
info = plugin.sys_info()
- write_formatted_info(self.debug, header, info)
+ write_formatted_info(self._debug, header, info)
wrote_any = True
if wrote_any:
- write_formatted_info(self.debug, "end", ())
-
- def _canonical_path(self, morf, directory=False):
- """Return the canonical path of the module or file `morf`.
-
- If the module is a package, then return its directory. If it is a
- module, then return its file, unless `directory` is True, in which
- case return its enclosing directory.
-
- """
- morf_path = PythonFileReporter(morf, self).filename
- if morf_path.endswith("__init__.py") or directory:
- morf_path = os.path.split(morf_path)[0]
- return morf_path
-
- def _name_for_module(self, module_globals, filename):
- """Get the name of the module for a set of globals and file name.
-
- For configurability's sake, we allow __main__ modules to be matched by
- their importable name.
-
- If loaded via runpy (aka -m), we can usually recover the "original"
- full dotted module name, otherwise, we resort to interpreting the
- file name to get the module's name. In the case that the module name
- can't be determined, None is returned.
-
- """
- if module_globals is None: # pragma: only ironpython
- # IronPython doesn't provide globals: https://github.com/IronLanguages/main/issues/1296
- module_globals = {}
-
- dunder_name = module_globals.get('__name__', None)
-
- if isinstance(dunder_name, str) and dunder_name != '__main__':
- # This is the usual case: an imported module.
- return dunder_name
-
- loader = module_globals.get('__loader__', None)
- for attrname in ('fullname', 'name'): # attribute renamed in py3.2
- if hasattr(loader, attrname):
- fullname = getattr(loader, attrname)
- else:
- continue
-
- if isinstance(fullname, str) and fullname != '__main__':
- # Module loaded via: runpy -m
- return fullname
-
- # Script as first argument to Python command line.
- inspectedname = inspect.getmodulename(filename)
- if inspectedname is not None:
- return inspectedname
- else:
- return dunder_name
-
- def _should_trace_internal(self, filename, frame):
- """Decide whether to trace execution in `filename`, with a reason.
-
- This function is called from the trace function. As each new file name
- is encountered, this function determines whether it is traced or not.
-
- Returns a FileDisposition object.
-
- """
- original_filename = filename
- disp = _disposition_init(self.collector.file_disposition_class, filename)
-
- def nope(disp, reason):
- """Simple helper to make it easy to return NO."""
- disp.trace = False
- disp.reason = reason
- return disp
-
- # Compiled Python files have two file names: frame.f_code.co_filename is
- # the file name at the time the .pyc was compiled. The second name is
- # __file__, which is where the .pyc was actually loaded from. Since
- # .pyc files can be moved after compilation (for example, by being
- # installed), we look for __file__ in the frame and prefer it to the
- # co_filename value.
- dunder_file = frame.f_globals and frame.f_globals.get('__file__')
- if dunder_file:
- filename = source_for_file(dunder_file)
- if original_filename and not original_filename.startswith('<'):
- orig = os.path.basename(original_filename)
- if orig != os.path.basename(filename):
- # Files shouldn't be renamed when moved. This happens when
- # exec'ing code. If it seems like something is wrong with
- # the frame's file name, then just use the original.
- filename = original_filename
-
- if not filename:
- # Empty string is pretty useless.
- return nope(disp, "empty string isn't a file name")
-
- if filename.startswith('memory:'):
- return nope(disp, "memory isn't traceable")
-
- if filename.startswith('<'):
- # Lots of non-file execution is represented with artificial
- # file names like "<string>", "<doctest readme.txt[0]>", or
- # "<exec_function>". Don't ever trace these executions, since we
- # can't do anything with the data later anyway.
- return nope(disp, "not a real file name")
-
- # pyexpat does a dumb thing, calling the trace function explicitly from
- # C code with a C file name.
- if re.search(r"[/\\]Modules[/\\]pyexpat.c", filename):
- return nope(disp, "pyexpat lies about itself")
-
- # Jython reports the .class file to the tracer, use the source file.
- if filename.endswith("$py.class"):
- filename = filename[:-9] + ".py"
-
- canonical = canonical_filename(filename)
- disp.canonical_filename = canonical
-
- # Try the plugins, see if they have an opinion about the file.
- plugin = None
- for plugin in self.plugins.file_tracers:
- if not plugin._coverage_enabled:
- continue
-
- try:
- file_tracer = plugin.file_tracer(canonical)
- if file_tracer is not None:
- file_tracer._coverage_plugin = plugin
- disp.trace = True
- disp.file_tracer = file_tracer
- if file_tracer.has_dynamic_source_filename():
- disp.has_dynamic_filename = True
- else:
- disp.source_filename = canonical_filename(
- file_tracer.source_filename()
- )
- break
- except Exception:
- self._warn(
- "Disabling plug-in %r due to an exception:" % (
- plugin._coverage_plugin_name
- )
- )
- traceback.print_exc()
- plugin._coverage_enabled = False
- continue
- else:
- # No plugin wanted it: it's Python.
- disp.trace = True
- disp.source_filename = canonical
-
- if not disp.has_dynamic_filename:
- if not disp.source_filename:
- raise CoverageException(
- "Plugin %r didn't set source_filename for %r" %
- (plugin, disp.original_filename)
- )
- reason = self._check_include_omit_etc_internal(
- disp.source_filename, frame,
- )
- if reason:
- nope(disp, reason)
-
- return disp
-
- def _check_include_omit_etc_internal(self, filename, frame):
- """Check a file name against the include, omit, etc, rules.
-
- Returns a string or None. String means, don't trace, and is the reason
- why. None means no reason found to not trace.
-
- """
- modulename = self._name_for_module(frame.f_globals, filename)
-
- # If the user specified source or include, then that's authoritative
- # about the outer bound of what to measure and we don't have to apply
- # any canned exclusions. If they didn't, then we have to exclude the
- # stdlib and coverage.py directories.
- if self.source_match:
- if self.source_pkgs_match.match(modulename):
- if modulename in self.source_pkgs_unmatched:
- self.source_pkgs_unmatched.remove(modulename)
- elif not self.source_match.match(filename):
- return "falls outside the --source trees"
- elif self.include_match:
- if not self.include_match.match(filename):
- return "falls outside the --include trees"
- else:
- # If we aren't supposed to trace installed code, then check if this
- # is near the Python standard library and skip it if so.
- if self.pylib_match and self.pylib_match.match(filename):
- return "is in the stdlib"
-
- # We exclude the coverage.py code itself, since a little of it
- # will be measured otherwise.
- if self.cover_match and self.cover_match.match(filename):
- return "is part of coverage.py"
-
- # Check the file against the omit pattern.
- if self.omit_match and self.omit_match.match(filename):
- return "is inside an --omit pattern"
-
- # No reason found to skip this file.
- return None
+ write_formatted_info(self._debug, "end", ())
def _should_trace(self, filename, frame):
"""Decide whether to trace execution in `filename`.
@@ -588,9 +313,9 @@
Calls `_should_trace_internal`, and returns the FileDisposition.
"""
- disp = self._should_trace_internal(filename, frame)
- if self.debug.should('trace'):
- self.debug.write(_disposition_debug_msg(disp))
+ disp = self._inorout.should_trace(filename, frame)
+ if self._debug.should('trace'):
+ self._debug.write(disposition_debug_msg(disp))
return disp
def _check_include_omit_etc(self, filename, frame):
@@ -599,32 +324,42 @@
Returns a boolean: True if the file should be traced, False if not.
"""
- reason = self._check_include_omit_etc_internal(filename, frame)
- if self.debug.should('trace'):
+ reason = self._inorout.check_include_omit_etc(filename, frame)
+ if self._debug.should('trace'):
if not reason:
msg = "Including %r" % (filename,)
else:
msg = "Not including %r: %s" % (filename, reason)
- self.debug.write(msg)
+ self._debug.write(msg)
return not reason
- def _warn(self, msg, slug=None):
+ def _warn(self, msg, slug=None, once=False):
"""Use `msg` as a warning.
For warning suppression, use `slug` as the shorthand.
+
+ If `once` is true, only show this warning once (determined by the
+ slug.)
+
"""
- if slug in self.config.disable_warnings:
+ if self._no_warn_slugs is None:
+ self._no_warn_slugs = list(self.config.disable_warnings)
+
+ if slug in self._no_warn_slugs:
# Don't issue the warning
return
self._warnings.append(msg)
if slug:
msg = "%s (%s)" % (msg, slug)
- if self.debug.should('pid'):
+ if self._debug.should('pid'):
msg = "[%d] %s" % (os.getpid(), msg)
sys.stderr.write("Coverage.py warning: %s\n" % msg)
+ if once:
+ self._no_warn_slugs.append(slug)
+
def get_option(self, option_name):
"""Get an option from the configuration.
@@ -664,17 +399,107 @@
"""
self.config.set_option(option_name, value)
- def use_cache(self, usecache):
- """Obsolete method."""
- self._init()
- if not usecache:
- self._warn("use_cache(False) is no longer supported.")
-
def load(self):
"""Load previously-collected coverage data from the data file."""
self._init()
- self.collector.reset()
- self.data_files.read(self.data)
+ if self._collector:
+ self._collector.reset()
+ should_skip = self.config.parallel and not os.path.exists(self.config.data_file)
+ if not should_skip:
+ self._init_data(suffix=None)
+ self._post_init()
+ if not should_skip:
+ self._data.read()
+
+ def _init_for_start(self):
+ """Initialization for start()"""
+ # Construct the collector.
+ concurrency = self.config.concurrency or ()
+ if "multiprocessing" in concurrency:
+ if not patch_multiprocessing:
+ raise CoverageException( # pragma: only jython
+ "multiprocessing is not supported on this Python"
+ )
+ patch_multiprocessing(rcfile=self.config.config_file)
+
+ dycon = self.config.dynamic_context
+ if not dycon or dycon == "none":
+ context_switchers = []
+ elif dycon == "test_function":
+ context_switchers = [should_start_context_test_function]
+ else:
+ raise CoverageException(
+ "Don't understand dynamic_context setting: {!r}".format(dycon)
+ )
+
+ context_switchers.extend(
+ plugin.dynamic_context for plugin in self._plugins.context_switchers
+ )
+
+ should_start_context = combine_context_switchers(context_switchers)
+
+ self._collector = Collector(
+ should_trace=self._should_trace,
+ check_include=self._check_include_omit_etc,
+ should_start_context=should_start_context,
+ file_mapper=self._file_mapper,
+ timid=self.config.timid,
+ branch=self.config.branch,
+ warn=self._warn,
+ concurrency=concurrency,
+ )
+
+ suffix = self._data_suffix_specified
+ if suffix or self.config.parallel:
+ if not isinstance(suffix, string_class):
+ # if data_suffix=True, use .machinename.pid.random
+ suffix = True
+ else:
+ suffix = None
+
+ self._init_data(suffix)
+
+ self._collector.use_data(self._data, self.config.context)
+
+ # Early warning if we aren't going to be able to support plugins.
+ if self._plugins.file_tracers and not self._collector.supports_plugins:
+ self._warn(
+ "Plugin file tracers (%s) aren't supported with %s" % (
+ ", ".join(
+ plugin._coverage_plugin_name
+ for plugin in self._plugins.file_tracers
+ ),
+ self._collector.tracer_name(),
+ )
+ )
+ for plugin in self._plugins.file_tracers:
+ plugin._coverage_enabled = False
+
+ # Create the file classifying substructure.
+ self._inorout = self._inorout_class(warn=self._warn)
+ self._inorout.configure(self.config)
+ self._inorout.plugins = self._plugins
+ self._inorout.disp_class = self._collector.file_disposition_class
+
+ # It's useful to write debug info after initing for start.
+ self._should_write_debug = True
+
+ atexit.register(self._atexit)
+
+ def _init_data(self, suffix):
+ """Create a data file if we don't have one yet."""
+ if self._data is None:
+ # Create the data file. We do this at construction time so that the
+ # data file will be written into the directory where the process
+ # started rather than wherever the process eventually chdir'd to.
+ ensure_dir_for_file(self.config.data_file)
+ self._data = CoverageData(
+ basename=self.config.data_file,
+ suffix=suffix,
+ warn=self._warn,
+ debug=self._debug,
+ no_disk=self._no_disk,
+ )
def start(self):
"""Start measuring code coverage.
@@ -688,45 +513,82 @@
"""
self._init()
- if self.include:
- if self.source or self.source_pkgs:
- self._warn("--include is ignored because --source is set", slug="include-ignored")
- if self.run_suffix:
- # Calling start() means we're running code, so use the run_suffix
- # as the data_suffix when we eventually save the data.
- self.data_suffix = self.run_suffix
+ if not self._inited_for_start:
+ self._inited_for_start = True
+ self._init_for_start()
+ self._post_init()
+
+ # Issue warnings for possible problems.
+ self._inorout.warn_conflicting_settings()
+
+ # See if we think some code that would eventually be measured has
+ # already been imported.
+ if self._warn_preimported_source:
+ self._inorout.warn_already_imported_files()
+
if self._auto_load:
self.load()
- self.collector.start()
+ self._collector.start()
self._started = True
+ self._instances.append(self)
def stop(self):
"""Stop measuring code coverage."""
+ if self._instances:
+ if self._instances[-1] is self:
+ self._instances.pop()
if self._started:
- self.collector.stop()
+ self._collector.stop()
self._started = False
def _atexit(self):
"""Clean up on process shutdown."""
- if self.debug.should("process"):
- self.debug.write("atexit: {0!r}".format(self))
+ if self._debug.should("process"):
+ self._debug.write("atexit: {!r}".format(self))
if self._started:
self.stop()
if self._auto_save:
self.save()
def erase(self):
- """Erase previously-collected coverage data.
+ """Erase previously collected coverage data.
This removes the in-memory data collected in this session as well as
discarding the data file.
"""
self._init()
- self.collector.reset()
- self.data.erase()
- self.data_files.erase(parallel=self.config.parallel)
+ self._post_init()
+ if self._collector:
+ self._collector.reset()
+ self._init_data(suffix=None)
+ self._data.erase(parallel=self.config.parallel)
+ self._data = None
+ self._inited_for_start = False
+
+ def switch_context(self, new_context):
+ """Switch to a new dynamic context.
+
+ `new_context` is a string to use as the :ref:`dynamic context
+ <dynamic_contexts>` label for collected data. If a :ref:`static
+ context <static_contexts>` is in use, the static and dynamic context
+ labels will be joined together with a pipe character.
+
+ Coverage collection must be started already.
+
+ .. versionadded:: 5.0
+
+ """
+ if not self._started: # pragma: part started
+ raise CoverageException(
+ "Cannot switch context, coverage is not started"
+ )
+
+ if self._collector.should_start_context:
+ self._warn("Conflicting dynamic contexts", slug="dynamic-conflict", once=True)
+
+ self._collector.switch_context(new_context)
def clear_exclude(self, which='exclude'):
"""Clear the exclude list."""
@@ -777,9 +639,8 @@
def save(self):
"""Save the collected coverage data to the data file."""
- self._init()
- self.get_data()
- self.data_files.write(self.data, suffix=self.data_suffix)
+ data = self.get_data()
+ data.write()
def combine(self, data_paths=None, strict=False):
"""Combine together a number of similarly-named coverage data files.
@@ -804,6 +665,8 @@
"""
self._init()
+ self._init_data(suffix=None)
+ self._post_init()
self.get_data()
aliases = None
@@ -814,9 +677,7 @@
for pattern in paths[1:]:
aliases.add(pattern, result)
- self.data_files.combine_parallel_data(
- self.data, aliases=aliases, data_paths=data_paths, strict=strict,
- )
+ combine_parallel_data(self._data, aliases=aliases, data_paths=data_paths, strict=strict)
def get_data(self):
"""Get the collected data.
@@ -829,11 +690,13 @@
"""
self._init()
+ self._init_data(suffix=None)
+ self._post_init()
- if self.collector.save_data(self.data):
+ if self._collector and self._collector.flush_data():
self._post_save_work()
- return self.data
+ return self._data
def _post_save_work(self):
"""After saving data, look for warnings, post-work, etc.
@@ -845,78 +708,21 @@
# If there are still entries in the source_pkgs_unmatched list,
# then we never encountered those packages.
if self._warn_unimported_source:
- for pkg in self.source_pkgs_unmatched:
- self._warn_about_unmeasured_code(pkg)
+ self._inorout.warn_unimported_source()
# Find out if we got any data.
- if not self.data and self._warn_no_data:
+ if not self._data and self._warn_no_data:
self._warn("No data was collected.", slug="no-data-collected")
- # Find files that were never executed at all.
- for pkg in self.source_pkgs:
- if (not pkg in sys.modules or
- not module_has_file(sys.modules[pkg])):
- continue
- pkg_file = source_for_file(sys.modules[pkg].__file__)
- self._find_unexecuted_files(self._canonical_path(pkg_file))
-
- for src in self.source:
- self._find_unexecuted_files(src)
+ # Touch all the files that could have executed, so that we can
+ # mark completely unexecuted files as 0% covered.
+ if self._data is not None:
+ for file_path, plugin_name in self._inorout.find_possibly_unexecuted_files():
+ file_path = self._file_mapper(file_path)
+ self._data.touch_file(file_path, plugin_name)
if self.config.note:
- self.data.add_run_info(note=self.config.note)
-
- def _warn_about_unmeasured_code(self, pkg):
- """Warn about a package or module that we never traced.
-
- `pkg` is a string, the name of the package or module.
-
- """
- mod = sys.modules.get(pkg)
- if mod is None:
- self._warn("Module %s was never imported." % pkg, slug="module-not-imported")
- return
-
- if module_is_namespace(mod):
- # A namespace package. It's OK for this not to have been traced,
- # since there is no code directly in it.
- return
-
- if not module_has_file(mod):
- self._warn("Module %s has no Python source." % pkg, slug="module-not-python")
- return
-
- # The module was in sys.modules, and seems like a module with code, but
- # we never measured it. I guess that means it was imported before
- # coverage even started.
- self._warn(
- "Module %s was previously imported, but not measured" % pkg,
- slug="module-not-measured",
- )
-
- def _find_plugin_files(self, src_dir):
- """Get executable files from the plugins."""
- for plugin in self.plugins.file_tracers:
- for x_file in plugin.find_executable_files(src_dir):
- yield x_file, plugin._coverage_plugin_name
-
- def _find_unexecuted_files(self, src_dir):
- """Find unexecuted files in `src_dir`.
-
- Search for files in `src_dir` that are probably importable,
- and add them as unexecuted files in `self.data`.
-
- """
- py_files = ((py_file, None) for py_file in find_python_files(src_dir))
- plugin_files = self._find_plugin_files(src_dir)
-
- for file_path, plugin_name in itertools.chain(py_files, plugin_files):
- file_path = canonical_filename(file_path)
- if self.omit_match and self.omit_match.match(file_path):
- # Turns out this file was omitted, so don't pull it back
- # in as unexecuted.
- continue
- self.data.touch_file(file_path, plugin_name)
+ self._warn("The '[run] note' setting is no longer supported.")
# Backward compatibility with version 1.
def analysis(self, morf):
@@ -941,7 +747,6 @@
coverage data.
"""
- self._init()
analysis = self._analyze(morf)
return (
analysis.filename,
@@ -957,11 +762,16 @@
Returns an `Analysis` object.
"""
- self.get_data()
+ # All reporting comes through here, so do reporting initialization.
+ self._init()
+ Numbers.set_precision(self.config.precision)
+ self._post_init()
+
+ data = self.get_data()
if not isinstance(it, FileReporter):
it = self._get_file_reporter(it)
- return Analysis(self.data, it)
+ return Analysis(data, it, self._file_mapper)
def _get_file_reporter(self, morf):
"""Get a FileReporter for a module or file name."""
@@ -969,19 +779,19 @@
file_reporter = "python"
if isinstance(morf, string_class):
- abs_morf = abs_file(morf)
- plugin_name = self.data.file_tracer(abs_morf)
+ mapped_morf = self._file_mapper(morf)
+ plugin_name = self._data.file_tracer(mapped_morf)
if plugin_name:
- plugin = self.plugins.get(plugin_name)
+ plugin = self._plugins.get(plugin_name)
- if plugin:
- file_reporter = plugin.file_reporter(abs_morf)
- if file_reporter is None:
- raise CoverageException(
- "Plugin %r did not provide a file reporter for %r." % (
- plugin._coverage_plugin_name, morf
- )
- )
+ if plugin:
+ file_reporter = plugin.file_reporter(mapped_morf)
+ if file_reporter is None:
+ raise CoverageException(
+ "Plugin %r did not provide a file reporter for %r." % (
+ plugin._coverage_plugin_name, morf
+ )
+ )
if file_reporter == "python":
file_reporter = PythonFileReporter(morf, self)
@@ -1000,49 +810,70 @@
"""
if not morfs:
- morfs = self.data.measured_files()
+ morfs = self._data.measured_files()
- # Be sure we have a list.
- if not isinstance(morfs, (list, tuple)):
+ # Be sure we have a collection.
+ if not isinstance(morfs, (list, tuple, set)):
morfs = [morfs]
- file_reporters = []
- for morf in morfs:
- file_reporter = self._get_file_reporter(morf)
- file_reporters.append(file_reporter)
-
+ file_reporters = [self._get_file_reporter(morf) for morf in morfs]
return file_reporters
def report(
self, morfs=None, show_missing=None, ignore_errors=None,
- file=None, # pylint: disable=redefined-builtin
- omit=None, include=None, skip_covered=None,
+ file=None, omit=None, include=None, skip_covered=None,
+ contexts=None, skip_empty=None,
):
- """Write a summary report to `file`.
+ """Write a textual summary report to `file`.
Each module in `morfs` is listed, with counts of statements, executed
statements, missing statements, and a list of lines missed.
+ If `show_missing` is true, then details of which lines or branches are
+ missing will be included in the report. If `ignore_errors` is true,
+ then a failure while reporting a single file will not stop the entire
+ report.
+
+ `file` is a file-like object, suitable for writing.
+
`include` is a list of file name patterns. Files that match will be
included in the report. Files matching `omit` will not be included in
the report.
- If `skip_covered` is True, don't report on files with 100% coverage.
+ If `skip_covered` is true, don't report on files with 100% coverage.
+
+ If `skip_empty` is true, don't report on empty files (those that have
+ no statements).
+
+ `contexts` is a list of regular expressions. Only data from
+ :ref:`dynamic contexts <dynamic_contexts>` that match one of those
+ expressions (using :func:`re.search <python:re.search>`) will be
+ included in the report.
+
+ All of the arguments default to the settings read from the
+ :ref:`configuration file <config>`.
Returns a float, the total percentage covered.
+ .. versionadded:: 4.0
+ The `skip_covered` parameter.
+
+ .. versionadded:: 5.0
+ The `contexts` and `skip_empty` parameters.
+
"""
- self.get_data()
- self.config.from_args(
+ with override_config(
+ self,
ignore_errors=ignore_errors, report_omit=omit, report_include=include,
show_missing=show_missing, skip_covered=skip_covered,
- )
- reporter = SummaryReporter(self, self.config)
- return reporter.report(morfs, outfile=file)
+ report_contexts=contexts, skip_empty=skip_empty,
+ ):
+ reporter = SummaryReporter(self)
+ return reporter.report(morfs, outfile=file)
def annotate(
self, morfs=None, directory=None, ignore_errors=None,
- omit=None, include=None,
+ omit=None, include=None, contexts=None,
):
"""Annotate a list of modules.
@@ -1054,16 +885,17 @@
See :meth:`report` for other arguments.
"""
- self.get_data()
- self.config.from_args(
- ignore_errors=ignore_errors, report_omit=omit, report_include=include
- )
- reporter = AnnotateReporter(self, self.config)
- reporter.report(morfs, directory=directory)
+ with override_config(self,
+ ignore_errors=ignore_errors, report_omit=omit,
+ report_include=include, report_contexts=contexts,
+ ):
+ reporter = AnnotateReporter(self)
+ reporter.report(morfs, directory=directory)
def html_report(self, morfs=None, directory=None, ignore_errors=None,
omit=None, include=None, extra_css=None, title=None,
- skip_covered=None):
+ skip_covered=None, show_contexts=None, contexts=None,
+ skip_empty=None):
"""Generate an HTML report.
The HTML is written to `directory`. The file "index.html" is the
@@ -1080,19 +912,25 @@
Returns a float, the total percentage covered.
+ .. note::
+ The HTML report files are generated incrementally based on the
+ source files and coverage results. If you modify the report files,
+ the changes will not be considered. You should be careful about
+ changing the files in the report folder.
+
"""
- self.get_data()
- self.config.from_args(
+ with override_config(self,
ignore_errors=ignore_errors, report_omit=omit, report_include=include,
html_dir=directory, extra_css=extra_css, html_title=title,
- skip_covered=skip_covered,
- )
- reporter = HtmlReporter(self, self.config)
- return reporter.report(morfs)
+ skip_covered=skip_covered, show_contexts=show_contexts, report_contexts=contexts,
+ skip_empty=skip_empty,
+ ):
+ reporter = HtmlReporter(self)
+ return reporter.report(morfs)
def xml_report(
self, morfs=None, outfile=None, ignore_errors=None,
- omit=None, include=None,
+ omit=None, include=None, contexts=None,
):
"""Generate an XML report of coverage results.
@@ -1106,40 +944,35 @@
Returns a float, the total percentage covered.
"""
- self.get_data()
- self.config.from_args(
+ with override_config(self,
ignore_errors=ignore_errors, report_omit=omit, report_include=include,
- xml_output=outfile,
- )
- file_to_close = None
- delete_file = False
- if self.config.xml_output:
- if self.config.xml_output == '-':
- outfile = sys.stdout
- else:
- # Ensure that the output directory is created; done here
- # because this report pre-opens the output file.
- # HTMLReport does this using the Report plumbing because
- # its task is more complex, being multiple files.
- output_dir = os.path.dirname(self.config.xml_output)
- if output_dir and not os.path.isdir(output_dir):
- os.makedirs(output_dir)
- open_kwargs = {}
- if env.PY3:
- open_kwargs['encoding'] = 'utf8'
- outfile = open(self.config.xml_output, "w", **open_kwargs)
- file_to_close = outfile
- try:
- reporter = XmlReporter(self, self.config)
- return reporter.report(morfs, outfile=outfile)
- except CoverageException:
- delete_file = True
- raise
- finally:
- if file_to_close:
- file_to_close.close()
- if delete_file:
- file_be_gone(self.config.xml_output)
+ xml_output=outfile, report_contexts=contexts,
+ ):
+ return render_report(self.config.xml_output, XmlReporter(self), morfs)
+
+ def json_report(
+ self, morfs=None, outfile=None, ignore_errors=None,
+ omit=None, include=None, contexts=None, pretty_print=None,
+ show_contexts=None
+ ):
+ """Generate a JSON report of coverage results.
+
+ Each module in `morfs` is included in the report. `outfile` is the
+ path to write the file to, "-" will write to stdout.
+
+ See :meth:`report` for other arguments.
+
+ Returns a float, the total percentage covered.
+
+ .. versionadded:: 5.0
+
+ """
+ with override_config(self,
+ ignore_errors=ignore_errors, report_omit=omit, report_include=include,
+ json_output=outfile, report_contexts=contexts, json_pretty_print=pretty_print,
+ json_show_contexts=show_contexts
+ ):
+ return render_report(self.config.json_output, JsonReporter(self), morfs)
def sys_info(self):
"""Return a list of (key, value) pairs showing internal information."""
@@ -1147,6 +980,7 @@
import coverage as covmod
self._init()
+ self._post_init()
def plugin_info(plugins):
"""Make an entry for the sys_info from a list of plug-ins."""
@@ -1161,84 +995,51 @@
info = [
('version', covmod.__version__),
('coverage', covmod.__file__),
- ('cover_paths', self.cover_paths),
- ('pylib_paths', self.pylib_paths),
- ('tracer', self.collector.tracer_name()),
- ('plugins.file_tracers', plugin_info(self.plugins.file_tracers)),
- ('plugins.configurers', plugin_info(self.plugins.configurers)),
- ('config_files', self.config.attempted_config_files),
- ('configs_read', self.config.config_files),
- ('data_path', self.data_files.filename),
+ ('tracer', self._collector.tracer_name() if self._collector else "-none-"),
+ ('CTracer', 'available' if CTracer else "unavailable"),
+ ('plugins.file_tracers', plugin_info(self._plugins.file_tracers)),
+ ('plugins.configurers', plugin_info(self._plugins.configurers)),
+ ('plugins.context_switchers', plugin_info(self._plugins.context_switchers)),
+ ('configs_attempted', self.config.attempted_config_files),
+ ('configs_read', self.config.config_files_read),
+ ('config_file', self.config.config_file),
+ ('config_contents',
+ repr(self.config._config_contents)
+ if self.config._config_contents
+ else '-none-'
+ ),
+ ('data_file', self._data.data_filename() if self._data is not None else "-none-"),
('python', sys.version.replace('\n', '')),
('platform', platform.platform()),
('implementation', platform.python_implementation()),
('executable', sys.executable),
+ ('def_encoding', sys.getdefaultencoding()),
+ ('fs_encoding', sys.getfilesystemencoding()),
+ ('pid', os.getpid()),
('cwd', os.getcwd()),
('path', sys.path),
('environment', sorted(
("%s = %s" % (k, v))
for k, v in iitems(os.environ)
- if k.startswith(("COV", "PY"))
+ if any(slug in k for slug in ("COV", "PY"))
)),
- ('command_line', " ".join(getattr(sys, 'argv', ['???']))),
+ ('command_line', " ".join(getattr(sys, 'argv', ['-none-']))),
]
- matcher_names = [
- 'source_match', 'source_pkgs_match',
- 'include_match', 'omit_match',
- 'cover_match', 'pylib_match',
- ]
+ if self._inorout:
+ info.extend(self._inorout.sys_info())
- for matcher_name in matcher_names:
- matcher = getattr(self, matcher_name)
- if matcher:
- matcher_info = matcher.info()
- else:
- matcher_info = '-none-'
- info.append((matcher_name, matcher_info))
+ info.extend(CoverageData.sys_info())
return info
-def module_is_namespace(mod):
- """Is the module object `mod` a PEP420 namespace module?"""
- return hasattr(mod, '__path__') and getattr(mod, '__file__', None) is None
-
-
-def module_has_file(mod):
- """Does the module object `mod` have an existing __file__ ?"""
- mod__file__ = getattr(mod, '__file__', None)
- if mod__file__ is None:
- return False
- return os.path.exists(mod__file__)
-
-
-# FileDisposition "methods": FileDisposition is a pure value object, so it can
-# be implemented in either C or Python. Acting on them is done with these
-# functions.
+# Mega debugging...
+# $set_env.py: COVERAGE_DEBUG_CALLS - Lots and lots of output about calls to Coverage.
+if int(os.environ.get("COVERAGE_DEBUG_CALLS", 0)): # pragma: debugging
+ from coverage.debug import decorate_methods, show_calls
-def _disposition_init(cls, original_filename):
- """Construct and initialize a new FileDisposition object."""
- disp = cls()
- disp.original_filename = original_filename
- disp.canonical_filename = original_filename
- disp.source_filename = None
- disp.trace = False
- disp.reason = ""
- disp.file_tracer = None
- disp.has_dynamic_filename = False
- return disp
-
-
-def _disposition_debug_msg(disp):
- """Make a nice debug message of what the FileDisposition is doing."""
- if disp.trace:
- msg = "Tracing %r" % (disp.original_filename,)
- if disp.file_tracer:
- msg += ": will be traced by %r" % disp.file_tracer
- else:
- msg = "Not tracing %r: %s" % (disp.original_filename, disp.reason)
- return msg
+ Coverage = decorate_methods(show_calls(show_args=True), butnot=['get_data'])(Coverage)
def process_startup():
@@ -1286,10 +1087,11 @@
cov = Coverage(config_file=cps)
process_startup.coverage = cov
- cov.start()
cov._warn_no_data = False
cov._warn_unimported_source = False
+ cov._warn_preimported_source = False
cov._auto_save = True
+ cov.start()
return cov
