From 53e8c408e018efad75161e6cebe73311bbbc4e99 Mon Sep 17 00:00:00 2001 From: David Lamparter Date: Fri, 13 Sep 2024 14:42:15 +0200 Subject: [PATCH] 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 --- topotato/assertions.py | 10 ++-- topotato/base.py | 133 ++++++++++++++++++++++++++++------------- 2 files changed, 98 insertions(+), 45 deletions(-) diff --git a/topotato/assertions.py b/topotato/assertions.py index 5e8788b..4943874 100644 --- a/topotato/assertions.py +++ b/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) diff --git a/topotato/base.py b/topotato/base.py index 3015135..943a77f 100644 --- a/topotato/base.py +++ b/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