topology: add topology controller

Topology controller can be associated with topology using the topology
marker. It provides various hooks to implement per-topology setup and
teardown.

docs/classes.rst

@@ -13,6 +13,7 @@ configuration options.
 * :class:`~pytest_mh.MultihostHost`: lives through the whole pytest session, gives low-level access to the host
 * :class:`~pytest_mh.MultihostRole`: lives only for a single test case, provides high-level API
 * :class:`~pytest_mh.MultihostUtility`: provides high-level API that can be shared between multiple roles
+* :class:`~pytest_mh.TopologyController`: control topology behavior such as per-topology setup and teardown
 
 .. mermaid::
     :caption: Class relationship
@@ -210,6 +211,59 @@ There are already some utility classes implemented in ``pytest-mh``. See
     Each change that is made through the utility object (such as writing to a
     file) is automatically reverted (the original file is restored).
 
+TopologyController
+==================
+
+Topology controller can be assigned to a topology via `@pytest.mark.topology`
+or through known topology class. This controller provides various methods to
+control the topology behavior:
+
+* per-topology setup and teardown, called once before the first test/after the
+  last test for given topology is executed
+* per-test topology setup and teardown, called before and after every test case
+  for given topology
+* check topology requirements and skip the test if these are not satisfied
+
+In order to use the controller, you need to inherit from
+:class:`~pytest_mh.TopologyController` and override desired methods. Each method
+can take any parameter as defined by the topology fixtures. The parameter value
+is an instance of a :class:`~pytest_mh.MultihostHost` object.
+
+See :class:`~pytest_mh.TopologyController` for API documentation
+
+.. code-block: python
+    :caption: Topology controller example
+
+    class ExampleController(TopologyController):
+        # Override methods that you wish to use
+
+        def skip(self, client: ClientHost) -> str | None:
+            result = client.ssh.run(check_requirements, raise_on_error=False)
+            if result.rc != 0:
+                return "Topology requirements were not met"
+
+            return None
+
+    # Set the controller using the pytest marker
+    @pytest.mark.topology(
+        "example", Topology(TopologyDomain("example", client=1)),
+        controller=ExampleController(),
+        fixtures=dict(client="example.client[0]")
+    )
+    def test_marker(client: Client):
+        pass
+
+    # Set the controller in known topology class
+    @final
+    @unique
+    class KnownTopology(KnownTopologyBase):
+        EXAMPLE = TopologyMark(
+            name='example',
+            topology=Topology(TopologyDomain("example", client=1)),
+            controller=ExampleController(),
+            fixtures=dict(client='example.client[0]'),
+        )
+
 .. _setup-and-teardown:
 
 Setup and teardown
@@ -227,22 +281,23 @@ role, and utility objects are executed.
 
         subgraph run [ ]
             subgraph setup [Setup before test]
-                hs(host.setup) --> rs[role.setup]
+                hs(host.setup) --> cs(controller.setup) --> rs[role.setup]
                 rs --> us[utility.setup]
             end
 
             setup -->|run test| teardown
 
             subgraph teardown [Teardown after test]
                 ut[utility.teadown] --> rt[role.teardown]
-                rt --> ht(host.teardown)
+                rt --> ct(controller.teardown)
+                ct --> ht(host.teardown)
             end
         end
 
-        hps -->|run tests| run
-        run -->|all tests finished| hpt(host.pytest_teardown)
+        hps -->|run tests| cts(controller.topopology_setup) -->|run all tests for topology| run
+        run -->|all tests for topology finished| ctt(controller.topology_teardown) -->|all tests finished| hpt(host.pytest_teardown)
         hpt --> e([end])
 
         style run fill:#FFF
         style setup fill:#DFD,stroke-width:2px,stroke:#AFA
-        style teardown fill:#FDD,stroke-width:2px,stroke:#FAA
+        style teardown fill:#FDD,stroke-width:2px,stroke:#FAA

pytest_mh/__init__.py

@@ -17,6 +17,7 @@
 )
 from ._private.plugin import MultihostPlugin, pytest_addoption, pytest_configure
 from ._private.topology import Topology, TopologyDomain
