LFPlugin: use sub-plugins to deselect during collection (#6448)

Fixes https://github.com/pytest-dev/pytest/issues/5301.

Refactor/steps:

- use var
- harden test_lastfailed_usecase
- harden test_failedfirst_order
- revisit last_failed_paths
- harden test_lastfailed_with_known_failures_not_being_selected
This commit is contained in:
Daniel Hahler 2020-02-19 21:33:03 +01:00 committed by GitHub
parent af2b0e1174
commit 1b30514783
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 127 additions and 41 deletions

View File

@ -0,0 +1 @@
Fix ``--last-failed`` to collect new tests from files with known failures.

View File

@ -7,7 +7,11 @@ ignores the external pytest-cache
import json
import os
from collections import OrderedDict
from typing import Dict
from typing import Generator
from typing import List
from typing import Optional
from typing import Set
import attr
import py
@ -16,10 +20,12 @@ import pytest
from .pathlib import Path
from .pathlib import resolve_from_str
from .pathlib import rm_rf
from .reports import CollectReport
from _pytest import nodes
from _pytest._io import TerminalWriter
from _pytest.config import Config
from _pytest.main import Session
from _pytest.python import Module
README_CONTENT = """\
# pytest cache directory #
@ -161,42 +167,88 @@ class Cache:
cachedir_tag_path.write_bytes(CACHEDIR_TAG_CONTENT)
class LFPluginCollWrapper:
def __init__(self, lfplugin: "LFPlugin"):
self.lfplugin = lfplugin
self._collected_at_least_one_failure = False
@pytest.hookimpl(hookwrapper=True)
def pytest_make_collect_report(self, collector) -> Generator:
if isinstance(collector, Session):
out = yield
res = out.get_result() # type: CollectReport
# Sort any lf-paths to the beginning.
lf_paths = self.lfplugin._last_failed_paths
res.result = sorted(
res.result, key=lambda x: 0 if Path(x.fspath) in lf_paths else 1,
)
out.force_result(res)
return
elif isinstance(collector, Module):
if Path(collector.fspath) in self.lfplugin._last_failed_paths:
out = yield
res = out.get_result()
filtered_result = [
x for x in res.result if x.nodeid in self.lfplugin.lastfailed
]
if filtered_result:
res.result = filtered_result
out.force_result(res)
if not self._collected_at_least_one_failure:
self.lfplugin.config.pluginmanager.register(
LFPluginCollSkipfiles(self.lfplugin), "lfplugin-collskip"
)
self._collected_at_least_one_failure = True
return res
yield
class LFPluginCollSkipfiles:
def __init__(self, lfplugin: "LFPlugin"):
self.lfplugin = lfplugin
@pytest.hookimpl
def pytest_make_collect_report(self, collector) -> Optional[CollectReport]:
if isinstance(collector, Module):
if Path(collector.fspath) not in self.lfplugin._last_failed_paths:
self.lfplugin._skipped_files += 1
return CollectReport(
collector.nodeid, "passed", longrepr=None, result=[]
)
return None
class LFPlugin:
""" Plugin which implements the --lf (run last-failing) option """
def __init__(self, config):
def __init__(self, config: Config) -> None:
self.config = config
active_keys = "lf", "failedfirst"
self.active = any(config.getoption(key) for key in active_keys)
self.lastfailed = config.cache.get("cache/lastfailed", {})
assert config.cache
self.lastfailed = config.cache.get(
"cache/lastfailed", {}
) # type: Dict[str, bool]
self._previously_failed_count = None
self._report_status = None
self._skipped_files = 0 # count skipped files during collection due to --lf
def last_failed_paths(self):
"""Returns a set with all Paths()s of the previously failed nodeids (cached).
"""
try:
return self._last_failed_paths
except AttributeError:
rootpath = Path(self.config.rootdir)
result = {rootpath / nodeid.split("::")[0] for nodeid in self.lastfailed}
result = {x for x in result if x.exists()}
self._last_failed_paths = result
return result
if config.getoption("lf"):
self._last_failed_paths = self.get_last_failed_paths()
config.pluginmanager.register(
LFPluginCollWrapper(self), "lfplugin-collwrapper"
)
def pytest_ignore_collect(self, path):
"""
Ignore this file path if we are in --lf mode and it is not in the list of
previously failed files.
"""
if self.active and self.config.getoption("lf") and path.isfile():
last_failed_paths = self.last_failed_paths()
if last_failed_paths:
skip_it = Path(path) not in self.last_failed_paths()
if skip_it:
self._skipped_files += 1
return skip_it
def get_last_failed_paths(self) -> Set[Path]:
"""Returns a set with all Paths()s of the previously failed nodeids."""
rootpath = Path(self.config.rootdir)
result = {rootpath / nodeid.split("::")[0] for nodeid in self.lastfailed}
return {x for x in result if x.exists()}
def pytest_report_collectionfinish(self):
if self.active and self.config.getoption("verbose") >= 0:
@ -380,7 +432,7 @@ def pytest_cmdline_main(config):
@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
def pytest_configure(config: Config) -> None:
config.cache = Cache.for_config(config)
config.pluginmanager.register(LFPlugin(config), "lfplugin")
config.pluginmanager.register(NFPlugin(config), "nfplugin")

View File

@ -795,6 +795,11 @@ class Config:
kwargs=dict(parser=self._parser, pluginmanager=self.pluginmanager)
)
if False: # TYPE_CHECKING
from _pytest.cacheprovider import Cache
self.cache = None # type: Optional[Cache]
@property
def invocation_dir(self):
"""Backward compatibility"""

View File

@ -265,7 +265,13 @@ class TestLastFailed:
"""
)
result = testdir.runpytest(str(p), "--lf")
result.stdout.fnmatch_lines(["*2 passed*1 desel*"])
result.stdout.fnmatch_lines(
[
"collected 2 items",
"run-last-failure: rerun previous 2 failures",
"*= 2 passed in *",
]
)
result = testdir.runpytest(str(p), "--lf")
result.stdout.fnmatch_lines(
[
@ -295,8 +301,15 @@ class TestLastFailed:
# Test order will be collection order; alphabetical
result.stdout.fnmatch_lines(["test_a.py*", "test_b.py*"])
result = testdir.runpytest("--ff")
# Test order will be failing tests firs
result.stdout.fnmatch_lines(["test_b.py*", "test_a.py*"])
# Test order will be failing tests first
result.stdout.fnmatch_lines(
[
"collected 2 items",
"run-last-failure: rerun previous 1 failure first",
"test_b.py*",
"test_a.py*",
]
)
def test_lastfailed_failedfirst_order(self, testdir):
testdir.makepyfile(
@ -307,7 +320,7 @@ class TestLastFailed:
# Test order will be collection order; alphabetical
result.stdout.fnmatch_lines(["test_a.py*", "test_b.py*"])
result = testdir.runpytest("--lf", "--ff")
# Test order will be failing tests firs
# Test order will be failing tests first
result.stdout.fnmatch_lines(["test_b.py*"])
result.stdout.no_fnmatch_line("*test_a.py*")
@ -332,7 +345,7 @@ class TestLastFailed:
result = testdir.runpytest("--lf", p2)
result.stdout.fnmatch_lines(["*1 passed*"])
result = testdir.runpytest("--lf", p)
result.stdout.fnmatch_lines(["*1 failed*1 desel*"])
result.stdout.fnmatch_lines(["collected 1 item", "*= 1 failed in *"])
def test_lastfailed_usecase_splice(self, testdir, monkeypatch):
monkeypatch.setattr("sys.dont_write_bytecode", True)
@ -658,7 +671,13 @@ class TestLastFailed:
assert self.get_cached_last_failed(testdir) == ["test_foo.py::test_foo_4"]
result = testdir.runpytest("--last-failed")
result.stdout.fnmatch_lines(["*1 failed, 1 deselected*"])
result.stdout.fnmatch_lines(
[
"collected 1 item",
"run-last-failure: rerun previous 1 failure (skipped 1 file)",
"*= 1 failed in *",
]
)
assert self.get_cached_last_failed(testdir) == ["test_foo.py::test_foo_4"]
# 3. fix test_foo_4, run only test_foo.py
@ -669,7 +688,13 @@ class TestLastFailed:
"""
)
result = testdir.runpytest(test_foo, "--last-failed")
result.stdout.fnmatch_lines(["*1 passed, 1 deselected*"])
result.stdout.fnmatch_lines(
[
"collected 1 item",
"run-last-failure: rerun previous 1 failure",
"*= 1 passed in *",
]
)
assert self.get_cached_last_failed(testdir) == []
result = testdir.runpytest("--last-failed")
@ -759,9 +784,9 @@ class TestLastFailed:
result = testdir.runpytest("--lf")
result.stdout.fnmatch_lines(
[
"collected 5 items / 3 deselected / 2 selected",
"collected 2 items",
"run-last-failure: rerun previous 2 failures (skipped 1 file)",
"*2 failed*3 deselected*",
"*= 2 failed in *",
]
)
@ -776,9 +801,9 @@ class TestLastFailed:
result = testdir.runpytest("--lf")
result.stdout.fnmatch_lines(
[
"collected 5 items / 3 deselected / 2 selected",
"collected 2 items",
"run-last-failure: rerun previous 2 failures (skipped 2 files)",
"*2 failed*3 deselected*",
"*= 2 failed in *",
]
)
@ -815,12 +840,15 @@ class TestLastFailed:
# Remove/rename test.
testdir.makepyfile(**{"pkg1/test_1.py": """def test_renamed(): assert 0"""})
result = testdir.runpytest("--lf")
result = testdir.runpytest("--lf", "-rf")
result.stdout.fnmatch_lines(
[
"collected 1 item",
"run-last-failure: 1 known failures not in selected tests (skipped 1 file)",
"* 1 failed in *",
"collected 2 items",
"run-last-failure: 1 known failures not in selected tests",
"pkg1/test_1.py F *",
"pkg1/test_2.py . *",
"FAILED pkg1/test_1.py::test_renamed - assert 0",
"* 1 failed, 1 passed in *",
]
)