Hide keyboard shortcuts

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 

5 

6import coverage 

7import pytest 

8from coverage.misc import CoverageException 

9 

10from . import compat 

11from . import embed 

12from . import engine 

13 

14 

15class CoverageError(Exception): 

16 """Indicates that our coverage is too low""" 

17 

18 

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) 

29 

30 if len(values) == 1: 

31 return report_type, None 

32 

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 

36 

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) 

41 

42 return values 

43 

44 

45def validate_fail_under(num_str): 

46 try: 

47 return int(num_str) 

48 except ValueError: 

49 return float(num_str) 

50 

51 

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 

58 

59 

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 

64 

65 

66def pytest_addoption(parser): 

67 """Add options to control coverage.""" 

68 

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.') 

103 

104 

105def _prepare_cov_source(cov_source): 

106 """ 

107 Prepare cov_source so that: 

108 

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] 

113 

114 

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') 

120 

121 

122class CovPlugin(object): 

123 """Use coverage package to produce code coverage reports. 

124 

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 """ 

129 

130 def __init__(self, options, pluginmanager, start=True): 

131 """Creates a coverage pytest plugin. 

132 

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 """ 

137 

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 

147 

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 

154 

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) 

160 

161 if is_dist and start: 

162 self.start(engine.DistMaster) 

163 elif start: 

164 self.start(engine.Central) 

165 

166 # worker is started in pytest hook 

167 

168 def start(self, controller_cls, config=None, nodeid=None): 

169 

170 if config is None: 

171 # fake config option for engine 

172 class Config(object): 

173 option = self.options 

174 

175 config = Config() 

176 

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 

191 

192 def _is_worker(self, session): 

193 return compat.workerinput(session.config, None) is not None 

194 

195 def pytest_sessionstart(self, session): 

196 """At session start determine our implementation and delegate to it.""" 

197 

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 

202 

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) 

212 

213 if self.options.cov_context == 'test': 

214 session.config.pluginmanager.register(TestContextPlugin(self.cov_controller.cov), '_cov_contexts') 

215 

216 def pytest_configure_node(self, node): 

217 """Delegate to our implementation. 

218 

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 

224 

225 def pytest_testnodedown(self, node, error): 

226 """Delegate to our implementation. 

227 

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 

233 

234 def _should_report(self): 

235 return not (self.failed and self.options.no_cov_on_fail) 

236 

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 

240 

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 

246 

247 if self._disabled: 

248 return 

249 

250 compat_session = compat.SessionWrapper(session) 

251 

252 self.failed = bool(compat_session.testsfailed) 

253 if self.cov_controller is not None: 

254 self.cov_controller.finish() 

255 

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 

272 

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 

284 

285 if self.cov_total is None: 

286 # we shouldn't report, or report generation failed (error raised above) 

287 return 

288 

289 terminalreporter.write('\n' + self.cov_report.getvalue() + '\n') 

290 

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) 

305 

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() 

311 

312 def pytest_runtest_teardown(self, item): 

313 embed.cleanup() 

314 

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 

324 

325 

326class TestContextPlugin(object): 

327 def __init__(self, cov): 

328 self.cov = cov 

329 

330 def pytest_runtest_setup(self, item): 

331 self.switch_context(item, 'setup') 

332 

333 def pytest_runtest_teardown(self, item): 

334 self.switch_context(item, 'teardown') 

335 

336 def pytest_runtest_call(self, item): 

337 self.switch_context(item, 'run') 

338 

339 def switch_context(self, item, when): 

340 context = "{item.nodeid}|{when}".format(item=item, when=when) 

341 self.cov.switch_context(context) 

342 

343 

344@pytest.fixture 

345def no_cover(): 

346 """A pytest fixture to disable coverage.""" 

347 pass 

348 

349 

350@pytest.fixture 

351def cov(request): 

352 """A pytest fixture to provide access to the underlying coverage object.""" 

353 

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 

360 

361 

362def pytest_configure(config): 

363 config.addinivalue_line("markers", "no_cover: disable coverage for this test.")