Coverage for /usr/local/lib/python3.7/site-packages/pytest_cov/plugin.py : 9%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Coverage plugin for pytest."""
2import argparse
3import os
4import warnings
6import coverage
7import pytest
8from coverage.misc import CoverageException
10from . import compat
11from . import embed
12from . import engine
15class CoverageError(Exception):
16 """Indicates that our coverage is too low"""
19def validate_report(arg):
20 file_choices = ['annotate', 'html', 'xml']
21 term_choices = ['term', 'term-missing']
22 term_modifier_choices = ['skip-covered']
23 all_choices = term_choices + file_choices
24 values = arg.split(":", 1)
25 report_type = values[0]
26 if report_type not in all_choices + ['']:
27 msg = 'invalid choice: "{}" (choose from "{}")'.format(arg, all_choices)
28 raise argparse.ArgumentTypeError(msg)
30 if len(values) == 1:
31 return report_type, None
33 report_modifier = values[1]
34 if report_type in term_choices and report_modifier in term_modifier_choices:
35 return report_type, report_modifier
37 if report_type not in file_choices:
38 msg = 'output specifier not supported for: "{}" (choose from "{}")'.format(arg,
39 file_choices)
40 raise argparse.ArgumentTypeError(msg)
42 return values
45def validate_fail_under(num_str):
46 try:
47 return int(num_str)
48 except ValueError:
49 return float(num_str)
52def validate_context(arg):
53 if coverage.version_info <= (5, 0):
54 raise argparse.ArgumentTypeError('Contexts are only supported with coverage.py >= 5.x')
55 if arg != "test":
56 raise argparse.ArgumentTypeError('--cov-context=test is the only supported value')
57 return arg
60class StoreReport(argparse.Action):
61 def __call__(self, parser, namespace, values, option_string=None):
62 report_type, file = values
63 namespace.cov_report[report_type] = file
66def pytest_addoption(parser):
67 """Add options to control coverage."""
69 group = parser.getgroup(
70 'cov', 'coverage reporting with distributed testing support')
71 group.addoption('--cov', action='append', default=[], metavar='SOURCE',
72 nargs='?', const=True, dest='cov_source',
73 help='Path or package name to measure during execution (multi-allowed). '
74 'Use --cov= to not do any source filtering and record everything.')
75 group.addoption('--cov-report', action=StoreReport, default={},
76 metavar='TYPE', type=validate_report,
77 help='Type of report to generate: term, term-missing, '
78 'annotate, html, xml (multi-allowed). '
79 'term, term-missing may be followed by ":skip-covered". '
80 'annotate, html and xml may be followed by ":DEST" '
81 'where DEST specifies the output location. '
82 'Use --cov-report= to not generate any output.')
83 group.addoption('--cov-config', action='store', default='.coveragerc',
84 metavar='PATH',
85 help='Config file for coverage. Default: .coveragerc')
86 group.addoption('--no-cov-on-fail', action='store_true', default=False,
87 help='Do not report coverage if test run fails. '
88 'Default: False')
89 group.addoption('--no-cov', action='store_true', default=False,
90 help='Disable coverage report completely (useful for debuggers). '
91 'Default: False')
92 group.addoption('--cov-fail-under', action='store', metavar='MIN',
93 type=validate_fail_under,
94 help='Fail if the total coverage is less than MIN.')
95 group.addoption('--cov-append', action='store_true', default=False,
96 help='Do not delete coverage but append to current. '
97 'Default: False')
98 group.addoption('--cov-branch', action='store_true', default=None,
99 help='Enable branch coverage.')
100 group.addoption('--cov-context', action='store', metavar='CONTEXT',
101 type=validate_context,
102 help='Dynamic contexts to use. "test" for now.')
105def _prepare_cov_source(cov_source):
106 """
107 Prepare cov_source so that:
109 --cov --cov=foobar is equivalent to --cov (cov_source=None)
110 --cov=foo --cov=bar is equivalent to cov_source=['foo', 'bar']
111 """
112 return None if True in cov_source else [path for path in cov_source if path is not True]
115@pytest.mark.tryfirst
116def pytest_load_initial_conftests(early_config, parser, args):
117 if early_config.known_args_namespace.cov_source:
118 plugin = CovPlugin(early_config.known_args_namespace, early_config.pluginmanager)
119 early_config.pluginmanager.register(plugin, '_cov')
122class CovPlugin(object):
123 """Use coverage package to produce code coverage reports.
125 Delegates all work to a particular implementation based on whether
126 this test process is centralised, a distributed master or a
127 distributed worker.
128 """
130 def __init__(self, options, pluginmanager, start=True):
131 """Creates a coverage pytest plugin.
133 We read the rc file that coverage uses to get the data file
134 name. This is needed since we give coverage through it's API
135 the data file name.
136 """
138 # Our implementation is unknown at this time.
139 self.pid = None
140 self.cov_controller = None
141 self.cov_report = compat.StringIO()
142 self.cov_total = None
143 self.failed = False
144 self._started = False
145 self._disabled = False
146 self.options = options
148 is_dist = (getattr(options, 'numprocesses', False) or
149 getattr(options, 'distload', False) or
150 getattr(options, 'dist', 'no') != 'no')
151 if getattr(options, 'no_cov', False):
152 self._disabled = True
153 return
155 if not self.options.cov_report:
156 self.options.cov_report = ['term']
157 elif len(self.options.cov_report) == 1 and '' in self.options.cov_report:
158 self.options.cov_report = {}
159 self.options.cov_source = _prepare_cov_source(self.options.cov_source)
161 if is_dist and start:
162 self.start(engine.DistMaster)
163 elif start:
164 self.start(engine.Central)
166 # worker is started in pytest hook
168 def start(self, controller_cls, config=None, nodeid=None):
170 if config is None:
171 # fake config option for engine
172 class Config(object):
173 option = self.options
175 config = Config()
177 self.cov_controller = controller_cls(
178 self.options.cov_source,
179 self.options.cov_report,
180 self.options.cov_config,
181 self.options.cov_append,
182 self.options.cov_branch,
183 config,
184 nodeid
185 )
186 self.cov_controller.start()
187 self._started = True
188 cov_config = self.cov_controller.cov.config
189 if self.options.cov_fail_under is None and hasattr(cov_config, 'fail_under'):
190 self.options.cov_fail_under = cov_config.fail_under
192 def _is_worker(self, session):
193 return compat.workerinput(session.config, None) is not None
195 def pytest_sessionstart(self, session):
196 """At session start determine our implementation and delegate to it."""
198 if self.options.no_cov:
199 # Coverage can be disabled because it does not cooperate with debuggers well.
200 self._disabled = True
201 return
203 self.pid = os.getpid()
204 if self._is_worker(session):
205 nodeid = (
206 compat.workerinput(session.config)
207 .get(compat.workerid, getattr(session, 'nodeid'))
208 )
209 self.start(engine.DistWorker, session.config, nodeid)
210 elif not self._started:
211 self.start(engine.Central)
213 if self.options.cov_context == 'test':
214 session.config.pluginmanager.register(TestContextPlugin(self.cov_controller.cov), '_cov_contexts')
216 def pytest_configure_node(self, node):
217 """Delegate to our implementation.
219 Mark this hook as optional in case xdist is not installed.
220 """
221 if not self._disabled:
222 self.cov_controller.configure_node(node)
223 pytest_configure_node.optionalhook = True
225 def pytest_testnodedown(self, node, error):
226 """Delegate to our implementation.
228 Mark this hook as optional in case xdist is not installed.
229 """
230 if not self._disabled:
231 self.cov_controller.testnodedown(node, error)
232 pytest_testnodedown.optionalhook = True
234 def _should_report(self):
235 return not (self.failed and self.options.no_cov_on_fail)
237 def _failed_cov_total(self):
238 cov_fail_under = self.options.cov_fail_under
239 return cov_fail_under is not None and self.cov_total < cov_fail_under
241 # we need to wrap pytest_runtestloop. by the time pytest_sessionfinish
242 # runs, it's too late to set testsfailed
243 @compat.hookwrapper
244 def pytest_runtestloop(self, session):
245 yield
247 if self._disabled:
248 return
250 compat_session = compat.SessionWrapper(session)
252 self.failed = bool(compat_session.testsfailed)
253 if self.cov_controller is not None:
254 self.cov_controller.finish()
256 if not self._is_worker(session) and self._should_report():
257 try:
258 self.cov_total = self.cov_controller.summary(self.cov_report)
259 except CoverageException as exc:
260 message = 'Failed to generate report: %s\n' % exc
261 session.config.pluginmanager.getplugin("terminalreporter").write(
262 'WARNING: %s\n' % message, red=True, bold=True)
263 if pytest.__version__ >= '3.8':
264 warnings.warn(pytest.PytestWarning(message))
265 else:
266 session.config.warn(code='COV-2', message=message)
267 self.cov_total = 0
268 assert self.cov_total is not None, 'Test coverage should never be `None`'
269 if self._failed_cov_total():
270 # make sure we get the EXIT_TESTSFAILED exit code
271 compat_session.testsfailed += 1
273 def pytest_terminal_summary(self, terminalreporter):
274 if self._disabled:
275 message = 'Coverage disabled via --no-cov switch!'
276 terminalreporter.write('WARNING: %s\n' % message, red=True, bold=True)
277 if pytest.__version__ >= '3.8':
278 warnings.warn(pytest.PytestWarning(message))
279 else:
280 terminalreporter.config.warn(code='COV-1', message=message)
281 return
282 if self.cov_controller is None:
283 return
285 if self.cov_total is None:
286 # we shouldn't report, or report generation failed (error raised above)
287 return
289 terminalreporter.write('\n' + self.cov_report.getvalue() + '\n')
291 if self.options.cov_fail_under is not None and self.options.cov_fail_under > 0:
292 failed = self.cov_total < self.options.cov_fail_under
293 markup = {'red': True, 'bold': True} if failed else {'green': True}
294 message = (
295 '{fail}Required test coverage of {required}% {reached}. '
296 'Total coverage: {actual:.2f}%\n'
297 .format(
298 required=self.options.cov_fail_under,
299 actual=self.cov_total,
300 fail="FAIL " if failed else "",
301 reached="not reached" if failed else "reached"
302 )
303 )
304 terminalreporter.write(message, **markup)
306 def pytest_runtest_setup(self, item):
307 if os.getpid() != self.pid:
308 # test is run in another process than session, run
309 # coverage manually
310 embed.init()
312 def pytest_runtest_teardown(self, item):
313 embed.cleanup()
315 @compat.hookwrapper
316 def pytest_runtest_call(self, item):
317 if (item.get_closest_marker('no_cover')
318 or 'no_cover' in getattr(item, 'fixturenames', ())):
319 self.cov_controller.pause()
320 yield
321 self.cov_controller.resume()
322 else:
323 yield
326class TestContextPlugin(object):
327 def __init__(self, cov):
328 self.cov = cov
330 def pytest_runtest_setup(self, item):
331 self.switch_context(item, 'setup')
333 def pytest_runtest_teardown(self, item):
334 self.switch_context(item, 'teardown')
336 def pytest_runtest_call(self, item):
337 self.switch_context(item, 'run')
339 def switch_context(self, item, when):
340 context = "{item.nodeid}|{when}".format(item=item, when=when)
341 self.cov.switch_context(context)
344@pytest.fixture
345def no_cover():
346 """A pytest fixture to disable coverage."""
347 pass
350@pytest.fixture
351def cov(request):
352 """A pytest fixture to provide access to the underlying coverage object."""
354 # Check with hasplugin to avoid getplugin exception in older pytest.
355 if request.config.pluginmanager.hasplugin('_cov'):
356 plugin = request.config.pluginmanager.getplugin('_cov')
357 if plugin.cov_controller:
358 return plugin.cov_controller.cov
359 return None
362def pytest_configure(config):
363 config.addinivalue_line("markers", "no_cover: disable coverage for this test.")