Show cache information and reset cache if test count changes

Author: nicoddemus

changelog/13122.improvement.rst

@@ -1,11 +1,15 @@
-Improve the ``--stepwise``/``--sw`` flag to not forget the last failed test in case pytest is executed later without the flag.
+The ``--stepwise`` mode received a number of improvements:
 
-This enables the following workflow:
+* It no longer forgets the last failed test in case pytest is executed later without the flag.
 
-1. Execute pytest with ``--stepwise``, pytest then stops at the first failing test;
-2. Iteratively update the code and run the test in isolation, without the ``--stepwise`` flag (for example in an IDE), until it is fixed.
-3. Execute pytest with ``--stepwise`` again and pytest will continue from the previously failed test, and if it passes, continue on to the next tests.
+  This enables the following workflow:
 
-Previously, at step 3, pytest would start from the beginning, forgetting the previously failed test.
+  1. Execute pytest with ``--stepwise``, pytest then stops at the first failing test;
+  2. Iteratively update the code and run the test in isolation, without the ``--stepwise`` flag (for example in an IDE), until it is fixed.
+  3. Execute pytest with ``--stepwise`` again and pytest will continue from the previously failed test, and if it passes, continue on to the next tests.
 
-Also added the new ``--stepwise-reset``/``--sw-reset``, allowing the user to explicitly reset the stepwise state and restart the workflow from the beginning.
+  Previously, at step 3, pytest would start from the beginning, forgetting the previously failed test.
+
+  This change however might cause issues if the ``--stepwise`` mode is used far apart in time, as the state might get stale, so the internal state will be reset automatically in case the test suite changes (for now only the number of tests are considered for this, we might change/improve this on the future).
+
+* New ``--stepwise-reset``/``--sw-reset`` flag, allowing the user to explicitly reset the stepwise state and restart the workflow from the beginning.

src/_pytest/stepwise.py

@@ -1,5 +1,10 @@
 from __future__ import annotations
 
+import dataclasses
+from datetime import datetime
+from typing import Any
+from typing import TYPE_CHECKING
+
 from _pytest import nodes
 from _pytest.cacheprovider import Cache
 from _pytest.config import Config
@@ -8,7 +13,12 @@
 from _pytest.reports import TestReport
 
 
-STEPWISE_CACHE_DIR = "cache/stepwise"
+if TYPE_CHECKING:
+    from typing import ClassVar
+
+    from typing_extensions import Self
+
+STEPWISE_CACHE_DIR = "cache/stepwise2"
 
 
 def pytest_addoption(parser: Parser) -> None:
@@ -61,41 +71,105 @@ def pytest_sessionfinish(session: Session) -> None:
             return
 
 
+@dataclasses.dataclass
+class StepwiseCacheInfo:
+    # The nodeid of the last failed test.
+    last_failed: str | None
+
+    # The number of tests in the last time --stepwise was run.
+    # We use this information as a simple way to invalidate the cache information, avoiding
+    # confusing behavior in case the cache is stale.
+    last_test_count: int | None
+
+    # The date when the cache was last updated, for information purposes only.
+    last_cache_date_str: str
+
+    _DATE_FORMAT: ClassVar[str] = "%Y-%m-%d %H:%M:%S"
+
+    @property
+    def last_cache_date(self) -> datetime:
+        return datetime.strptime(self.last_cache_date_str, self._DATE_FORMAT)
+
+    @classmethod
+    def empty(cls) -> Self:
+        return cls(
+            last_failed=None,
+            last_test_count=None,
+            last_cache_date_str=datetime.now().strftime(cls._DATE_FORMAT),
+        )
+
+    def update_date_to_now(self) -> None:
+        self.last_cache_date_str = datetime.now().strftime(self._DATE_FORMAT)
+
+
 class StepwisePlugin:
     def __init__(self, config: Config) -> None:
         self.config = config
         self.session: Session | None = None
-        self.report_status = ""
+        self.report_status: list[str] = []
         assert config.cache is not None
         self.cache: Cache = config.cache
-        self.lastfailed: str | None = self.cache.get(STEPWISE_CACHE_DIR, None)
         self.skip: bool = config.getoption("stepwise_skip")
