topotato: refactor skip-cascade into context mgr

Put the skip-later-items code into a context manager, which makes for a
great exception funnel.

Signed-off-by: David Lamparter <[email protected]>

topotato/assertions.py

@@ -31,10 +31,9 @@
     from typing_extensions import Literal  # type: ignore
 
 from scapy.packet import Packet  # type: ignore
-import pytest
 
 from .utils import json_cmp, text_rich_cmp, deindent
-from .base import TopotatoItem, TopotatoFunction, skiptrace
+from .base import TopotatoItem, TopotatoFunction, skiptrace, SkipMode
 from .livescapy import TimedScapy
 from .frr.livelog import LogMessage
 from .timeline import TimingParams
@@ -69,6 +68,8 @@ class TopotatoAssertion(TopotatoItem):
     Common base for assertions (test items that do NOT modify state)
     """
 
+    cascade_failures = SkipMode.SkipThis
+
 
 class TopotatoModifier(TopotatoItem):
     """
@@ -79,6 +80,8 @@ class TopotatoModifier(TopotatoItem):
     test continues.  If a modifier fails, the remainder of the test is skipped.
     """
 
+    cascade_failures = SkipMode.SkipThisAndLater
+
 
 class TimedMixin:
     """
@@ -456,9 +459,6 @@ def do(self, router):
         pass
 
     def runtest(self):
-        if self.skipall:
-            pytest.skip(self.skipall)
-
         router = self.instance.routers[self._rtr.name]
         self.do(router)
 

topotato/base.py

@@ -9,10 +9,10 @@
 import os
 import inspect
 from collections import OrderedDict
-import functools
 import time
 import logging
 import string
