Plugin which implements the --lf (run last-failing) option.
| 280 | |
| 281 | |
| 282 | class LFPlugin: |
| 283 | """Plugin which implements the --lf (run last-failing) option.""" |
| 284 | |
| 285 | def __init__(self, config: Config) -> None: |
| 286 | self.config = config |
| 287 | active_keys = "lf", "failedfirst" |
| 288 | self.active = any(config.getoption(key) for key in active_keys) |
| 289 | assert config.cache |
| 290 | self.lastfailed: Dict[str, bool] = config.cache.get("cache/lastfailed", {}) |
| 291 | self._previously_failed_count: Optional[int] = None |
| 292 | self._report_status: Optional[str] = None |
| 293 | self._skipped_files = 0 # count skipped files during collection due to --lf |
| 294 | |
| 295 | if config.getoption("lf"): |
| 296 | self._last_failed_paths = self.get_last_failed_paths() |
| 297 | config.pluginmanager.register( |
| 298 | LFPluginCollWrapper(self), "lfplugin-collwrapper" |
| 299 | ) |
| 300 | |
| 301 | def get_last_failed_paths(self) -> Set[Path]: |
| 302 | """Return a set with all Paths()s of the previously failed nodeids.""" |
| 303 | rootpath = self.config.rootpath |
| 304 | result = {rootpath / nodeid.split("::")[0] for nodeid in self.lastfailed} |
| 305 | return {x for x in result if x.exists()} |
| 306 | |
| 307 | def pytest_report_collectionfinish(self) -> Optional[str]: |
| 308 | if self.active and self.config.getoption("verbose") >= 0: |
| 309 | return "run-last-failure: %s" % self._report_status |
| 310 | return None |
| 311 | |
| 312 | def pytest_runtest_logreport(self, report: TestReport) -> None: |
| 313 | if (report.when == "call" and report.passed) or report.skipped: |
| 314 | self.lastfailed.pop(report.nodeid, None) |
| 315 | elif report.failed: |
| 316 | self.lastfailed[report.nodeid] = True |
| 317 | |
| 318 | def pytest_collectreport(self, report: CollectReport) -> None: |
| 319 | passed = report.outcome in ("passed", "skipped") |
| 320 | if passed: |
| 321 | if report.nodeid in self.lastfailed: |
| 322 | self.lastfailed.pop(report.nodeid) |
| 323 | self.lastfailed.update((item.nodeid, True) for item in report.result) |
| 324 | else: |
| 325 | self.lastfailed[report.nodeid] = True |
| 326 | |
| 327 | @hookimpl(hookwrapper=True, tryfirst=True) |
| 328 | def pytest_collection_modifyitems( |
| 329 | self, config: Config, items: List[nodes.Item] |
| 330 | ) -> Generator[None, None, None]: |
| 331 | yield |
| 332 | |
| 333 | if not self.active: |
| 334 | return |
| 335 | |
| 336 | if self.lastfailed: |
| 337 | previously_failed = [] |
| 338 | previously_passed = [] |
| 339 | for item in items: |