Coverage for /usr/local/lib/python3.7/site-packages/_pytest/cacheprovider.py : 16%

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
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
12import attr
13import py
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
23README_CONTENT = """\
24# pytest cache directory #
26This directory contains data from the pytest's cache plugin,
27which provides the `--lf` and `--ff` options, as well as the `cache` fixture.
29**Do not** commit this to version control.
31See [the docs](https://docs.pytest.org/en/latest/cache.html) for more information.
32"""
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"""
42@attr.s
43class Cache:
44 _cachedir = attr.ib(repr=False)
45 _config = attr.ib(repr=False)
47 # sub-directory under cache-dir for directories created by "makedir"
48 _CACHE_PREFIX_DIRS = "d"
50 # sub-directory under cache-dir for values created by "set"
51 _CACHE_PREFIX_VALUES = "v"
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)
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)
68 @staticmethod
69 def cache_dir_from_config(config):
70 return resolve_from_str(config.getini("cache_dir"), config.rootdir)
72 def warn(self, fmt, **args):
73 from _pytest.warnings import _issue_warning_captured
74 from _pytest.warning_types import PytestCacheWarning
76 _issue_warning_captured(
77 PytestCacheWarning(fmt.format(**args) if args else fmt),
78 self._config.hook,
79 stacklevel=3,
80 )
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.
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)
99 def _getvaluepath(self, key):
100 return self._cachedir.joinpath(self._CACHE_PREFIX_VALUES, Path(key))
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.
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.
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
120 def set(self, key, value):
121 """ save value for the given key.
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)
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)
155 gitignore_path = self._cachedir.joinpath(".gitignore")
156 msg = "# Created by pytest automatically.\n*\n"
157 gitignore_path.write_text(msg, encoding="UTF-8")
159 cachedir_tag_path = self._cachedir.joinpath("CACHEDIR.TAG")
160 cachedir_tag_path.write_bytes(CACHEDIR_TAG_CONTENT)
163class LFPlugin:
164 """ Plugin which implements the --lf (run last-failing) option """
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
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
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
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
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
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
219 def pytest_collection_modifyitems(self, session, config, items):
220 if not self.active:
221 return
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)
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
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 )
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."
266 def pytest_sessionfinish(self, session):
267 config = self.config
268 if config.getoption("cacheshow") or hasattr(config, "slaveinput"):
269 return
271 saved_lastfailed = config.cache.get("cache/lastfailed", {})
272 if saved_lastfailed != self.lastfailed:
273 config.cache.set("cache/lastfailed", self.lastfailed)
276class NFPlugin:
277 """ Plugin which implements the --nf (run new-first) option """
279 def __init__(self, config):
280 self.config = config
281 self.active = config.option.newfirst
282 self.cached_nodeids = config.cache.get("cache/nodeids", [])
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
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)
305 def _get_increasing_order(self, items):
306 return sorted(items, key=lambda item: item.fspath.mtime(), reverse=True)
308 def pytest_sessionfinish(self, session):
309 config = self.config
310 if config.getoption("cacheshow") or hasattr(config, "slaveinput"):
311 return
313 config.cache.set("cache/nodeids", self.cached_nodeids)
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 )
374def pytest_cmdline_main(config):
375 if config.option.cacheshow:
376 from _pytest.main import wrap_session
378 return wrap_session(config, cacheshow)
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")
388@pytest.fixture
389def cache(request):
390 """
391 Return a cache object that can persist state between testing sessions.
393 cache.get(key, default)
394 cache.set(key, value)
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.
399 Values can be any object handled by the json stdlib module.
400 """
401 return request.config.cache
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
411 try:
412 displaypath = cachedir.relative_to(config.rootdir)
413 except ValueError:
414 displaypath = cachedir
415 return "cachedir: {}".format(displaypath)
418def cacheshow(config, session):
419 from pprint import pformat
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
427 glob = config.option.cacheshow[0]
428 if glob is None:
429 glob = "*"
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)
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