+from enum import Enum
 
 import typing
 from typing import (
@@ -144,6 +144,29 @@ def conditional(item):
             item.skipchecks.append(fn)
 
 
+class SkipMode(Enum):
+    """
+    Behavior regarding failures in earlier/later topotato test items.
+    """
+
+    DontSkip = 0
+    """
+    Always execute this test item, even if earlier items had problems.
+    """
+
+    SkipThis = 1
+    """
+    Skip this test item if something before it cascade-failed, but don't
+    cascade failures from this item.
+    """
+
+    SkipThisAndLater = 2
+    """
+    Skip this test item if something before it cascade-failed, and also make
+    failures in this test item cascade to later items.
+    """
+
+
 # false warning on get_closest_marker()
 # pylint: disable=abstract-method
 class TopotatoItem(nodes.Item):
@@ -169,6 +192,14 @@ class TopotatoItem(nodes.Item):
     Filtered heavily to condense down useful information.
     """
 
+    cascade_failures: ClassVar[SkipMode] = SkipMode.SkipThis
+    """
+    Refer to :py:class:`SkipMode`.  Normal test items derived from
+    :py:class:`.assertions.TopotatoAssertion` or
+    :py:class:`.assertions.TopotatoModifier` needn't worry about this since
+    those two set it correctly.
+    """
+
     nodeid_children_sep: Optional[str] = None
 
     _obj: "TestBase"
@@ -183,9 +214,6 @@ class TopotatoItem(nodes.Item):
     """
     timeline: "Timeline"
 
-    # TBD: replace/rework skipping functionality
-    skipall = None
-
     skipchecks: List[Callable[["TopotatoItem"], None]]
     """
     List of additional callables to check before running this test item.  The
@@ -308,12 +336,9 @@ def setup(self):
             # pylint: disable=attribute-defined-outside-init
             fn.started_ts = time.time()
 
-        tcls = self.getparent(TopotatoClass)
-        if tcls.skipall:
-            raise TopotatoEarlierFailSkip(tcls.skipall.topotato_node) from tcls.skipall
-
-        self.instance = tcls.netinst
-        self.timeline = self.instance.timeline
+        with _SkipMgr(self) as tcls:
+            self.instance = tcls.netinst
+            self.timeline = self.instance.timeline
 
     # pylint: disable=unused-argument
     @pytest.hookimpl()
@@ -327,15 +352,11 @@ def runtest(self):
         """
         Called by pytest in the "call" stage (pytest_runtest_call)
         """
-        testinst = self.getparent(TopotatoClass)
-        if testinst.skipall:
-            raise TopotatoEarlierFailSkip(
-                testinst.skipall.topotato_node
-            ) from testinst.skipall
-        for check in self.skipchecks:
-            check(self)
+        with _SkipMgr(self):
+            for check in self.skipchecks:
+                check(self)
 
-        self.session.config.hook.pytest_topotato_run(item=self, testfunc=self)
+            self.session.config.hook.pytest_topotato_run(item=self, testfunc=self)
 
     def sleep(self, step=None, until=None):
         obj = self
@@ -460,6 +481,47 @@ def repr_failure(self, excinfo, style=None):
         return res
 
 
+class _SkipMgr:
+    """
+    Simple context manager used by :py:class:`TopotatoItem` to make errors
+    skip later test items on the same network.
+    """
+
+    _item: TopotatoItem
+    _cls: "TopotatoClass"
+
+    def __init__(self, item: TopotatoItem):
+        tcls = item.getparent(TopotatoClass)
+        assert tcls is not None
+
+        self._item = item
+        self._cls = tcls
+
+    def __enter__(self) -> "TopotatoClass":
+        if self._item.cascade_failures == SkipMode.DontSkip:
+            return self._cls
+        if self._cls.skipall:
+            raise TopotatoEarlierFailSkip(self._cls.skipall_node) from self._cls.skipall
+        return self._cls
+
+    def __exit__(
+        self,
+        exc_type: Optional[Type[BaseException]],
+        exc_value: Optional[BaseException],
+        tb: Optional["TracebackType"],
+    ) -> None:
+        if exc_value is None:
+            return
+        if isinstance(exc_value, Skipped):
+            return
+        if not isinstance(exc_value, (Exception, Failed)):
+            return
+
+        if self._item.cascade_failures == SkipMode.SkipThisAndLater:
+            self._cls.skipall_node = self._item
+            self._cls.skipall = exc_value
+
+
 # false warning on get_closest_marker()
 # pylint: disable=abstract-method
 class InstanceStartup(TopotatoItem):
@@ -469,6 +531,8 @@ class InstanceStartup(TopotatoItem):
     Includes starting tshark and checking all daemons are running.
     """
 
+    cascade_failures = SkipMode.SkipThisAndLater
+
     commands: OrderedDict
 
     def __init__(self, **kwargs):
@@ -481,29 +545,18 @@ def reportinfo(self):
         return fspath, float("-inf"), "startup"
 
     def setup(self):
-        tcls = self.getparent(TopotatoClass)
-        # pylint: disable=protected-access
-        try:
+        # this needs to happen before TopotatoItem.setup, since that accesses
+        # tcls.netinst
+        with _SkipMgr(self) as tcls:
+            # pylint: disable=protected-access
             tcls.netinst = tcls.obj._setup(self.session, tcls.nodeid).prepare()
-        except (Exception, Failed, Skipped) as e:
-            e.topotato_node = self
-            self.parent.skipall = e
-            raise
         super().setup()
 
     @endtrace
     @skiptrace
     def runtest(self):
-        try:
+        with _SkipMgr(self):
             self.parent.do_start()
-        except TopotatoFail as e:
-            e.topotato_node = self
-            self.parent.skipall = e
-            raise
-        except (Exception, Failed, Skipped) as e:
-            e.topotato_node = self
-            self.parent.skipall = e
-            raise
 
 
 # false warning on get_closest_marker()
@@ -516,6 +569,9 @@ class InstanceShutdown(TopotatoItem):
     orderly fashion (otherwise you get truncated pcap files.)
     """
 
+    # kinda irrelevant here
+    cascade_failures = SkipMode.DontSkip
+
     def __init__(self, **kwargs):
         super().__init__(name="shutdown", **kwargs)
 
@@ -525,9 +581,8 @@ def reportinfo(self):
         fspath, _, _ = tcls.reportinfo()
         return fspath, float("inf"), "shutdown"
 
-    def runtest(self):
-        testfunc = functools.partial(self.parent.do_stop, self)
-        self.session.config.hook.pytest_topotato_run(item=self, testfunc=testfunc)
+    def __call__(self):
+        self.parent.do_stop(self)
 
 
 class TestBase:
@@ -787,6 +842,7 @@ class TopotatoClass(_pytest.python.Class):
     """
 
     skipall: Optional[Exception | Failed | Skipped]
+    skipall_node: Optional[TopotatoItem]
 
     starting_ts: float
     started_ts: float
@@ -841,9 +897,6 @@ def collect(self) -> Iterable[Union[nodes.Item, nodes.Collector]]:
             yield InstanceShutdown.from_parent(self)
 
     def do_start(self):
-        if self.skipall:
-            pytest.skip(self.skipall)
-
         self.starting_ts = time.time()
 
         netinst = self.netinst