11from typing import List
22from typing import Optional
3+ from typing import TYPE_CHECKING
34
45import pytest
56from _pytest import nodes
89from _pytest .main import Session
910from _pytest .reports import TestReport
1011
12+ if TYPE_CHECKING :
13+ from _pytest .cacheprovider import Cache
14+
15+ STEPWISE_CACHE_DIR = "cache/stepwise"
16+
1117
1218def pytest_addoption (parser : Parser ) -> None :
1319 group = parser .getgroup ("general" )
1420 group .addoption (
1521 "--sw" ,
1622 "--stepwise" ,
1723 action = "store_true" ,
24+ default = False ,
1825 dest = "stepwise" ,
1926 help = "exit on test failure and continue from last failing test next time" ,
2027 )
2128 group .addoption (
29+ "--sw-skip" ,
2230 "--stepwise-skip" ,
2331 action = "store_true" ,
32+ default = False ,
2433 dest = "stepwise_skip" ,
2534 help = "ignore the first failing test but stop on the next failing test" ,
2635 )
2736
2837
2938@pytest .hookimpl
3039def pytest_configure (config : Config ) -> None :
31- config .pluginmanager .register (StepwisePlugin (config ), "stepwiseplugin" )
40+ # We should always have a cache as cache provider plugin uses tryfirst=True
41+ if config .getoption ("stepwise" ):
42+ config .pluginmanager .register (StepwisePlugin (config ), "stepwiseplugin" )
43+
44+
45+ def pytest_sessionfinish (session : Session ) -> None :
46+ if not session .config .getoption ("stepwise" ):
47+ assert session .config .cache is not None
48+ # Clear the list of failing tests if the plugin is not active.
49+ session .config .cache .set (STEPWISE_CACHE_DIR , [])
3250
3351
3452class StepwisePlugin :
3553 def __init__ (self , config : Config ) -> None :
3654 self .config = config
37- self .active = config .getvalue ("stepwise" )
3855 self .session : Optional [Session ] = None
3956 self .report_status = ""
40-
41- if self .active :
42- assert config .cache is not None
43- self .lastfailed = config .cache .get ("cache/stepwise" , None )
44- self .skip = config .getvalue ("stepwise_skip" )
57+ assert config .cache is not None
58+ self .cache : Cache = config .cache
59+ self .lastfailed : Optional [str ] = self .cache .get (STEPWISE_CACHE_DIR , None )
60+ self .skip : bool = config .getoption ("stepwise_skip" )
4561
4662 def pytest_sessionstart (self , session : Session ) -> None :
4763 self .session = session
4864
4965 def pytest_collection_modifyitems (
50- self , session : Session , config : Config , items : List [nodes .Item ]
66+ self , config : Config , items : List [nodes .Item ]
5167 ) -> None :
52- if not self .active :
53- return
5468 if not self .lastfailed :
5569 self .report_status = "no previously failed tests, not skipping."
5670 return
5771
58- already_passed = []
59- found = False
60-
61- # Make a list of all tests that have been run before the last failing one.
62- for item in items :
72+ # check all item nodes until we find a match on last failed
73+ failed_index = None
74+ for index , item in enumerate (items ):
6375 if item .nodeid == self .lastfailed :
64- found = True
76+ failed_index = index
6577 break
66- else :
67- already_passed .append (item )
6878
6979 # If the previously failed test was not found among the test items,
7080 # do not skip any tests.
71- if not found :
81+ if failed_index is None :
7282 self .report_status = "previously failed test not found, not skipping."
73- already_passed = []
7483 else :
75- self .report_status = "skipping {} already passed items." .format (
76- len (already_passed )
77- )
78-
79- for item in already_passed :
80- items .remove (item )
81-
82- config .hook .pytest_deselected (items = already_passed )
84+ self .report_status = f"skipping { failed_index } already passed items."
85+ deselected = items [:failed_index ]
86+ del items [:failed_index ]
87+ config .hook .pytest_deselected (items = deselected )
8388
8489 def pytest_runtest_logreport (self , report : TestReport ) -> None :
85- if not self .active :
86- return
87-
8890 if report .failed :
8991 if self .skip :
9092 # Remove test from the failed ones (if it exists) and unset the skip option
@@ -109,14 +111,9 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
109111 self .lastfailed = None
110112
111113 def pytest_report_collectionfinish (self ) -> Optional [str ]:
112- if self .active and self . config .getoption ("verbose" ) >= 0 and self .report_status :
113- return "stepwise: %s" % self .report_status
114+ if self .config .getoption ("verbose" ) >= 0 and self .report_status :
115+ return f "stepwise: { self .report_status } "
114116 return None
115117
116- def pytest_sessionfinish (self , session : Session ) -> None :
117- assert self .config .cache is not None
118- if self .active :
119- self .config .cache .set ("cache/stepwise" , self .lastfailed )
120- else :
121- # Clear the list of failing tests if the plugin is not active.
122- self .config .cache .set ("cache/stepwise" , [])
118+ def pytest_sessionfinish (self ) -> None :
119+ self .cache .set (STEPWISE_CACHE_DIR , self .lastfailed )
0 commit comments