Add python venv
This commit is contained in:
Isaac Shoebottom
@ -0,0 +1,2 @@
|
||||
"""pytest-cov: avoid already-imported warning: PYTEST_DONT_REWRITE."""
|
||||
__version__ = '4.0.0'
|
||||
24
utils/python-venv/Lib/site-packages/pytest_cov/compat.py
Normal file
24
utils/python-venv/Lib/site-packages/pytest_cov/compat.py
Normal file
@ -0,0 +1,24 @@
|
||||
try:
|
||||
from StringIO import StringIO
|
||||
except ImportError:
|
||||
from io import StringIO
|
||||
|
||||
|
||||
StringIO # pyflakes, this is for re-export
|
||||
|
||||
|
||||
class SessionWrapper:
|
||||
def __init__(self, session):
|
||||
self._session = session
|
||||
if hasattr(session, 'testsfailed'):
|
||||
self._attr = 'testsfailed'
|
||||
else:
|
||||
self._attr = '_testsfailed'
|
||||
|
||||
@property
|
||||
def testsfailed(self):
|
||||
return getattr(self._session, self._attr)
|
||||
|
||||
@testsfailed.setter
|
||||
def testsfailed(self, value):
|
||||
setattr(self._session, self._attr, value)
|
||||
122
utils/python-venv/Lib/site-packages/pytest_cov/embed.py
Normal file
122
utils/python-venv/Lib/site-packages/pytest_cov/embed.py
Normal file
@ -0,0 +1,122 @@
|
||||
"""Activate coverage at python startup if appropriate.
|
||||
|
||||
The python site initialisation will ensure that anything we import
|
||||
will be removed and not visible at the end of python startup. However
|
||||
we minimise all work by putting these init actions in this separate
|
||||
module and only importing what is needed when needed.
|
||||
|
||||
For normal python startup when coverage should not be activated the pth
|
||||
file checks a single env var and does not import or call the init fn
|
||||
here.
|
||||
|
||||
For python startup when an ancestor process has set the env indicating
|
||||
that code coverage is being collected we activate coverage based on
|
||||
info passed via env vars.
|
||||
"""
|
||||
import atexit
|
||||
import os
|
||||
import signal
|
||||
|
||||
_active_cov = None
|
||||
|
||||
|
||||
def init():
|
||||
# Only continue if ancestor process has set everything needed in
|
||||
# the env.
|
||||
global _active_cov
|
||||
|
||||
cov_source = os.environ.get('COV_CORE_SOURCE')
|
||||
cov_config = os.environ.get('COV_CORE_CONFIG')
|
||||
cov_datafile = os.environ.get('COV_CORE_DATAFILE')
|
||||
cov_branch = True if os.environ.get('COV_CORE_BRANCH') == 'enabled' else None
|
||||
cov_context = os.environ.get('COV_CORE_CONTEXT')
|
||||
|
||||
if cov_datafile:
|
||||
if _active_cov:
|
||||
cleanup()
|
||||
# Import what we need to activate coverage.
|
||||
import coverage
|
||||
|
||||
# Determine all source roots.
|
||||
if cov_source in os.pathsep:
|
||||
cov_source = None
|
||||
else:
|
||||
cov_source = cov_source.split(os.pathsep)
|
||||
if cov_config == os.pathsep:
|
||||
cov_config = True
|
||||
|
||||
# Activate coverage for this process.
|
||||
cov = _active_cov = coverage.Coverage(
|
||||
source=cov_source,
|
||||
branch=cov_branch,
|
||||
data_suffix=True,
|
||||
config_file=cov_config,
|
||||
auto_data=True,
|
||||
data_file=cov_datafile
|
||||
)
|
||||
cov.load()
|
||||
cov.start()
|
||||
if cov_context:
|
||||
cov.switch_context(cov_context)
|
||||
cov._warn_no_data = False
|
||||
cov._warn_unimported_source = False
|
||||
return cov
|
||||
|
||||
|
||||
def _cleanup(cov):
|
||||
if cov is not None:
|
||||
cov.stop()
|
||||
cov.save()
|
||||
cov._auto_save = False # prevent autosaving from cov._atexit in case the interpreter lacks atexit.unregister
|
||||
try:
|
||||
atexit.unregister(cov._atexit)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def cleanup():
|
||||
global _active_cov
|
||||
global _cleanup_in_progress
|
||||
global _pending_signal
|
||||
|
||||
_cleanup_in_progress = True
|
||||
_cleanup(_active_cov)
|
||||
_active_cov = None
|
||||
_cleanup_in_progress = False
|
||||
if _pending_signal:
|
||||
pending_signal = _pending_signal
|
||||
_pending_signal = None
|
||||
_signal_cleanup_handler(*pending_signal)
|
||||
|
||||
|
||||
_previous_handlers = {}
|
||||
_pending_signal = None
|
||||
_cleanup_in_progress = False
|
||||
|
||||
|
||||
def _signal_cleanup_handler(signum, frame):
|
||||
global _pending_signal
|
||||
if _cleanup_in_progress:
|
||||
_pending_signal = signum, frame
|
||||
return
|
||||
cleanup()
|
||||
_previous_handler = _previous_handlers.get(signum)
|
||||
if _previous_handler == signal.SIG_IGN:
|
||||
return
|
||||
elif _previous_handler and _previous_handler is not _signal_cleanup_handler:
|
||||
_previous_handler(signum, frame)
|
||||
elif signum == signal.SIGTERM:
|
||||
os._exit(128 + signum)
|
||||
elif signum == signal.SIGINT:
|
||||
raise KeyboardInterrupt()
|
||||
|
||||
|
||||
def cleanup_on_signal(signum):
|
||||
previous = signal.getsignal(signum)
|
||||
if previous is not _signal_cleanup_handler:
|
||||
_previous_handlers[signum] = previous
|
||||
signal.signal(signum, _signal_cleanup_handler)
|
||||
|
||||
|
||||
def cleanup_on_sigterm():
|
||||
cleanup_on_signal(signal.SIGTERM)
|
||||
414
utils/python-venv/Lib/site-packages/pytest_cov/engine.py
Normal file
414
utils/python-venv/Lib/site-packages/pytest_cov/engine.py
Normal file
@ -0,0 +1,414 @@
|
||||
"""Coverage controllers for use by pytest-cov and nose-cov."""
|
||||
import contextlib
|
||||
import copy
|
||||
import functools
|
||||
import os
|
||||
import random
|
||||
import socket
|
||||
import sys
|
||||
|
||||
import coverage
|
||||
from coverage.data import CoverageData
|
||||
|
||||
from .compat import StringIO
|
||||
from .embed import cleanup
|
||||
|
||||
|
||||
class _NullFile:
|
||||
@staticmethod
|
||||
def write(v):
|
||||
pass
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _backup(obj, attr):
|
||||
backup = getattr(obj, attr)
|
||||
try:
|
||||
setattr(obj, attr, copy.copy(backup))
|
||||
yield
|
||||
finally:
|
||||
setattr(obj, attr, backup)
|
||||
|
||||
|
||||
def _ensure_topdir(meth):
|
||||
@functools.wraps(meth)
|
||||
def ensure_topdir_wrapper(self, *args, **kwargs):
|
||||
try:
|
||||
original_cwd = os.getcwd()
|
||||
except OSError:
|
||||
# Looks like it's gone, this is non-ideal because a side-effect will
|
||||
# be introduced in the tests here but we can't do anything about it.
|
||||
original_cwd = None
|
||||
os.chdir(self.topdir)
|
||||
try:
|
||||
return meth(self, *args, **kwargs)
|
||||
finally:
|
||||
if original_cwd is not None:
|
||||
os.chdir(original_cwd)
|
||||
|
||||
return ensure_topdir_wrapper
|
||||
|
||||
|
||||
class CovController:
|
||||
"""Base class for different plugin implementations."""
|
||||
|
||||
def __init__(self, cov_source, cov_report, cov_config, cov_append, cov_branch, config=None, nodeid=None):
|
||||
"""Get some common config used by multiple derived classes."""
|
||||
self.cov_source = cov_source
|
||||
self.cov_report = cov_report
|
||||
self.cov_config = cov_config
|
||||
self.cov_append = cov_append
|
||||
self.cov_branch = cov_branch
|
||||
self.config = config
|
||||
self.nodeid = nodeid
|
||||
|
||||
self.cov = None
|
||||
self.combining_cov = None
|
||||
self.data_file = None
|
||||
self.node_descs = set()
|
||||
self.failed_workers = []
|
||||
self.topdir = os.getcwd()
|
||||
self.is_collocated = None
|
||||
|
||||
@contextlib.contextmanager
|
||||
def ensure_topdir(self):
|
||||
original_cwd = os.getcwd()
|
||||
os.chdir(self.topdir)
|
||||
yield
|
||||
os.chdir(original_cwd)
|
||||
|
||||
@_ensure_topdir
|
||||
def pause(self):
|
||||
self.cov.stop()
|
||||
self.unset_env()
|
||||
|
||||
@_ensure_topdir
|
||||
def resume(self):
|
||||
self.cov.start()
|
||||
self.set_env()
|
||||
|
||||
@_ensure_topdir
|
||||
def set_env(self):
|
||||
"""Put info about coverage into the env so that subprocesses can activate coverage."""
|
||||
if self.cov_source is None:
|
||||
os.environ['COV_CORE_SOURCE'] = os.pathsep
|
||||
else:
|
||||
os.environ['COV_CORE_SOURCE'] = os.pathsep.join(self.cov_source)
|
||||
config_file = os.path.abspath(self.cov_config)
|
||||
if os.path.exists(config_file):
|
||||
os.environ['COV_CORE_CONFIG'] = config_file
|
||||
else:
|
||||
os.environ['COV_CORE_CONFIG'] = os.pathsep
|
||||
os.environ['COV_CORE_DATAFILE'] = os.path.abspath(self.cov.config.data_file)
|
||||
if self.cov_branch:
|
||||
os.environ['COV_CORE_BRANCH'] = 'enabled'
|
||||
|
||||
@staticmethod
|
||||
def unset_env():
|
||||
"""Remove coverage info from env."""
|
||||
os.environ.pop('COV_CORE_SOURCE', None)
|
||||
os.environ.pop('COV_CORE_CONFIG', None)
|
||||
os.environ.pop('COV_CORE_DATAFILE', None)
|
||||
os.environ.pop('COV_CORE_BRANCH', None)
|
||||
os.environ.pop('COV_CORE_CONTEXT', None)
|
||||
|
||||
@staticmethod
|
||||
def get_node_desc(platform, version_info):
|
||||
"""Return a description of this node."""
|
||||
|
||||
return 'platform {}, python {}'.format(platform, '%s.%s.%s-%s-%s' % version_info[:5])
|
||||
|
||||
@staticmethod
|
||||
def sep(stream, s, txt):
|
||||
if hasattr(stream, 'sep'):
|
||||
stream.sep(s, txt)
|
||||
else:
|
||||
sep_total = max((70 - 2 - len(txt)), 2)
|
||||
sep_len = sep_total // 2
|
||||
sep_extra = sep_total % 2
|
||||
out = f'{s * sep_len} {txt} {s * (sep_len + sep_extra)}\n'
|
||||
stream.write(out)
|
||||
|
||||
@_ensure_topdir
|
||||
def summary(self, stream):
|
||||
"""Produce coverage reports."""
|
||||
total = None
|
||||
|
||||
if not self.cov_report:
|
||||
with _backup(self.cov, "config"):
|
||||
return self.cov.report(show_missing=True, ignore_errors=True, file=_NullFile)
|
||||
|
||||
# Output coverage section header.
|
||||
if len(self.node_descs) == 1:
|
||||
self.sep(stream, '-', 'coverage: %s' % ''.join(self.node_descs))
|
||||
else:
|
||||
self.sep(stream, '-', 'coverage')
|
||||
for node_desc in sorted(self.node_descs):
|
||||
self.sep(stream, ' ', '%s' % node_desc)
|
||||
|
||||
# Report on any failed workers.
|
||||
if self.failed_workers:
|
||||
self.sep(stream, '-', 'coverage: failed workers')
|
||||
stream.write('The following workers failed to return coverage data, '
|
||||
'ensure that pytest-cov is installed on these workers.\n')
|
||||
for node in self.failed_workers:
|
||||
stream.write('%s\n' % node.gateway.id)
|
||||
|
||||
# Produce terminal report if wanted.
|
||||
if any(x in self.cov_report for x in ['term', 'term-missing']):
|
||||
options = {
|
||||
'show_missing': ('term-missing' in self.cov_report) or None,
|
||||
'ignore_errors': True,
|
||||
'file': stream,
|
||||
}
|
||||
skip_covered = isinstance(self.cov_report, dict) and 'skip-covered' in self.cov_report.values()
|
||||
options.update({'skip_covered': skip_covered or None})
|
||||
with _backup(self.cov, "config"):
|
||||
total = self.cov.report(**options)
|
||||
|
||||
# Produce annotated source code report if wanted.
|
||||
if 'annotate' in self.cov_report:
|
||||
annotate_dir = self.cov_report['annotate']
|
||||
|
||||
with _backup(self.cov, "config"):
|
||||
self.cov.annotate(ignore_errors=True, directory=annotate_dir)
|
||||
# We need to call Coverage.report here, just to get the total
|
||||
# Coverage.annotate don't return any total and we need it for --cov-fail-under.
|
||||
|
||||
with _backup(self.cov, "config"):
|
||||
total = self.cov.report(ignore_errors=True, file=_NullFile)
|
||||
if annotate_dir:
|
||||
stream.write('Coverage annotated source written to dir %s\n' % annotate_dir)
|
||||
else:
|
||||
stream.write('Coverage annotated source written next to source\n')
|
||||
|
||||
# Produce html report if wanted.
|
||||
if 'html' in self.cov_report:
|
||||
output = self.cov_report['html']
|
||||
with _backup(self.cov, "config"):
|
||||
total = self.cov.html_report(ignore_errors=True, directory=output)
|
||||
stream.write('Coverage HTML written to dir %s\n' % (self.cov.config.html_dir if output is None else output))
|
||||
|
||||
# Produce xml report if wanted.
|
||||
if 'xml' in self.cov_report:
|
||||
output = self.cov_report['xml']
|
||||
with _backup(self.cov, "config"):
|
||||
total = self.cov.xml_report(ignore_errors=True, outfile=output)
|
||||
stream.write('Coverage XML written to file %s\n' % (self.cov.config.xml_output if output is None else output))
|
||||
|
||||
# Produce lcov report if wanted.
|
||||
if 'lcov' in self.cov_report:
|
||||
output = self.cov_report['lcov']
|
||||
with _backup(self.cov, "config"):
|
||||
self.cov.lcov_report(ignore_errors=True, outfile=output)
|
||||
|
||||
# We need to call Coverage.report here, just to get the total
|
||||
# Coverage.lcov_report doesn't return any total and we need it for --cov-fail-under.
|
||||
total = self.cov.report(ignore_errors=True, file=_NullFile)
|
||||
|
||||
stream.write('Coverage LCOV written to file %s\n' % (self.cov.config.lcov_output if output is None else output))
|
||||
|
||||
return total
|
||||
|
||||
|
||||
class Central(CovController):
|
||||
"""Implementation for centralised operation."""
|
||||
|
||||
@_ensure_topdir
|
||||
def start(self):
|
||||
cleanup()
|
||||
|
||||
self.cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=True,
|
||||
config_file=self.cov_config)
|
||||
self.combining_cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=True,
|
||||
data_file=os.path.abspath(self.cov.config.data_file),
|
||||
config_file=self.cov_config)
|
||||
|
||||
# Erase or load any previous coverage data and start coverage.
|
||||
if not self.cov_append:
|
||||
self.cov.erase()
|
||||
self.cov.start()
|
||||
self.set_env()
|
||||
|
||||
@_ensure_topdir
|
||||
def finish(self):
|
||||
"""Stop coverage, save data to file and set the list of coverage objects to report on."""
|
||||
|
||||
self.unset_env()
|
||||
self.cov.stop()
|
||||
self.cov.save()
|
||||
|
||||
self.cov = self.combining_cov
|
||||
self.cov.load()
|
||||
self.cov.combine()
|
||||
self.cov.save()
|
||||
|
||||
node_desc = self.get_node_desc(sys.platform, sys.version_info)
|
||||
self.node_descs.add(node_desc)
|
||||
|
||||
|
||||
class DistMaster(CovController):
|
||||
"""Implementation for distributed master."""
|
||||
|
||||
@_ensure_topdir
|
||||
def start(self):
|
||||
cleanup()
|
||||
|
||||
# Ensure coverage rc file rsynced if appropriate.
|
||||
if self.cov_config and os.path.exists(self.cov_config):
|
||||
self.config.option.rsyncdir.append(self.cov_config)
|
||||
|
||||
self.cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=True,
|
||||
config_file=self.cov_config)
|
||||
self.cov._warn_no_data = False
|
||||
self.cov._warn_unimported_source = False
|
||||
self.cov._warn_preimported_source = False
|
||||
self.combining_cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=True,
|
||||
data_file=os.path.abspath(self.cov.config.data_file),
|
||||
config_file=self.cov_config)
|
||||
if not self.cov_append:
|
||||
self.cov.erase()
|
||||
self.cov.start()
|
||||
self.cov.config.paths['source'] = [self.topdir]
|
||||
|
||||
def configure_node(self, node):
|
||||
"""Workers need to know if they are collocated and what files have moved."""
|
||||
|
||||
node.workerinput.update({
|
||||
'cov_master_host': socket.gethostname(),
|
||||
'cov_master_topdir': self.topdir,
|
||||
'cov_master_rsync_roots': [str(root) for root in node.nodemanager.roots],
|
||||
})
|
||||
|
||||
def testnodedown(self, node, error):
|
||||
"""Collect data file name from worker."""
|
||||
|
||||
# If worker doesn't return any data then it is likely that this
|
||||
# plugin didn't get activated on the worker side.
|
||||
output = getattr(node, 'workeroutput', {})
|
||||
if 'cov_worker_node_id' not in output:
|
||||
self.failed_workers.append(node)
|
||||
return
|
||||
|
||||
# If worker is not collocated then we must save the data file
|
||||
# that it returns to us.
|
||||
if 'cov_worker_data' in output:
|
||||
data_suffix = '%s.%s.%06d.%s' % (
|
||||
socket.gethostname(), os.getpid(),
|
||||
random.randint(0, 999999),
|
||||
output['cov_worker_node_id']
|
||||
)
|
||||
|
||||
cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=data_suffix,
|
||||
config_file=self.cov_config)
|
||||
cov.start()
|
||||
if coverage.version_info < (5, 0):
|
||||
data = CoverageData()
|
||||
data.read_fileobj(StringIO(output['cov_worker_data']))
|
||||
cov.data.update(data)
|
||||
else:
|
||||
data = CoverageData(no_disk=True)
|
||||
data.loads(output['cov_worker_data'])
|
||||
cov.get_data().update(data)
|
||||
cov.stop()
|
||||
cov.save()
|
||||
path = output['cov_worker_path']
|
||||
self.cov.config.paths['source'].append(path)
|
||||
|
||||
# Record the worker types that contribute to the data file.
|
||||
rinfo = node.gateway._rinfo()
|
||||
node_desc = self.get_node_desc(rinfo.platform, rinfo.version_info)
|
||||
self.node_descs.add(node_desc)
|
||||
|
||||
@_ensure_topdir
|
||||
def finish(self):
|
||||
"""Combines coverage data and sets the list of coverage objects to report on."""
|
||||
|
||||
# Combine all the suffix files into the data file.
|
||||
self.cov.stop()
|
||||
self.cov.save()
|
||||
self.cov = self.combining_cov
|
||||
self.cov.load()
|
||||
self.cov.combine()
|
||||
self.cov.save()
|
||||
|
||||
|
||||
class DistWorker(CovController):
|
||||
"""Implementation for distributed workers."""
|
||||
|
||||
@_ensure_topdir
|
||||
def start(self):
|
||||
|
||||
cleanup()
|
||||
|
||||
# Determine whether we are collocated with master.
|
||||
self.is_collocated = (socket.gethostname() == self.config.workerinput['cov_master_host'] and
|
||||
self.topdir == self.config.workerinput['cov_master_topdir'])
|
||||
|
||||
# If we are not collocated then rewrite master paths to worker paths.
|
||||
if not self.is_collocated:
|
||||
master_topdir = self.config.workerinput['cov_master_topdir']
|
||||
worker_topdir = self.topdir
|
||||
if self.cov_source is not None:
|
||||
self.cov_source = [source.replace(master_topdir, worker_topdir)
|
||||
for source in self.cov_source]
|
||||
self.cov_config = self.cov_config.replace(master_topdir, worker_topdir)
|
||||
|
||||
# Erase any previous data and start coverage.
|
||||
self.cov = coverage.Coverage(source=self.cov_source,
|
||||
branch=self.cov_branch,
|
||||
data_suffix=True,
|
||||
config_file=self.cov_config)
|
||||
self.cov.start()
|
||||
self.set_env()
|
||||
|
||||
@_ensure_topdir
|
||||
def finish(self):
|
||||
"""Stop coverage and send relevant info back to the master."""
|
||||
self.unset_env()
|
||||
self.cov.stop()
|
||||
|
||||
if self.is_collocated:
|
||||
# We don't combine data if we're collocated - we can get
|
||||
# race conditions in the .combine() call (it's not atomic)
|
||||
# The data is going to be combined in the master.
|
||||
self.cov.save()
|
||||
|
||||
# If we are collocated then just inform the master of our
|
||||
# data file to indicate that we have finished.
|
||||
self.config.workeroutput['cov_worker_node_id'] = self.nodeid
|
||||
else:
|
||||
self.cov.combine()
|
||||
self.cov.save()
|
||||
# If we are not collocated then add the current path
|
||||
# and coverage data to the output so we can combine
|
||||
# it on the master node.
|
||||
|
||||
# Send all the data to the master over the channel.
|
||||
if coverage.version_info < (5, 0):
|
||||
buff = StringIO()
|
||||
self.cov.data.write_fileobj(buff)
|
||||
data = buff.getvalue()
|
||||
else:
|
||||
data = self.cov.get_data().dumps()
|
||||
|
||||
self.config.workeroutput.update({
|
||||
'cov_worker_path': self.topdir,
|
||||
'cov_worker_node_id': self.nodeid,
|
||||
'cov_worker_data': data,
|
||||
})
|
||||
|
||||
def summary(self, stream):
|
||||
"""Only the master reports so do nothing."""
|
||||
|
||||
pass
|
||||
411
utils/python-venv/Lib/site-packages/pytest_cov/plugin.py
Normal file
411
utils/python-venv/Lib/site-packages/pytest_cov/plugin.py
Normal file
@ -0,0 +1,411 @@
|
||||
"""Coverage plugin for pytest."""
|
||||
import argparse
|
||||
import os
|
||||
import warnings
|
||||
|
||||
import coverage
|
||||
import pytest
|
||||
|
||||
from . import compat
|
||||
from . import embed
|
||||
|
||||
|
||||
class CoverageError(Exception):
|
||||
"""Indicates that our coverage is too low"""
|
||||
|
||||
|
||||
class PytestCovWarning(pytest.PytestWarning):
|
||||
"""
|
||||
The base for all pytest-cov warnings, never raised directly
|
||||
"""
|
||||
|
||||
|
||||
class CovDisabledWarning(PytestCovWarning):
|
||||
"""Indicates that Coverage was manually disabled"""
|
||||
|
||||
|
||||
class CovReportWarning(PytestCovWarning):
|
||||
"""Indicates that we failed to generate a report"""
|
||||
|
||||
|
||||
def validate_report(arg):
|
||||
file_choices = ['annotate', 'html', 'xml', 'lcov']
|
||||
term_choices = ['term', 'term-missing']
|
||||
term_modifier_choices = ['skip-covered']
|
||||
all_choices = term_choices + file_choices
|
||||
values = arg.split(":", 1)
|
||||
report_type = values[0]
|
||||
if report_type not in all_choices + ['']:
|
||||
msg = f'invalid choice: "{arg}" (choose from "{all_choices}")'
|
||||
raise argparse.ArgumentTypeError(msg)
|
||||
|
||||
if report_type == 'lcov' and coverage.version_info <= (6, 3):
|
||||
raise argparse.ArgumentTypeError('LCOV output is only supported with coverage.py >= 6.3')
|
||||
|
||||
if len(values) == 1:
|
||||
return report_type, None
|
||||
|
||||
report_modifier = values[1]
|
||||
if report_type in term_choices and report_modifier in term_modifier_choices:
|
||||
return report_type, report_modifier
|
||||
|
||||
if report_type not in file_choices:
|
||||
msg = 'output specifier not supported for: "{}" (choose from "{}")'.format(arg,
|
||||
file_choices)
|
||||
raise argparse.ArgumentTypeError(msg)
|
||||
|
||||
return values
|
||||
|
||||
|
||||
def validate_fail_under(num_str):
|
||||
try:
|
||||
value = int(num_str)
|
||||
except ValueError:
|
||||
try:
|
||||
value = float(num_str)
|
||||
except ValueError:
|
||||
raise argparse.ArgumentTypeError('An integer or float value is required.')
|
||||
if value > 100:
|
||||
raise argparse.ArgumentTypeError('Your desire for over-achievement is admirable but misplaced. '
|
||||
'The maximum value is 100. Perhaps write more integration tests?')
|
||||
return value
|
||||
|
||||
|
||||
def validate_context(arg):
|
||||
if coverage.version_info <= (5, 0):
|
||||
raise argparse.ArgumentTypeError('Contexts are only supported with coverage.py >= 5.x')
|
||||
if arg != "test":
|
||||
raise argparse.ArgumentTypeError('The only supported value is "test".')
|
||||
return arg
|
||||
|
||||
|
||||
class StoreReport(argparse.Action):
|
||||
def __call__(self, parser, namespace, values, option_string=None):
|
||||
report_type, file = values
|
||||
namespace.cov_report[report_type] = file
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
"""Add options to control coverage."""
|
||||
|
||||
group = parser.getgroup(
|
||||
'cov', 'coverage reporting with distributed testing support')
|
||||
group.addoption('--cov', action='append', default=[], metavar='SOURCE',
|
||||
nargs='?', const=True, dest='cov_source',
|
||||
help='Path or package name to measure during execution (multi-allowed). '
|
||||
'Use --cov= to not do any source filtering and record everything.')
|
||||
group.addoption('--cov-reset', action='store_const', const=[], dest='cov_source',
|
||||
help='Reset cov sources accumulated in options so far. ')
|
||||
group.addoption('--cov-report', action=StoreReport, default={},
|
||||
metavar='TYPE', type=validate_report,
|
||||
help='Type of report to generate: term, term-missing, '
|
||||
'annotate, html, xml, lcov (multi-allowed). '
|
||||
'term, term-missing may be followed by ":skip-covered". '
|
||||
'annotate, html, xml and lcov may be followed by ":DEST" '
|
||||
'where DEST specifies the output location. '
|
||||
'Use --cov-report= to not generate any output.')
|
||||
group.addoption('--cov-config', action='store', default='.coveragerc',
|
||||
metavar='PATH',
|
||||
help='Config file for coverage. Default: .coveragerc')
|
||||
group.addoption('--no-cov-on-fail', action='store_true', default=False,
|
||||
help='Do not report coverage if test run fails. '
|
||||
'Default: False')
|
||||
group.addoption('--no-cov', action='store_true', default=False,
|
||||
help='Disable coverage report completely (useful for debuggers). '
|
||||
'Default: False')
|
||||
group.addoption('--cov-fail-under', action='store', metavar='MIN',
|
||||
type=validate_fail_under,
|
||||
help='Fail if the total coverage is less than MIN.')
|
||||
group.addoption('--cov-append', action='store_true', default=False,
|
||||
help='Do not delete coverage but append to current. '
|
||||
'Default: False')
|
||||
group.addoption('--cov-branch', action='store_true', default=None,
|
||||
help='Enable branch coverage.')
|
||||
group.addoption('--cov-context', action='store', metavar='CONTEXT',
|
||||
type=validate_context,
|
||||
help='Dynamic contexts to use. "test" for now.')
|
||||
|
||||
|
||||
def _prepare_cov_source(cov_source):
|
||||
"""
|
||||
Prepare cov_source so that:
|
||||
|
||||
--cov --cov=foobar is equivalent to --cov (cov_source=None)
|
||||
--cov=foo --cov=bar is equivalent to cov_source=['foo', 'bar']
|
||||
"""
|
||||
return None if True in cov_source else [path for path in cov_source if path is not True]
|
||||
|
||||
|
||||
@pytest.hookimpl(tryfirst=True)
|
||||
def pytest_load_initial_conftests(early_config, parser, args):
|
||||
options = early_config.known_args_namespace
|
||||
no_cov = options.no_cov_should_warn = False
|
||||
for arg in args:
|
||||
arg = str(arg)
|
||||
if arg == '--no-cov':
|
||||
no_cov = True
|
||||
elif arg.startswith('--cov') and no_cov:
|
||||
options.no_cov_should_warn = True
|
||||
break
|
||||
|
||||
if early_config.known_args_namespace.cov_source:
|
||||
plugin = CovPlugin(options, early_config.pluginmanager)
|
||||
early_config.pluginmanager.register(plugin, '_cov')
|
||||
|
||||
|
||||
class CovPlugin:
|
||||
"""Use coverage package to produce code coverage reports.
|
||||
|
||||
Delegates all work to a particular implementation based on whether
|
||||
this test process is centralised, a distributed master or a
|
||||
distributed worker.
|
||||
"""
|
||||
|
||||
def __init__(self, options, pluginmanager, start=True, no_cov_should_warn=False):
|
||||
"""Creates a coverage pytest plugin.
|
||||
|
||||
We read the rc file that coverage uses to get the data file
|
||||
name. This is needed since we give coverage through it's API
|
||||
the data file name.
|
||||
"""
|
||||
|
||||
# Our implementation is unknown at this time.
|
||||
self.pid = None
|
||||
self.cov_controller = None
|
||||
self.cov_report = compat.StringIO()
|
||||
self.cov_total = None
|
||||
self.failed = False
|
||||
self._started = False
|
||||
self._start_path = None
|
||||
self._disabled = False
|
||||
self.options = options
|
||||
|
||||
is_dist = (getattr(options, 'numprocesses', False) or
|
||||
getattr(options, 'distload', False) or
|
||||
getattr(options, 'dist', 'no') != 'no')
|
||||
if getattr(options, 'no_cov', False):
|
||||
self._disabled = True
|
||||
return
|
||||
|
||||
if not self.options.cov_report:
|
||||
self.options.cov_report = ['term']
|
||||
elif len(self.options.cov_report) == 1 and '' in self.options.cov_report:
|
||||
self.options.cov_report = {}
|
||||
self.options.cov_source = _prepare_cov_source(self.options.cov_source)
|
||||
|
||||
# import engine lazily here to avoid importing
|
||||
# it for unit tests that don't need it
|
||||
from . import engine
|
||||
|
||||
if is_dist and start:
|
||||
self.start(engine.DistMaster)
|
||||
elif start:
|
||||
self.start(engine.Central)
|
||||
|
||||
# worker is started in pytest hook
|
||||
|
||||
def start(self, controller_cls, config=None, nodeid=None):
|
||||
|
||||
if config is None:
|
||||
# fake config option for engine
|
||||
class Config:
|
||||
option = self.options
|
||||
|
||||
config = Config()
|
||||
|
||||
self.cov_controller = controller_cls(
|
||||
self.options.cov_source,
|
||||
self.options.cov_report,
|
||||
self.options.cov_config,
|
||||
self.options.cov_append,
|
||||
self.options.cov_branch,
|
||||
config,
|
||||
nodeid
|
||||
)
|
||||
self.cov_controller.start()
|
||||
self._started = True
|
||||
self._start_path = os.getcwd()
|
||||
cov_config = self.cov_controller.cov.config
|
||||
if self.options.cov_fail_under is None and hasattr(cov_config, 'fail_under'):
|
||||
self.options.cov_fail_under = cov_config.fail_under
|
||||
|
||||
def _is_worker(self, session):
|
||||
return getattr(session.config, 'workerinput', None) is not None
|
||||
|
||||
def pytest_sessionstart(self, session):
|
||||
"""At session start determine our implementation and delegate to it."""
|
||||
|
||||
if self.options.no_cov:
|
||||
# Coverage can be disabled because it does not cooperate with debuggers well.
|
||||
self._disabled = True
|
||||
return
|
||||
|
||||
# import engine lazily here to avoid importing
|
||||
# it for unit tests that don't need it
|
||||
from . import engine
|
||||
|
||||
self.pid = os.getpid()
|
||||
if self._is_worker(session):
|
||||
nodeid = (
|
||||
session.config.workerinput.get('workerid', getattr(session, 'nodeid'))
|
||||
)
|
||||
self.start(engine.DistWorker, session.config, nodeid)
|
||||
elif not self._started:
|
||||
self.start(engine.Central)
|
||||
|
||||
if self.options.cov_context == 'test':
|
||||
session.config.pluginmanager.register(TestContextPlugin(self.cov_controller.cov), '_cov_contexts')
|
||||
|
||||
@pytest.hookimpl(optionalhook=True)
|
||||
def pytest_configure_node(self, node):
|
||||
"""Delegate to our implementation.
|
||||
|
||||
Mark this hook as optional in case xdist is not installed.
|
||||
"""
|
||||
if not self._disabled:
|
||||
self.cov_controller.configure_node(node)
|
||||
|
||||
@pytest.hookimpl(optionalhook=True)
|
||||
def pytest_testnodedown(self, node, error):
|
||||
"""Delegate to our implementation.
|
||||
|
||||
Mark this hook as optional in case xdist is not installed.
|
||||
"""
|
||||
if not self._disabled:
|
||||
self.cov_controller.testnodedown(node, error)
|
||||
|
||||
def _should_report(self):
|
||||
return not (self.failed and self.options.no_cov_on_fail)
|
||||
|
||||
def _failed_cov_total(self):
|
||||
cov_fail_under = self.options.cov_fail_under
|
||||
return cov_fail_under is not None and self.cov_total < cov_fail_under
|
||||
|
||||
# we need to wrap pytest_runtestloop. by the time pytest_sessionfinish
|
||||
# runs, it's too late to set testsfailed
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtestloop(self, session):
|
||||
yield
|
||||
|
||||
if self._disabled:
|
||||
return
|
||||
|
||||
compat_session = compat.SessionWrapper(session)
|
||||
|
||||
self.failed = bool(compat_session.testsfailed)
|
||||
if self.cov_controller is not None:
|
||||
self.cov_controller.finish()
|
||||
|
||||
if not self._is_worker(session) and self._should_report():
|
||||
|
||||
# import coverage lazily here to avoid importing
|
||||
# it for unit tests that don't need it
|
||||
from coverage.misc import CoverageException
|
||||
|
||||
try:
|
||||
self.cov_total = self.cov_controller.summary(self.cov_report)
|
||||
except CoverageException as exc:
|
||||
message = 'Failed to generate report: %s\n' % exc
|
||||
session.config.pluginmanager.getplugin("terminalreporter").write(
|
||||
'WARNING: %s\n' % message, red=True, bold=True)
|
||||
warnings.warn(CovReportWarning(message))
|
||||
self.cov_total = 0
|
||||
assert self.cov_total is not None, 'Test coverage should never be `None`'
|
||||
if self._failed_cov_total() and not self.options.collectonly:
|
||||
# make sure we get the EXIT_TESTSFAILED exit code
|
||||
compat_session.testsfailed += 1
|
||||
|
||||
def pytest_terminal_summary(self, terminalreporter):
|
||||
if self._disabled:
|
||||
if self.options.no_cov_should_warn:
|
||||
message = 'Coverage disabled via --no-cov switch!'
|
||||
terminalreporter.write('WARNING: %s\n' % message, red=True, bold=True)
|
||||
warnings.warn(CovDisabledWarning(message))
|
||||
return
|
||||
if self.cov_controller is None:
|
||||
return
|
||||
|
||||
if self.cov_total is None:
|
||||
# we shouldn't report, or report generation failed (error raised above)
|
||||
return
|
||||
|
||||
report = self.cov_report.getvalue()
|
||||
|
||||
# Avoid undesirable new lines when output is disabled with "--cov-report=".
|
||||
if report:
|
||||
terminalreporter.write('\n' + report + '\n')
|
||||
|
||||
if self.options.cov_fail_under is not None and self.options.cov_fail_under > 0:
|
||||
failed = self.cov_total < self.options.cov_fail_under
|
||||
markup = {'red': True, 'bold': True} if failed else {'green': True}
|
||||
message = (
|
||||
'{fail}Required test coverage of {required}% {reached}. '
|
||||
'Total coverage: {actual:.2f}%\n'
|
||||
.format(
|
||||
required=self.options.cov_fail_under,
|
||||
actual=self.cov_total,
|
||||
fail="FAIL " if failed else "",
|
||||
reached="not reached" if failed else "reached"
|
||||
)
|
||||
)
|
||||
terminalreporter.write(message, **markup)
|
||||
|
||||
def pytest_runtest_setup(self, item):
|
||||
if os.getpid() != self.pid:
|
||||
# test is run in another process than session, run
|
||||
# coverage manually
|
||||
embed.init()
|
||||
|
||||
def pytest_runtest_teardown(self, item):
|
||||
embed.cleanup()
|
||||
|
||||
@pytest.hookimpl(hookwrapper=True)
|
||||
def pytest_runtest_call(self, item):
|
||||
if (item.get_closest_marker('no_cover')
|
||||
or 'no_cover' in getattr(item, 'fixturenames', ())):
|
||||
self.cov_controller.pause()
|
||||
yield
|
||||
self.cov_controller.resume()
|
||||
else:
|
||||
yield
|
||||
|
||||
|
||||
class TestContextPlugin:
|
||||
def __init__(self, cov):
|
||||
self.cov = cov
|
||||
|
||||
def pytest_runtest_setup(self, item):
|
||||
self.switch_context(item, 'setup')
|
||||
|
||||
def pytest_runtest_teardown(self, item):
|
||||
self.switch_context(item, 'teardown')
|
||||
|
||||
def pytest_runtest_call(self, item):
|
||||
self.switch_context(item, 'run')
|
||||
|
||||
def switch_context(self, item, when):
|
||||
context = f"{item.nodeid}|{when}"
|
||||
self.cov.switch_context(context)
|
||||
os.environ['COV_CORE_CONTEXT'] = context
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_cover():
|
||||
"""A pytest fixture to disable coverage."""
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cov(request):
|
||||
"""A pytest fixture to provide access to the underlying coverage object."""
|
||||
|
||||
# Check with hasplugin to avoid getplugin exception in older pytest.
|
||||
if request.config.pluginmanager.hasplugin('_cov'):
|
||||
plugin = request.config.pluginmanager.getplugin('_cov')
|
||||
if plugin.cov_controller:
|
||||
return plugin.cov_controller.cov
|
||||
return None
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
config.addinivalue_line("markers", "no_cover: disable coverage for this test.")
|
||||
Reference in New Issue
Block a user