LFPlugin: use sub-plugins to deselect during collection

changelog/5301.bugfix.rst

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

src/_pytest/cacheprovider.py

@@ -7,7 +7,11 @@
 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 @@
 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 @@ def _ensure_supporting_files(self):
         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
-
-    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
+        if config.getoption("lf"):
+            self._last_failed_paths = self.get_last_failed_paths()
+            config.pluginmanager.register(
+                LFPluginCollWrapper(self), "lfplugin-collwrapper"
+            )
+
+    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")

src/_pytest/config/__init__.py

@@ -795,6 +795,11 @@ def __init__(self, pluginmanager, *, invocation_params=None) -> None:
             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"""

testing/test_cacheprovider.py

@@ -265,7 +265,13 @@ def test_3(): assert 0
             """
         )
         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 @@ def test_failedfirst_order(self, testdir):
         # 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 @@ def test_lastfailed_failedfirst_order(self, testdir):
         # 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 @@ def test_a2(): assert 1
         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 @@ def test_bar_2(): pass
         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 @@ def test_foo_4(): pass
         """
         )
         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 @@ def test_1(i):
         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 @@ def test_3(): pass
         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 @@ def test_lastfailed_with_known_failures_not_being_selected(self, testdir):
 
         # 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 *",
             ]
         )