-        if config.getoption("stepwise_reset"):
-            self.lastfailed = None
+        self.reset: bool = config.getoption("stepwise_reset")
+        self.cached_info = self._load_cached_info()
+
+    def _load_cached_info(self) -> StepwiseCacheInfo:
+        cached_dict: dict[str, Any] | None = self.cache.get(STEPWISE_CACHE_DIR, None)
+        if cached_dict:
+            try:
+                return StepwiseCacheInfo(
+                    cached_dict["last_failed"],
+                    cached_dict["last_test_count"],
+                    cached_dict["last_cache_date_str"],
+                )
+            except Exception as e:
+                error = f"{type(e).__name__}: {e}"
+                self.report_status.append(f"error reading cache, discarding ({error})")
+
+        # Cache not found or error during load, return a new cache.
+        return StepwiseCacheInfo.empty()
 
     def pytest_sessionstart(self, session: Session) -> None:
         self.session = session
 
     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."
+        last_test_count = self.cached_info.last_test_count
+        self.cached_info.last_test_count = len(items)
+
+        if self.reset:
+            self.report_status.append("resetting state, not skipping.")
+            self.cached_info.last_failed = None
+            return
+
+        if not self.cached_info.last_failed:
+            self.report_status.append("no previously failed tests, not skipping.")
+            return
+
+        if last_test_count is not None and last_test_count != len(items):
+            self.report_status.append(
+                f"test count changed, not skipping (now {len(items)} tests, previously {last_test_count})."
+            )
+            self.cached_info.last_failed = None
             return
 
-        # check all item nodes until we find a match on last failed
+        # 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:
+            if item.nodeid == self.cached_info.last_failed:
                 failed_index = index
                 break
 
         # If the previously failed test was not found among the test items,
         # do not skip any tests.
         if failed_index is None:
-            self.report_status = "previously failed test not found, not skipping."
+            self.report_status.append("previously failed test not found, not skipping.")
         else:
-            self.report_status = f"skipping {failed_index} already passed items."
+            self.report_status.append(
+                f"skipping {failed_index} already passed items (cache from {self.cached_info.last_cache_date},"
+                f" use --sw-reset to discard)."
+            )
             deselected = items[:failed_index]
             del items[:failed_index]
             config.hook.pytest_deselected(items=deselected)
@@ -105,13 +179,13 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
             if self.skip:
                 # Remove test from the failed ones (if it exists) and unset the skip option
                 # to make sure the following tests will not be skipped.
-                if report.nodeid == self.lastfailed:
-                    self.lastfailed = None
+                if report.nodeid == self.cached_info.last_failed:
+                    self.cached_info.last_failed = None
 
                 self.skip = False
             else:
                 # Mark test as the last failing and interrupt the test session.
-                self.lastfailed = report.nodeid
+                self.cached_info.last_failed = report.nodeid
                 assert self.session is not None
                 self.session.shouldstop = (
                     "Test failed, continuing from this test next run."
@@ -121,17 +195,18 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
             # If the test was actually run and did pass.
             if report.when == "call":
                 # Remove test from the failed ones, if exists.
-                if report.nodeid == self.lastfailed:
-                    self.lastfailed = None
+                if report.nodeid == self.cached_info.last_failed:
+                    self.cached_info.last_failed = None
 
-    def pytest_report_collectionfinish(self) -> str | None:
+    def pytest_report_collectionfinish(self) -> list[str] | None:
         if self.config.get_verbosity() >= 0 and self.report_status:
-            return f"stepwise: {self.report_status}"
+            return [f"stepwise: {x}" for x in self.report_status]
         return None
 
     def pytest_sessionfinish(self) -> None:
         if hasattr(self.config, "workerinput"):
             # Do not update cache if this process is a xdist worker to prevent
             # race conditions (#10641).
             return
-        self.cache.set(STEPWISE_CACHE_DIR, self.lastfailed)
+        self.cached_info.update_date_to_now()
+        self.cache.set(STEPWISE_CACHE_DIR, dataclasses.asdict(self.cached_info))