stepwise.py 7.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  1. from __future__ import annotations
  2. import dataclasses
  3. from datetime import datetime
  4. from datetime import timedelta
  5. from typing import Any
  6. from typing import TYPE_CHECKING
  7. from _pytest import nodes
  8. from _pytest.cacheprovider import Cache
  9. from _pytest.config import Config
  10. from _pytest.config.argparsing import Parser
  11. from _pytest.main import Session
  12. from _pytest.reports import TestReport
  13. if TYPE_CHECKING:
  14. from typing_extensions import Self
  15. STEPWISE_CACHE_DIR = "cache/stepwise"
  16. def pytest_addoption(parser: Parser) -> None:
  17. group = parser.getgroup("general")
  18. group.addoption(
  19. "--sw",
  20. "--stepwise",
  21. action="store_true",
  22. default=False,
  23. dest="stepwise",
  24. help="Exit on test failure and continue from last failing test next time",
  25. )
  26. group.addoption(
  27. "--sw-skip",
  28. "--stepwise-skip",
  29. action="store_true",
  30. default=False,
  31. dest="stepwise_skip",
  32. help="Ignore the first failing test but stop on the next failing test. "
  33. "Implicitly enables --stepwise.",
  34. )
  35. group.addoption(
  36. "--sw-reset",
  37. "--stepwise-reset",
  38. action="store_true",
  39. default=False,
  40. dest="stepwise_reset",
  41. help="Resets stepwise state, restarting the stepwise workflow. "
  42. "Implicitly enables --stepwise.",
  43. )
  44. def pytest_configure(config: Config) -> None:
  45. # --stepwise-skip/--stepwise-reset implies stepwise.
  46. if config.option.stepwise_skip or config.option.stepwise_reset:
  47. config.option.stepwise = True
  48. if config.getoption("stepwise"):
  49. config.pluginmanager.register(StepwisePlugin(config), "stepwiseplugin")
  50. def pytest_sessionfinish(session: Session) -> None:
  51. if not session.config.getoption("stepwise"):
  52. assert session.config.cache is not None
  53. if hasattr(session.config, "workerinput"):
  54. # Do not update cache if this process is a xdist worker to prevent
  55. # race conditions (#10641).
  56. return
  57. @dataclasses.dataclass
  58. class StepwiseCacheInfo:
  59. # The nodeid of the last failed test.
  60. last_failed: str | None
  61. # The number of tests in the last time --stepwise was run.
  62. # We use this information as a simple way to invalidate the cache information, avoiding
  63. # confusing behavior in case the cache is stale.
  64. last_test_count: int | None
  65. # The date when the cache was last updated, for information purposes only.
  66. last_cache_date_str: str
  67. @property
  68. def last_cache_date(self) -> datetime:
  69. return datetime.fromisoformat(self.last_cache_date_str)
  70. @classmethod
  71. def empty(cls) -> Self:
  72. return cls(
  73. last_failed=None,
  74. last_test_count=None,
  75. last_cache_date_str=datetime.now().isoformat(),
  76. )
  77. def update_date_to_now(self) -> None:
  78. self.last_cache_date_str = datetime.now().isoformat()
  79. class StepwisePlugin:
  80. def __init__(self, config: Config) -> None:
  81. self.config = config
  82. self.session: Session | None = None
  83. self.report_status: list[str] = []
  84. assert config.cache is not None
  85. self.cache: Cache = config.cache
  86. self.skip: bool = config.getoption("stepwise_skip")
  87. self.reset: bool = config.getoption("stepwise_reset")
  88. self.cached_info = self._load_cached_info()
  89. def _load_cached_info(self) -> StepwiseCacheInfo:
  90. cached_dict: dict[str, Any] | None = self.cache.get(STEPWISE_CACHE_DIR, None)
  91. if cached_dict:
  92. try:
  93. return StepwiseCacheInfo(
  94. cached_dict["last_failed"],
  95. cached_dict["last_test_count"],
  96. cached_dict["last_cache_date_str"],
  97. )
  98. except (KeyError, TypeError) as e:
  99. error = f"{type(e).__name__}: {e}"
  100. self.report_status.append(f"error reading cache, discarding ({error})")
  101. # Cache not found or error during load, return a new cache.
  102. return StepwiseCacheInfo.empty()
  103. def pytest_sessionstart(self, session: Session) -> None:
  104. self.session = session
  105. def pytest_collection_modifyitems(
  106. self, config: Config, items: list[nodes.Item]
  107. ) -> None:
  108. last_test_count = self.cached_info.last_test_count
  109. self.cached_info.last_test_count = len(items)
  110. if self.reset:
  111. self.report_status.append("resetting state, not skipping.")
  112. self.cached_info.last_failed = None
  113. return
  114. if not self.cached_info.last_failed:
  115. self.report_status.append("no previously failed tests, not skipping.")
  116. return
  117. if last_test_count is not None and last_test_count != len(items):
  118. self.report_status.append(
  119. f"test count changed, not skipping (now {len(items)} tests, previously {last_test_count})."
  120. )
  121. self.cached_info.last_failed = None
  122. return
  123. # Check all item nodes until we find a match on last failed.
  124. failed_index = None
  125. for index, item in enumerate(items):
  126. if item.nodeid == self.cached_info.last_failed:
  127. failed_index = index
  128. break
  129. # If the previously failed test was not found among the test items,
  130. # do not skip any tests.
  131. if failed_index is None:
  132. self.report_status.append("previously failed test not found, not skipping.")
  133. else:
  134. cache_age = datetime.now() - self.cached_info.last_cache_date
  135. # Round up to avoid showing microseconds.
  136. cache_age = timedelta(seconds=int(cache_age.total_seconds()))
  137. self.report_status.append(
  138. f"skipping {failed_index} already passed items (cache from {cache_age} ago,"
  139. f" use --sw-reset to discard)."
  140. )
  141. deselected = items[:failed_index]
  142. del items[:failed_index]
  143. config.hook.pytest_deselected(items=deselected)
  144. def pytest_runtest_logreport(self, report: TestReport) -> None:
  145. if report.failed:
  146. if self.skip:
  147. # Remove test from the failed ones (if it exists) and unset the skip option
  148. # to make sure the following tests will not be skipped.
  149. if report.nodeid == self.cached_info.last_failed:
  150. self.cached_info.last_failed = None
  151. self.skip = False
  152. else:
  153. # Mark test as the last failing and interrupt the test session.
  154. self.cached_info.last_failed = report.nodeid
  155. assert self.session is not None
  156. self.session.shouldstop = (
  157. "Test failed, continuing from this test next run."
  158. )
  159. else:
  160. # If the test was actually run and did pass.
  161. if report.when == "call":
  162. # Remove test from the failed ones, if exists.
  163. if report.nodeid == self.cached_info.last_failed:
  164. self.cached_info.last_failed = None
  165. def pytest_report_collectionfinish(self) -> list[str] | None:
  166. if self.config.get_verbosity() >= 0 and self.report_status:
  167. return [f"stepwise: {x}" for x in self.report_status]
  168. return None
  169. def pytest_sessionfinish(self) -> None:
  170. if hasattr(self.config, "workerinput"):
  171. # Do not update cache if this process is a xdist worker to prevent
  172. # race conditions (#10641).
  173. return
  174. self.cached_info.update_date_to_now()
  175. self.cache.set(STEPWISE_CACHE_DIR, dataclasses.asdict(self.cached_info))