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

2merged implementation of the cache provider 

3 

4the name cache was not chosen to ensure pluggy automatically 

5ignores the external pytest-cache 

6""" 

7import json 

8import os 

9from collections import OrderedDict 

10from typing import List 

11 

12import attr 

13import py 

14 

15import pytest 

16from .pathlib import Path 

17from .pathlib import resolve_from_str 

18from .pathlib import rm_rf 

19from _pytest import nodes 

20from _pytest.config import Config 

21from _pytest.main import Session 

22 

23README_CONTENT = """\ 

24# pytest cache directory # 

25 

26This directory contains data from the pytest's cache plugin, 

27which provides the `--lf` and `--ff` options, as well as the `cache` fixture. 

28 

29**Do not** commit this to version control. 

30 

31See [the docs](https://docs.pytest.org/en/latest/cache.html) for more information. 

32""" 

33 

34CACHEDIR_TAG_CONTENT = b"""\ 

35Signature: 8a477f597d28d172789f06886806bc55 

36# This file is a cache directory tag created by pytest. 

37# For information about cache directory tags, see: 

38# http://www.bford.info/cachedir/spec.html 

39""" 

40 

41 

42@attr.s 

43class Cache: 

44 _cachedir = attr.ib(repr=False) 

45 _config = attr.ib(repr=False) 

46 

47 # sub-directory under cache-dir for directories created by "makedir" 

48 _CACHE_PREFIX_DIRS = "d" 

49 

50 # sub-directory under cache-dir for values created by "set" 

51 _CACHE_PREFIX_VALUES = "v" 

52 

53 @classmethod 

54 def for_config(cls, config): 

55 cachedir = cls.cache_dir_from_config(config) 

56 if config.getoption("cacheclear") and cachedir.is_dir(): 

57 cls.clear_cache(cachedir) 

58 return cls(cachedir, config) 

59 

60 @classmethod 

61 def clear_cache(cls, cachedir: Path): 

62 """Clears the sub-directories used to hold cached directories and values.""" 

63 for prefix in (cls._CACHE_PREFIX_DIRS, cls._CACHE_PREFIX_VALUES): 

64 d = cachedir / prefix 

65 if d.is_dir(): 

66 rm_rf(d) 

67 

68 @staticmethod 

69 def cache_dir_from_config(config): 

70 return resolve_from_str(config.getini("cache_dir"), config.rootdir) 

71 

72 def warn(self, fmt, **args): 

73 from _pytest.warnings import _issue_warning_captured 

74 from _pytest.warning_types import PytestCacheWarning 

75 

76 _issue_warning_captured( 

77 PytestCacheWarning(fmt.format(**args) if args else fmt), 

78 self._config.hook, 

79 stacklevel=3, 

80 ) 

81 

82 def makedir(self, name): 

83 """ return a directory path object with the given name. If the 

84 directory does not yet exist, it will be created. You can use it 

85 to manage files likes e. g. store/retrieve database 

86 dumps across test sessions. 

87 

88 :param name: must be a string not containing a ``/`` separator. 

89 Make sure the name contains your plugin or application 

90 identifiers to prevent clashes with other cache users. 

91 """ 

92 name = Path(name) 

93 if len(name.parts) > 1: 

94 raise ValueError("name is not allowed to contain path separators") 

95 res = self._cachedir.joinpath(self._CACHE_PREFIX_DIRS, name) 

96 res.mkdir(exist_ok=True, parents=True) 

97 return py.path.local(res) 

98 

99 def _getvaluepath(self, key): 

100 return self._cachedir.joinpath(self._CACHE_PREFIX_VALUES, Path(key)) 

101 

102 def get(self, key, default): 

103 """ return cached value for the given key. If no value 

104 was yet cached or the value cannot be read, the specified 

105 default is returned. 

106 

107 :param key: must be a ``/`` separated value. Usually the first 

108 name is the name of your plugin or your application. 

109 :param default: must be provided in case of a cache-miss or 

110 invalid cache values. 

111 

112 """ 

113 path = self._getvaluepath(key) 

114 try: 

115 with path.open("r") as f: 

116 return json.load(f) 

117 except (ValueError, IOError, OSError): 

118 return default 

119 

120 def set(self, key, value): 

121 """ save value for the given key. 

122 

123 :param key: must be a ``/`` separated value. Usually the first 

124 name is the name of your plugin or your application. 

125 :param value: must be of any combination of basic 

126 python types, including nested types 

127 like e. g. lists of dictionaries. 

128 """ 

129 path = self._getvaluepath(key) 

130 try: 

131 if path.parent.is_dir(): 

132 cache_dir_exists_already = True 

133 else: 

134 cache_dir_exists_already = self._cachedir.exists() 

135 path.parent.mkdir(exist_ok=True, parents=True) 

136 except (IOError, OSError): 

137 self.warn("could not create cache path {path}", path=path) 

138 return 

139 if not cache_dir_exists_already: 

140 self._ensure_supporting_files() 

141 data = json.dumps(value, indent=2, sort_keys=True) 

142 try: 

143 f = path.open("w") 

144 except (IOError, OSError): 

145 self.warn("cache could not write path {path}", path=path) 

146 else: 

147 with f: 

148 f.write(data) 

149 

150 def _ensure_supporting_files(self): 

151 """Create supporting files in the cache dir that are not really part of the cache.""" 

152 readme_path = self._cachedir / "README.md" 

153 readme_path.write_text(README_CONTENT) 

154 

155 gitignore_path = self._cachedir.joinpath(".gitignore") 

156 msg = "# Created by pytest automatically.\n*\n" 

157 gitignore_path.write_text(msg, encoding="UTF-8") 

158 

159 cachedir_tag_path = self._cachedir.joinpath("CACHEDIR.TAG") 

160 cachedir_tag_path.write_bytes(CACHEDIR_TAG_CONTENT) 

161 

162 

163class LFPlugin: 

164 """ Plugin which implements the --lf (run last-failing) option """ 

165 

166 def __init__(self, config): 

167 self.config = config 

168 active_keys = "lf", "failedfirst" 

169 self.active = any(config.getoption(key) for key in active_keys) 

170 self.lastfailed = config.cache.get("cache/lastfailed", {}) 

171 self._previously_failed_count = None 

172 self._report_status = None 

173 self._skipped_files = 0 # count skipped files during collection due to --lf 

174 

175 def last_failed_paths(self): 

176 """Returns a set with all Paths()s of the previously failed nodeids (cached). 

177 """ 

178 try: 

179 return self._last_failed_paths 

180 except AttributeError: 

181 rootpath = Path(self.config.rootdir) 

182 result = {rootpath / nodeid.split("::")[0] for nodeid in self.lastfailed} 

183 result = {x for x in result if x.exists()} 

184 self._last_failed_paths = result 

185 return result 

186 

187 def pytest_ignore_collect(self, path): 

188 """ 

189 Ignore this file path if we are in --lf mode and it is not in the list of 

190 previously failed files. 

191 """ 

192 if self.active and self.config.getoption("lf") and path.isfile(): 

193 last_failed_paths = self.last_failed_paths() 

194 if last_failed_paths: 

195 skip_it = Path(path) not in self.last_failed_paths() 

196 if skip_it: 

197 self._skipped_files += 1 

198 return skip_it 

199 

200 def pytest_report_collectionfinish(self): 

201 if self.active and self.config.getoption("verbose") >= 0: 

202 return "run-last-failure: %s" % self._report_status 

203 

204 def pytest_runtest_logreport(self, report): 

205 if (report.when == "call" and report.passed) or report.skipped: 

206 self.lastfailed.pop(report.nodeid, None) 

207 elif report.failed: 

208 self.lastfailed[report.nodeid] = True 

209 

210 def pytest_collectreport(self, report): 

211 passed = report.outcome in ("passed", "skipped") 

212 if passed: 

213 if report.nodeid in self.lastfailed: 

214 self.lastfailed.pop(report.nodeid) 

215 self.lastfailed.update((item.nodeid, True) for item in report.result) 

216 else: 

217 self.lastfailed[report.nodeid] = True 

218 

219 def pytest_collection_modifyitems(self, session, config, items): 

220 if not self.active: 

221 return 

222 

223 if self.lastfailed: 

224 previously_failed = [] 

225 previously_passed = [] 

226 for item in items: 

227 if item.nodeid in self.lastfailed: 

228 previously_failed.append(item) 

229 else: 

230 previously_passed.append(item) 

231 self._previously_failed_count = len(previously_failed) 

232 

233 if not previously_failed: 

234 # Running a subset of all tests with recorded failures 

235 # only outside of it. 

236 self._report_status = "%d known failures not in selected tests" % ( 

237 len(self.lastfailed), 

238 ) 

239 else: 

240 if self.config.getoption("lf"): 

241 items[:] = previously_failed 

242 config.hook.pytest_deselected(items=previously_passed) 

243 else: # --failedfirst 

244 items[:] = previously_failed + previously_passed 

245 

246 noun = "failure" if self._previously_failed_count == 1 else "failures" 

247 suffix = " first" if self.config.getoption("failedfirst") else "" 

248 self._report_status = "rerun previous {count} {noun}{suffix}".format( 

249 count=self._previously_failed_count, suffix=suffix, noun=noun 

250 ) 

251 

252 if self._skipped_files > 0: 

253 files_noun = "file" if self._skipped_files == 1 else "files" 

254 self._report_status += " (skipped {files} {files_noun})".format( 

255 files=self._skipped_files, files_noun=files_noun 

256 ) 

257 else: 

258 self._report_status = "no previously failed tests, " 

259 if self.config.getoption("last_failed_no_failures") == "none": 

260 self._report_status += "deselecting all items." 

261 config.hook.pytest_deselected(items=items) 

262 items[:] = [] 

263 else: 

264 self._report_status += "not deselecting items." 

265 

266 def pytest_sessionfinish(self, session): 

267 config = self.config 

268 if config.getoption("cacheshow") or hasattr(config, "slaveinput"): 

269 return 

270 

271 saved_lastfailed = config.cache.get("cache/lastfailed", {}) 

272 if saved_lastfailed != self.lastfailed: 

273 config.cache.set("cache/lastfailed", self.lastfailed) 

274 

275 

276class NFPlugin: 

277 """ Plugin which implements the --nf (run new-first) option """ 

278 

279 def __init__(self, config): 

280 self.config = config 

281 self.active = config.option.newfirst 

282 self.cached_nodeids = config.cache.get("cache/nodeids", []) 

283 

284 def pytest_collection_modifyitems( 

285 self, session: Session, config: Config, items: List[nodes.Item] 

286 ) -> None: 

287 new_items = OrderedDict() # type: OrderedDict[str, nodes.Item] 

288 if self.active: 

289 other_items = OrderedDict() # type: OrderedDict[str, nodes.Item] 

290 for item in items: 

291 if item.nodeid not in self.cached_nodeids: 

292 new_items[item.nodeid] = item 

293 else: 

294 other_items[item.nodeid] = item 

295 

296 items[:] = self._get_increasing_order( 

297 new_items.values() 

298 ) + self._get_increasing_order(other_items.values()) 

299 else: 

300 for item in items: 

301 if item.nodeid not in self.cached_nodeids: 

302 new_items[item.nodeid] = item 

303 self.cached_nodeids.extend(new_items) 

304 

305 def _get_increasing_order(self, items): 

306 return sorted(items, key=lambda item: item.fspath.mtime(), reverse=True) 

307 

308 def pytest_sessionfinish(self, session): 

309 config = self.config 

310 if config.getoption("cacheshow") or hasattr(config, "slaveinput"): 

311 return 

312 

313 config.cache.set("cache/nodeids", self.cached_nodeids) 

314 

315 

316def pytest_addoption(parser): 

317 group = parser.getgroup("general") 

318 group.addoption( 

319 "--lf", 

320 "--last-failed", 

321 action="store_true", 

322 dest="lf", 

323 help="rerun only the tests that failed " 

324 "at the last run (or all if none failed)", 

325 ) 

326 group.addoption( 

327 "--ff", 

328 "--failed-first", 

329 action="store_true", 

330 dest="failedfirst", 

331 help="run all tests but run the last failures first. " 

332 "This may re-order tests and thus lead to " 

333 "repeated fixture setup/teardown", 

334 ) 

335 group.addoption( 

336 "--nf", 

337 "--new-first", 

338 action="store_true", 

339 dest="newfirst", 

340 help="run tests from new files first, then the rest of the tests " 

341 "sorted by file mtime", 

342 ) 

343 group.addoption( 

344 "--cache-show", 

345 action="append", 

346 nargs="?", 

347 dest="cacheshow", 

348 help=( 

349 "show cache contents, don't perform collection or tests. " 

350 "Optional argument: glob (default: '*')." 

351 ), 

352 ) 

353 group.addoption( 

354 "--cache-clear", 

355 action="store_true", 

356 dest="cacheclear", 

357 help="remove all cache contents at start of test run.", 

358 ) 

359 cache_dir_default = ".pytest_cache" 

360 if "TOX_ENV_DIR" in os.environ: 

361 cache_dir_default = os.path.join(os.environ["TOX_ENV_DIR"], cache_dir_default) 

362 parser.addini("cache_dir", default=cache_dir_default, help="cache directory path.") 

363 group.addoption( 

364 "--lfnf", 

365 "--last-failed-no-failures", 

366 action="store", 

367 dest="last_failed_no_failures", 

368 choices=("all", "none"), 

369 default="all", 

370 help="which tests to run with no previously (known) failures.", 

371 ) 

372 

373 

374def pytest_cmdline_main(config): 

375 if config.option.cacheshow: 

376 from _pytest.main import wrap_session 

377 

378 return wrap_session(config, cacheshow) 

379 

380 

381@pytest.hookimpl(tryfirst=True) 

382def pytest_configure(config): 

383 config.cache = Cache.for_config(config) 

384 config.pluginmanager.register(LFPlugin(config), "lfplugin") 

385 config.pluginmanager.register(NFPlugin(config), "nfplugin") 

386 

387 

388@pytest.fixture 

389def cache(request): 

390 """ 

391 Return a cache object that can persist state between testing sessions. 

392 

393 cache.get(key, default) 

394 cache.set(key, value) 

395 

396 Keys must be a ``/`` separated value, where the first part is usually the 

397 name of your plugin or application to avoid clashes with other cache users. 

398 

399 Values can be any object handled by the json stdlib module. 

400 """ 

401 return request.config.cache 

402 

403 

404def pytest_report_header(config): 

405 """Display cachedir with --cache-show and if non-default.""" 

406 if config.option.verbose > 0 or config.getini("cache_dir") != ".pytest_cache": 

407 cachedir = config.cache._cachedir 

408 # TODO: evaluate generating upward relative paths 

409 # starting with .., ../.. if sensible 

410 

411 try: 

412 displaypath = cachedir.relative_to(config.rootdir) 

413 except ValueError: 

414 displaypath = cachedir 

415 return "cachedir: {}".format(displaypath) 

416 

417 

418def cacheshow(config, session): 

419 from pprint import pformat 

420 

421 tw = py.io.TerminalWriter() 

422 tw.line("cachedir: " + str(config.cache._cachedir)) 

423 if not config.cache._cachedir.is_dir(): 

424 tw.line("cache is empty") 

425 return 0 

426 

427 glob = config.option.cacheshow[0] 

428 if glob is None: 

429 glob = "*" 

430 

431 dummy = object() 

432 basedir = config.cache._cachedir 

433 vdir = basedir / Cache._CACHE_PREFIX_VALUES 

434 tw.sep("-", "cache values for %r" % glob) 

435 for valpath in sorted(x for x in vdir.rglob(glob) if x.is_file()): 

436 key = valpath.relative_to(vdir) 

437 val = config.cache.get(key, dummy) 

438 if val is dummy: 

439 tw.line("%s contains unreadable content, will be ignored" % key) 

440 else: 

441 tw.line("%s contains:" % key) 

442 for line in pformat(val).splitlines(): 

443 tw.line(" " + line) 

444 

445 ddir = basedir / Cache._CACHE_PREFIX_DIRS 

446 if ddir.is_dir(): 

447 contents = sorted(ddir.rglob(glob)) 

448 tw.sep("-", "cache directories for %r" % glob) 

449 for p in contents: 

450 # if p.check(dir=1): 

451 # print("%s/" % p.relto(basedir)) 

452 if p.is_file(): 

453 key = p.relative_to(basedir) 

454 tw.line("{} is a file of length {:d}".format(key, p.stat().st_size)) 

455 return 0