intermediate changes

ref:614ed510ddd3cdf86a8c5dbf19afd113397e0172

1 files changed, 56 insertions, 45 deletions

diff --git a/contrib/python/pytest/py3/_pytest/stepwise.py b/contrib/python/pytest/py3/_pytest/stepwise.py
index 6fa21cd1c6..197577c790 100644
--- a/contrib/python/pytest/py3/_pytest/stepwise.py
+++ b/contrib/python/pytest/py3/_pytest/stepwise.py
@@ -1,79 +1,92 @@
+from typing import List
+from typing import Optional
+from typing import TYPE_CHECKING
+
 import pytest
+from _pytest import nodes
+from _pytest.config import Config
+from _pytest.config.argparsing import Parser
+from _pytest.main import Session
+from _pytest.reports import TestReport
+
+if TYPE_CHECKING:
+    from _pytest.cacheprovider import Cache
 
+STEPWISE_CACHE_DIR = "cache/stepwise"
 
-def pytest_addoption(parser):
+
+def pytest_addoption(parser: Parser) -> None:
     group = parser.getgroup("general")
     group.addoption(
         "--sw",
         "--stepwise",
         action="store_true",
+        default=False,
         dest="stepwise",
         help="exit on test failure and continue from last failing test next time",
     )
     group.addoption(
+        "--sw-skip",
         "--stepwise-skip",
         action="store_true",
+        default=False,
         dest="stepwise_skip",
         help="ignore the first failing test but stop on the next failing test",
     )
 
 
 @pytest.hookimpl
-def pytest_configure(config):
-    config.pluginmanager.register(StepwisePlugin(config), "stepwiseplugin")
+def pytest_configure(config: Config) -> None:
+    # We should always have a cache as cache provider plugin uses tryfirst=True
+    if config.getoption("stepwise"):
+        config.pluginmanager.register(StepwisePlugin(config), "stepwiseplugin")
+
+
+def pytest_sessionfinish(session: Session) -> None:
+    if not session.config.getoption("stepwise"):
+        assert session.config.cache is not None
+        # Clear the list of failing tests if the plugin is not active.
+        session.config.cache.set(STEPWISE_CACHE_DIR, [])
 
 
 class StepwisePlugin:
-    def __init__(self, config):
+    def __init__(self, config: Config) -> None:
         self.config = config
-        self.active = config.getvalue("stepwise")
-        self.session = None
+        self.session: Optional[Session] = None
         self.report_status = ""
+        assert config.cache is not None
+        self.cache: Cache = config.cache
+        self.lastfailed: Optional[str] = self.cache.get(STEPWISE_CACHE_DIR, None)
+        self.skip: bool = config.getoption("stepwise_skip")
 
-        if self.active:
-            self.lastfailed = config.cache.get("cache/stepwise", None)
-            self.skip = config.getvalue("stepwise_skip")
-
-    def pytest_sessionstart(self, session):
+    def pytest_sessionstart(self, session: Session) -> None:
         self.session = session
 
-    def pytest_collection_modifyitems(self, session, config, items):
-        if not self.active:
-            return
+    def pytest_collection_modifyitems(
+        self, config: Config, items: List[nodes.Item]
+    ) -> None:
         if not self.lastfailed:
             self.report_status = "no previously failed tests, not skipping."
             return
 
-        already_passed = []
-        found = False
-
-        # Make a list of all tests that have been run before the last failing one.
-        for item in items:
+        # check all item nodes until we find a match on last failed
+        failed_index = None
+        for index, item in enumerate(items):
             if item.nodeid == self.lastfailed:
-                found = True
+                failed_index = index
                 break
-            else:
-                already_passed.append(item)
 
         # If the previously failed test was not found among the test items,
         # do not skip any tests.
-        if not found:
+        if failed_index is None:
             self.report_status = "previously failed test not found, not skipping."
-            already_passed = []
         else:
-            self.report_status = "skipping {} already passed items.".format(
-                len(already_passed)
-            )
-
-        for item in already_passed:
-            items.remove(item)
-
-        config.hook.pytest_deselected(items=already_passed)
-
-    def pytest_runtest_logreport(self, report):
-        if not self.active:
-            return
+            self.report_status = f"skipping {failed_index} already passed items."
+            deselected = items[:failed_index]
+            del items[:failed_index]
+            config.hook.pytest_deselected(items=deselected)
 
+    def pytest_runtest_logreport(self, report: TestReport) -> None:
         if report.failed:
             if self.skip:
                 # Remove test from the failed ones (if it exists) and unset the skip option
@@ -85,6 +98,7 @@ class StepwisePlugin:
             else:
                 # Mark test as the last failing and interrupt the test session.
                 self.lastfailed = report.nodeid
+                assert self.session is not None
                 self.session.shouldstop = (
                     "Test failed, continuing from this test next run."
                 )
@@ -96,13 +110,10 @@ class StepwisePlugin:
                 if report.nodeid == self.lastfailed:
                     self.lastfailed = None
 
-    def pytest_report_collectionfinish(self):
-        if self.active and self.config.getoption("verbose") >= 0 and self.report_status:
-            return "stepwise: %s" % self.report_status
+    def pytest_report_collectionfinish(self) -> Optional[str]:
+        if self.config.getoption("verbose") >= 0 and self.report_status:
+            return f"stepwise: {self.report_status}"
+        return None
 
-    def pytest_sessionfinish(self, session):
-        if self.active:
-            self.config.cache.set("cache/stepwise", self.lastfailed)
-        else:
-            # Clear the list of failing tests if the plugin is not active.
-            self.config.cache.set("cache/stepwise", [])
+    def pytest_sessionfinish(self) -> None:
+        self.cache.set(STEPWISE_CACHE_DIR, self.lastfailed)