- fixed/added suport for --exitfirst and --stepwise;

src/pytest_cleanslate/plugin.py

@@ -1,29 +1,26 @@
 import pytest
+from pathlib import Path
 
 
 class CleanSlateItem(pytest.Item):
     """Item that stands for a Module until it can be collected from its forked subprocess"""
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
     def runtest(self):
         raise RuntimeError("This should never execute")
 
-    def run_forked(self, cleanslate_plugin):
+    def collect_and_run(self):
         # adapted from pytest-forked
         import marshal
         import _pytest
         import pytest_forked as ptf # FIXME pytest-forked is unmaintained
         import py                   # FIXME py is maintenance only
 
-        ihook = self.ihook
-        ihook.pytest_runtest_logstart(nodeid=self.nodeid, location=self.location)
-
         def runforked():
             # Use 'parent' as it would have been in pytest_pycollect_makemodule, so that our
             # nodes aren't included in the chain, as they might confuse other plugins (such as 'mark')
             module = pytest.Module.from_parent(parent=self.parent.parent, path=self.path)
-            reports = list()
 
             def collect_items(collector):
                 for it in collector.collect():
@@ -32,34 +29,40 @@ def collect_items(collector):
                     else:
                         yield it
 
-            items = list(collect_items(module))
+            self.session.items = list(collect_items(module))
+
+            pm = self.config.pluginmanager
+            caller = pm.subset_hook_caller('pytest_collection_modifyitems', remove_plugins=[self.parent.plugin])
+            caller(session=self.session, config=self.config, items=self.session.items)
 
-            caller = self.config.pluginmanager.subset_hook_caller('pytest_collection_modifyitems',
-                                                                  remove_plugins=[cleanslate_plugin])
-            caller(session=self.session, config=self.config, items=items)
+            reports = list()
+            class ReportSaver:
+                @pytest.hookimpl
+                def pytest_runtest_logreport(self, report):
+                    reports.append(report)
+
+            pm.register(ReportSaver())
 
-            for it in items:
-                reports.extend(ptf.forked_run_report(it))
+            try:
+                self.ihook.pytest_runtestloop(session=self.session)
+            except (pytest.Session.Interrupted, pytest.Session.Failed):
+                pass
 
             return marshal.dumps([self.config.hook.pytest_report_to_serializable(config=self.config, report=r) for r in reports])
 
         ff = py.process.ForkedFunc(runforked)
         result = ff.waitfinish()
 
-        if result.retval is not None:
-            reports = [self.config.hook.pytest_report_from_serializable(config=self.config, data=r) for r in marshal.loads(result.retval)]
-        else:
-            reports = [ptf.report_process_crash(self, result)]
-
-        for r in reports:
-            ihook.pytest_runtest_logreport(report=r)
+        if result.retval is None:
+            return [ptf.report_process_crash(self, result)]
 
-        ihook.pytest_runtest_logfinish(nodeid=self.nodeid, location=self.location)
+        return [self.config.hook.pytest_report_from_serializable(config=self.config, data=r) for r in marshal.loads(result.retval)]
 
 
 class CleanSlateCollector(pytest.File, pytest.Collector):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+    def __init__(self, *, plugin, **kwargs):
+        super().__init__(**kwargs)
+        self.plugin = plugin
 
     def collect(self):
         yield CleanSlateItem.from_parent(parent=self, name=self.name)
@@ -71,14 +74,26 @@ class CleanSlatePlugin:
 
 
     @pytest.hookimpl(tryfirst=True)
-    def pytest_pycollect_makemodule(self, module_path: pytest.Path, parent):
-        return CleanSlateCollector.from_parent(parent, path=module_path)
+    def pytest_pycollect_makemodule(self, module_path: Path, parent):
+        return CleanSlateCollector.from_parent(parent, path=module_path, plugin=self)
 
 
     @pytest.hookimpl(tryfirst=True)
-    def pytest_runtestloop(self, session: pytest.Session):
-        for item in session.items:
-            item.run_forked(self)
+    def pytest_runtest_protocol(self, item: pytest.Item, nextitem: pytest.Item):
+        import pytest_forked as ptf # FIXME pytest-forked is unmaintained
+
+        ihook = item.ihook
+        ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
+        if isinstance(item, CleanSlateItem):
+            reports = item.collect_and_run()
+        else:
+            # note any side effects, such as setting session.shouldstop, are lost...
+            reports = ptf.forked_run_report(item)
+
+        for rep in reports:
+            ihook.pytest_runtest_logreport(report=rep)
+
+        ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
         return True
 
 

tests/test_cleanslate.py

@@ -155,7 +155,9 @@ async def test_asyncio():
 
 
 @pytest.mark.parametrize("tf", ['test_one', 'test_two'])
-def test_filter(tests_dir, tf):
+def test_mark(tests_dir, tf):
+    # the built-in "mark" plugin implements '-k' and '-m'
+
     test = seq2p(tests_dir, 1)
     test.write_text("""\
 def test_one():
@@ -166,8 +168,47 @@ def test_two():
 """)
 
     p = subprocess.run([sys.executable, '-m', 'pytest', '--cleanslate', '-k', tf, tests_dir], check=False)
-#    p = subprocess.run([sys.executable, '-m', 'pytest', '-k', tf, tests_dir], check=False)
     if tf == 'test_two':
         assert p.returncode == pytest.ExitCode.TESTS_FAILED
     else:
         assert p.returncode == pytest.ExitCode.OK
+
+
+def test_exitfirst(tests_dir):
+    test = seq2p(tests_dir, 1)
+    test.write_text("""\
+from pathlib import Path
+
+def test_one():
+    assert False
+
+def test_two():
+    Path('litmus.txt').touch()
+""")
+
+    p = subprocess.run([sys.executable, '-m', 'pytest', '--cleanslate', '--exitfirst', '-s', tests_dir], check=False,
+                       capture_output=True)
+    assert p.returncode == pytest.ExitCode.TESTS_FAILED
+    assert not Path('litmus.txt').exists()
+    assert 'CRASHED' not in str(p.stdout, 'utf-8')
+
+
+def test_shouldstop(tests_dir):
+    test = seq2p(tests_dir, 1)
+    test.write_text("""\
+import pytest
+from pathlib import Path
+
+def test_one():
+    assert False
+
+def test_two():
+    Path('litmus.txt').touch()
+""")
+
+    # --stepwise sets session.shouldstop upon a test failure.
+    p = subprocess.run([sys.executable, '-m', 'pytest', '--cleanslate', '--stepwise', '-s', tests_dir], check=False,
+                       capture_output=True)
+    assert p.returncode == pytest.ExitCode.INTERRUPTED
+    assert not Path('litmus.txt').exists()
+    assert 'CRASHED' not in str(p.stdout, 'utf-8')