updates
This commit is contained in:
Iliyan Angelov
468
Backend/venv/lib/python3.12/site-packages/pytest_cov/plugin.py
Normal file
468
Backend/venv/lib/python3.12/site-packages/pytest_cov/plugin.py
Normal file
@@ -0,0 +1,468 @@
|
||||
"""Coverage plugin for pytest."""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import warnings
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
from . import CovDisabledWarning
|
||||
from . import CovReportWarning
|
||||
from . import PytestCovWarning
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .engine import CovController
|
||||
|
||||
COVERAGE_SQLITE_WARNING_RE = re.compile('unclosed database in <sqlite3.Connection object at', re.I)
|
||||
|
||||
|
||||
def validate_report(arg):
|
||||
file_choices = ['annotate', 'html', 'xml', 'json', 'markdown', 'markdown-append', '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 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 = f'output specifier not supported for: "{arg}" (choose from "{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.') from None
|
||||
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 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
|
||||
|
||||
# coverage.py doesn't set a default file for markdown output_format
|
||||
if report_type in ['markdown', 'markdown-append'] and file is None:
|
||||
namespace.cov_report[report_type] = 'coverage.md'
|
||||
if all(x in namespace.cov_report for x in ['markdown', 'markdown-append']):
|
||||
self._validate_markdown_dest_files(namespace.cov_report, parser)
|
||||
|
||||
def _validate_markdown_dest_files(self, cov_report_options, parser):
|
||||
markdown_file = cov_report_options['markdown']
|
||||
markdown_append_file = cov_report_options['markdown-append']
|
||||
if markdown_file == markdown_append_file:
|
||||
error_message = f"markdown and markdown-append options cannot point to the same file: '{markdown_file}'."
|
||||
error_message += ' Please redirect one of them using :DEST (e.g. --cov-report=markdown:dest_file.md)'
|
||||
parser.error(error_message)
|
||||
|
||||
|
||||
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, json, markdown, markdown-append, lcov (multi-allowed). '
|
||||
'term, term-missing may be followed by ":skip-covered". '
|
||||
'annotate, html, xml, json, markdown, markdown-append 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-precision',
|
||||
type=int,
|
||||
default=None,
|
||||
help='Override the reporting precision.',
|
||||
)
|
||||
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: argparse.Namespace, 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 = StringIO()
|
||||
self.cov_total = None
|
||||
self.failed = False
|
||||
self._started = False
|
||||
self._start_path = None
|
||||
self._disabled = False
|
||||
self.options = options
|
||||
self._wrote_heading = False
|
||||
|
||||
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: type['CovController'], 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, config, nodeid)
|
||||
self.cov_controller.start()
|
||||
self._started = True
|
||||
self._start_path = Path.cwd()
|
||||
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
|
||||
if self.options.cov_precision is None:
|
||||
self.options.cov_precision = getattr(cov_config, 'precision', 0)
|
||||
|
||||
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', 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_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):
|
||||
needed = self.options.cov_report or self.options.cov_fail_under
|
||||
return needed and not (self.failed and self.options.no_cov_on_fail)
|
||||
|
||||
# we need to wrap pytest_runtestloop. by the time pytest_sessionfinish
|
||||
# runs, it's too late to set testsfailed
|
||||
@pytest.hookimpl(wrapper=True)
|
||||
def pytest_runtestloop(self, session):
|
||||
if self._disabled:
|
||||
return (yield)
|
||||
|
||||
# we add default warning configuration to prevent certain warnings to bubble up as errors due to rigid filterwarnings configuration
|
||||
for _, message, category, _, _ in warnings.filters:
|
||||
if category is ResourceWarning and message == COVERAGE_SQLITE_WARNING_RE:
|
||||
break
|
||||
else:
|
||||
warnings.filterwarnings('default', 'unclosed database in <sqlite3.Connection object at', ResourceWarning)
|
||||
for _, _, category, _, _ in warnings.filters:
|
||||
if category is PytestCovWarning:
|
||||
break
|
||||
else:
|
||||
warnings.simplefilter('once', PytestCovWarning)
|
||||
from coverage.exceptions import CoverageWarning
|
||||
|
||||
for _, _, category, _, _ in warnings.filters:
|
||||
if category is CoverageWarning:
|
||||
break
|
||||
else:
|
||||
warnings.simplefilter('once', CoverageWarning)
|
||||
|
||||
result = yield
|
||||
|
||||
self.failed = bool(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
|
||||
from coverage.results import display_covered
|
||||
from coverage.results import should_fail_under
|
||||
|
||||
try:
|
||||
self.cov_total = self.cov_controller.summary(self.cov_report)
|
||||
except CoverageException as exc:
|
||||
message = f'Failed to generate report: {exc}\n'
|
||||
session.config.pluginmanager.getplugin('terminalreporter').write(f'\nWARNING: {message}\n', red=True, bold=True)
|
||||
warnings.warn(CovReportWarning(message), stacklevel=1)
|
||||
self.cov_total = 0
|
||||
assert self.cov_total is not None, 'Test coverage should never be `None`'
|
||||
cov_fail_under = self.options.cov_fail_under
|
||||
cov_precision = self.options.cov_precision
|
||||
if cov_fail_under is None or self.options.collectonly:
|
||||
return
|
||||
if should_fail_under(self.cov_total, cov_fail_under, cov_precision):
|
||||
message = 'Coverage failure: total of {total} is less than fail-under={fail_under:.{p}f}'.format(
|
||||
total=display_covered(self.cov_total, cov_precision),
|
||||
fail_under=cov_fail_under,
|
||||
p=cov_precision,
|
||||
)
|
||||
session.config.pluginmanager.getplugin('terminalreporter').write(f'\nERROR: {message}\n', red=True, bold=True)
|
||||
# make sure we get the EXIT_TESTSFAILED exit code
|
||||
session.testsfailed += 1
|
||||
|
||||
return result
|
||||
|
||||
def write_heading(self, terminalreporter):
|
||||
if not self._wrote_heading:
|
||||
terminalreporter.write_sep('=', 'tests coverage')
|
||||
self._wrote_heading = True
|
||||
|
||||
def pytest_terminal_summary(self, terminalreporter):
|
||||
if self._disabled:
|
||||
if self.options.no_cov_should_warn:
|
||||
self.write_heading(terminalreporter)
|
||||
message = 'Coverage disabled via --no-cov switch!'
|
||||
terminalreporter.write(f'WARNING: {message}\n', red=True, bold=True)
|
||||
warnings.warn(CovDisabledWarning(message), stacklevel=1)
|
||||
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()
|
||||
|
||||
if report:
|
||||
self.write_heading(terminalreporter)
|
||||
terminalreporter.write(report)
|
||||
|
||||
if self.options.cov_fail_under is not None and self.options.cov_fail_under > 0:
|
||||
self.write_heading(terminalreporter)
|
||||
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)
|
||||
|
||||
@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:
|
||||
cov_controller: 'CovController'
|
||||
|
||||
def __init__(self, cov_controller):
|
||||
self.cov_controller = cov_controller
|
||||
|
||||
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):
|
||||
if self.cov_controller.started:
|
||||
self.cov_controller.cov.switch_context(f'{item.nodeid}|{when}')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def no_cover():
|
||||
"""A pytest fixture to disable coverage."""
|
||||
|
||||
|
||||
@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