+from ._private.topology_controller import TopologyController
 
 __all__ = [
     "mh",
@@ -32,6 +33,7 @@
     "pytest_addoption",
     "pytest_configure",
     "Topology",
+    "TopologyController",
     "TopologyDomain",
     "TopologyMark",
     "KnownTopologyBase",

pytest_mh/_private/data.py

@@ -27,6 +27,21 @@ def __init__(self, multihost: MultihostConfig | None, topology_mark: TopologyMar
         Test run outcome, available in fixture finalizers.
         """
 
+    def _init(self) -> None:
+        """
+        Postponed initialization. This is called once we know that current
+        mh configuration supports desired topology.
+        """
+        # Initialize topology controller
+        if self.multihost is not None and self.topology_mark is not None:
+            self.topology_mark.controller._init(
+                self.topology_mark.name,
+                self.multihost,
+                self.multihost.logger,
+                self.topology_mark.topology,
+                self.topology_mark.fixtures,
+            )
+
     @staticmethod
     def SetData(item: pytest.Item, data: MultihostItemData | None) -> None:
         item.stash[DataStashKey] = data

pytest_mh/_private/fixtures.py

@@ -1,16 +1,18 @@
 from __future__ import annotations
 
 from types import SimpleNamespace
-from typing import Any, Generator
+from typing import Any, Callable, Generator
 
 import colorama
 import pytest
 
 from .data import MultihostItemData
 from .logging import MultihostLogger
+from .marks import TopologyMark
 from .misc import invoke_callback
 from .multihost import MultihostConfig, MultihostDomain, MultihostHost, MultihostRole
 from .topology import Topology, TopologyDomain
+from .topology_controller import TopologyController
 
 
 class MultihostFixture(object):
@@ -51,7 +53,11 @@ def test_example(mh: MultihostFixture):
     """
 
     def __init__(
-        self, request: pytest.FixtureRequest, data: MultihostItemData, multihost: MultihostConfig, topology: Topology
+        self,
+        request: pytest.FixtureRequest,
+        data: MultihostItemData,
+        multihost: MultihostConfig,
+        topology_mark: TopologyMark,
     ) -> None:
         """
         :param request: Pytest request.
@@ -60,8 +66,8 @@ def __init__(
         :type data: MultihostItemData
         :param multihost: Multihost configuration.
         :type multihost: MultihostConfig
-        :param topology: Multihost topology for this request.
-        :type topology: Topology
+        :param topology_mark: Multihost topology mark.
+        :type topology_mark: TopologyMark
         """
 
         self.data: MultihostItemData = data
@@ -79,11 +85,21 @@ def __init__(
         Multihost configuration.
         """
 
-        self.topology: Topology = topology
+        self.topology_mark: TopologyMark = topology_mark
+        """
+        Topology mark.
+        """
+
+        self.topology: Topology = topology_mark.topology
         """
         Topology data.
         """
 
+        self.topology_controller: TopologyController = topology_mark.controller
+        """
+        Topology controller.
+        """
+
         self.logger: MultihostLogger = multihost.logger
         """
         Multihost logger.
@@ -112,8 +128,8 @@ def __init__(
         self._skipped: bool = False
 
         for domain in self.multihost.domains:
-            if domain.id in topology:
-                setattr(self.ns, domain.id, self._domain_to_namespace(domain, topology.get(domain.id)))
+            if domain.id in self.topology:
+                setattr(self.ns, domain.id, self._domain_to_namespace(domain, self.topology.get(domain.id)))
 
         self.roles = sorted([x for x in self._paths.values() if isinstance(x, MultihostRole)], key=lambda x: x.role)
         self.hosts = sorted(list({x.host for x in self.roles}), key=lambda x: x.hostname)
@@ -153,24 +169,34 @@ def _lookup(self, path: str) -> MultihostRole | list[MultihostRole]:
         return self._paths[path]
 
     def _skip(self) -> bool:
-        if self.data.topology_mark is None:
-            raise ValueError("Multihost fixture is available but no topology mark was set")
-
         self._skipped = False
 
-        fixtures: dict[str, Any] = {k: None for k in self.data.topology_mark.fixtures.keys()}
-        fixtures.update(self.request.node.funcargs)
-        self.data.topology_mark.apply(self, fixtures)
+        reason = self._skip_by_topology(self.topology_controller)
+        if reason is not None:
+            self._skipped = True
+            pytest.skip(reason)
+
+        reason = self._skip_by_require_marker(self.topology_mark, self.request.node)
+        if reason is not None:
+            self._skipped = True
+            pytest.skip(reason)
+
+        return self._skipped
+
+    def _skip_by_topology(self, controller: TopologyController):
+        return controller._invoke_with_args(controller.skip)
+
+    def _skip_by_require_marker(self, topology_mark: TopologyMark, node: pytest.Function) -> str | None:
+        fixtures: dict[str, Any] = {k: None for k in topology_mark.fixtures.keys()}
+        fixtures.update(node.funcargs)
+        topology_mark.apply(self, fixtures)
 
         # Make sure mh fixture is always available
         fixtures["mh"] = self
 
-        for mark in self.request.node.iter_markers("require"):
+        for mark in node.iter_markers("require"):
             if len(mark.args) not in [1, 2]:
-                raise ValueError(
-                    f"{self.request.node.nodeid}::{self.request.node.originalname}: "
-                    "invalid arguments for @pytest.mark.require"
-                )
+                raise ValueError(f"{node.nodeid}::{node.originalname}: " "invalid arguments for @pytest.mark.require")
 
             condition = mark.args[0]
             reason = "Required condition was not met" if len(mark.args) != 2 else mark.args[1]
@@ -179,8 +205,7 @@ def _skip(self) -> bool:
             if isinstance(callresult, tuple):
                 if len(callresult) != 2:
                     raise ValueError(
-                        f"{self.request.node.nodeid}::{self.request.node.originalname}: "
-                        "invalid arguments for @pytest.mark.require"
+                        f"{node.nodeid}::{node.originalname}: " "invalid arguments for @pytest.mark.require"
                     )
 
                 result = callresult[0]
@@ -189,10 +214,27 @@ def _skip(self) -> bool:
                 result = callresult
 
             if not result:
-                self._skipped = True
-                pytest.skip(reason)
+                return reason
 
-        return self._skipped
+        return None
+
+    def _topology_setup(self) -> None:
+        """
+        Run per-test setup from topology controller.
+        """
+        if self._skipped:
+            return
+
+        self.topology_controller._invoke_with_args(self.topology_controller.setup)
+
+    def _topology_teardown(self) -> None:
+        """
+        Run per-test teardown from topology controller.
+        """
+        if self._skipped:
+            return
+
+        self.topology_controller._invoke_with_args(self.topology_controller.teardown)
 
     def _setup(self) -> None:
         """
@@ -296,6 +338,20 @@ def _flush_logs(self) -> None:
         else:
             self.logger.write_to_file(f"{path}/test.log")
 
+    def _invoke_phase(self, name: str, cb: Callable, catch: bool = False) -> Exception | None:
+        self.log_phase(name)
+        try:
+            cb()
+        except Exception as e:
+            if catch:
+                return e
+
+            raise
+        finally:
+            self.log_phase(f"{name} DONE")
+
+        return None
+
     def log_phase(self, phase: str) -> None:
         """
         Log current test phase.
@@ -317,22 +373,22 @@ def __enter__(self) -> MultihostFixture:
             return self
 
         self.log_phase("BEGIN")
-        self.log_phase("SETUP")
-        try:
-            self._setup()
-        finally:
-            self.log_phase("SETUP DONE")
+        self._invoke_phase("SETUP TOPOLOGY", self._topology_setup)
+        self._invoke_phase("SETUP TEST", self._setup)
 
         return self
 
     def __exit__(self, exception_type, exception_value, traceback) -> None:
-        self.log_phase("TEARDOWN")
-        try:
-            self._teardown()
-        finally:
-            self.log_phase("TEARDOWN DONE")
-            self.log_phase("END")
-            self._flush_logs()
+        errors: list[Exception | None] = []
+        errors.append(self._invoke_phase("TEARDOWN TEST", self._teardown, catch=True))
+        errors.append(self._invoke_phase("TEARDOWN TOPOLOGY", self._topology_teardown, catch=True))
+
+        self.log_phase("END")
+        self._flush_logs()
+
+        errors = [x for x in errors if x is not None]
+        if errors:
+            raise Exception(errors)
 
 
 @pytest.fixture(scope="function")
@@ -366,7 +422,7 @@ def mh(request: pytest.FixtureRequest) -> Generator[MultihostFixture, None, None
     if data.topology_mark is None:
         raise ValueError("data.topology_mark must not be None")
 
-    with MultihostFixture(request, data, data.multihost, data.topology_mark.topology) as mh:
+    with MultihostFixture(request, data, data.multihost, data.topology_mark) as mh:
         mh.log_phase("TEST")
         yield mh
         mh.log_phase("TEST DONE")