topotato: allow skipping individual items

Make use of `ItemGroup` (as was planned) to allow additional
modifications of test items, in this case adding a skip condition.

Use for FRR 8.4 compat since MLD debug messages cannot be enabled in
pim6d in that version.

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

test_mld_basic.py

@@ -81,7 +81,10 @@ def prepare(self, topo, dut, h1, h2, src):
         self.receiver = MulticastReceiver(h1, h1.iface_to('dut'))
 
         # wait for query before continuing
-        yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1] MLD query', maxwait=3.0)
+        logchecks = yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1] MLD query', maxwait=3.0)
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.dut.require_defun("debug_mld_cmd")
 
         # get out of initial reporting (prevents timing issues later)
         def expect_pkt(ipv6: IPv6, report: ICMPv6MLReport2):
@@ -99,7 +102,11 @@ def test_ssm(self, topo, dut, h1, h2, src):
 
         yield from self.receiver.join('ff05::2345', srcaddr)
 
-        yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1 (%s,ff05::2345)] NOINFO => JOIN' % srcaddr, maxwait=3.0)
+        logchecks = yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1 (%s,ff05::2345)] NOINFO => JOIN' % srcaddr, maxwait=3.0)
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.dut.require_defun("debug_mld_cmd")
+
         yield from AssertVtysh.make(dut, "pim6d", "debug show mld interface %s" % (dut.iface_to('h1').ifname))
 
         ip = IPv6(hlim=255, src=srcaddr, dst="ff05::2345")
@@ -120,7 +127,11 @@ def expect_pkt(ipv6: IPv6, udp: UDP):
     def test_asm(self, topo, dut, h1, h2, src):
         yield from self.receiver.join('ff05::1234')
 
-        yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1 (*,ff05::1234)] NOINFO => JOIN', maxwait=2.0)
+        logchecks = yield from AssertLog.make(dut, 'pim6d', '[MLD default:dut-h1 (*,ff05::1234)] NOINFO => JOIN', maxwait=2.0)
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.dut.require_defun("debug_mld_cmd")
+
         yield from AssertVtysh.make(dut, "pim6d", "debug show mld interface %s" % (dut.iface_to('h1').ifname))
 
     @topotatofunc

test_pim6_basic.py

@@ -101,7 +101,10 @@ def test_ssm(self, topo, h1, h2, r1, r2):
 
         yield from self.receiver.join('ff05::2345', srcaddr)
 
-        yield from AssertLog.make(r2, 'pim6d', '[MLD default:r2-h2 (%s,ff05::2345)] NOINFO => JOIN' % srcaddr, maxwait=3.0)
+        logchecks = yield from AssertLog.make(r2, 'pim6d', '[MLD default:r2-h2 (%s,ff05::2345)] NOINFO => JOIN' % srcaddr, maxwait=3.0)
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.r2.require_defun("debug_mld_cmd")
 
         yield from AssertLog.make(r1, 'pim6d', 'pim_forward_start: (S,G)=(%s,ff05::2345) oif=r1-r2' % srcaddr, maxwait=3.0)
 

test_pim6_prune_propagate.py

@@ -132,12 +132,16 @@ def join_traffic(self, topo, r1, r2, r3, r4, h1, h4):
 
         yield from self.receiver.join("ff35::2345", srcaddr)
 
-        yield from AssertLog.make(
+        logchecks = yield from AssertLog.make(
             r4,
             "pim6d",
             f"[MLD default:r4-lan4 ({srcaddr},ff35::2345)] NOINFO => JOIN",
             maxwait=2.0,
         )
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.r4.require_defun("debug_mld_cmd")
+
         for rtr in [r3, r2, r1]:
             yield from AssertLog.make(
                 rtr,
@@ -173,18 +177,21 @@ def mld_wait(self, topo, r1, r2, r3, r4, h1, h4):
         (Timing: query-max-response-time + robustness * interval)
         """
         yield from self.receiver.leave("ff35::2345", self.srcaddr)
-        yield from AssertLog.make(
+        logchecks = yield from AssertLog.make(
             r4,
             "pim6d",
             f"[MLD default:r4-lan4 ({self.srcaddr},ff35::2345)] JOIN => JOIN_EXPIRING",
             maxwait=2.0,
         )
-        yield from AssertLog.make(
+        logchecks += yield from AssertLog.make(
             r4,
             "pim6d",
             f"[MLD default:r4-lan4 ({self.srcaddr},ff35::2345)] JOIN_EXPIRING => NOINFO",
             maxwait=12.0,
         )
+        @logchecks.skip_on_exception
+        def need_debug_mld(testitem):
+            testitem.instance.r4.require_defun("debug_mld_cmd")
 
     @topotatofunc
     def prune(self, topo, r1, r2, r3, r4, h1, h4):

topotato/base.py

@@ -110,8 +110,36 @@ def get_callers(self) -> List[inspect.FrameInfo]:
 endtrace = _SkipTrace()
 
 
-class ItemGroup(list):
-    pass
+class ItemGroup(list["TopotatoItem"]):
+    """
+    Return value of the :py:meth:`TopotatoItem.make` generators.
+
+    This is handed back as a temporary second reference to the test items that
+    were just yielded back from the ``yield from`` in a test, in order to allow
+    optional additional tinkering with test items.
+
+    See :py:meth:`skip_on_exception` below for how to use this.
+    """
+
+    def skip_on_exception(self, fn: Callable[["TopotatoItem"], None]):
+        """
+        Add conditional skip to given test items.
+
+        For use either as a function decorator, or with helper functions.  Adds
+        the given callable function to prerequisites of all test items::
+
+            mytests = yield from AssertSomething.make(...)
+
+            # condition A (as decorator)
+            @mytests.skip_on_exception
+            def conditional(item):
+                raise TopotatoSkipped("don't actually run this")
+
+            # condition B (use helper)
+            mytests.skip_on_exception(check_feature_xyz)
+        """
+        for item in self:
+            item.skipchecks.append(fn)
 
 
 # false warning on get_closest_marker()
@@ -156,6 +184,16 @@ class TopotatoItem(nodes.Item):
     # TBD: replace/rework skipping functionality
     skipall = None
 
+    skipchecks: List[Callable[["TopotatoItem"], None]]
+    """
+    List of additional callables to check before running this test item.  The
+    called functions should either return or raise a :py:class:`TopotatoSkipped`
+    exception.
+
+    Exceptions from these checks do not automatically skip other tests, only
+    this test item is skipped.
+    """
+
     # pylint: disable=protected-access
     @classmethod
     def from_parent(
@@ -245,6 +283,7 @@ def _make(
     ) -> Generator[Optional["TopotatoItem"], Tuple["TopotatoClass", str], ItemGroup]:
         parent, _ = yield None
         self = cls.from_parent(parent, namesuffix, *args, **kwargs)
+        self.skipchecks = []
         self._codeloc = codeloc
         yield self
 
@@ -300,6 +339,8 @@ def runtest(self):
             raise TopotatoEarlierFailSkip(
                 testinst.skipall.topotato_node
             ) from testinst.skipall
+        for check in self.skipchecks:
+            check(self)
 
         self.session.config.hook.pytest_topotato_run(item=self, testfunc=self)