add topology controller

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,97 @@ 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: Example topology controller
+
+    class ExampleController(TopologyController):
+        def skip(self, client: ClientHost) -> str | None:
+            result = client.ssh.run(
+                '''
+                # Implement your requirement check here
+                exit 1
+                ''', raise_on_error=False)
+            if result.rc != 0:
+                return "Topology requirements were not met"
+
+            return None
+
+        def topology_setup(self, client: ClientHost):
+            # One-time setup, prepare the host for this topology
+            # Changes done here are shared for all tests
+            pass
+
+        def topology_teardown(self, client: ClientHost):
+            # One-time teardown, this should undo changes from
+            # topology_setup
+            pass
+
+        def setup(self, client: ClientHost):
+            # Perform per-topology test setup
+            # This is called before execution of every test
+            pass
+
+        def teardown(self, client: ClientHost):
+            # Perform per-topology test teardown, this should undo changes
+            # from setup
+            pass
+
+.. code-block:: python
+    :caption: Example with low-level topology mark
+
+    class ExampleController(TopologyController):
+        # Implement methods you are interested in here
+        pass
+
+    @pytest.mark.topology(
+        "example", Topology(TopologyDomain("example", client=1)),
+        controller=ExampleController(),
+        fixtures=dict(client="example.client[0]")
+    )
+    def test_example(client: Client):
+        pass
+
+.. code-block:: python
+    :caption: Example with KnownTopology (recommended)
+
+    class ExampleController(TopologyController):
+        # Implement methods you are interested in here
+        pass
+
+    @final
+    @unique
+    class KnownTopology(KnownTopologyBase):
+        EXAMPLE = TopologyMark(
+            name='example',
+            topology=Topology(TopologyDomain("example", client=1)),
+            controller=ExampleController(),
+            fixtures=dict(client='example.client[0]'),
+        )
+
+    @pytest.mark.topology(KnownTopology.EXAMPLE)
+    def test_example(client: Client):
+        pass
+
 .. _setup-and-teardown:
 
 Setup and teardown
@@ -227,22 +319,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

docs/quick-start.rst

@@ -165,7 +165,7 @@ fixtures.
 
     @pytest.mark.topology(
         "kdc", Topology(TopologyDomain("test", client=1, kdc=1)),
-        client="test.client[0]", kdc="test.kdc[0]"
+        fixtures=dict(client="test.client[0]", kdc="test.kdc[0]")
     )
     def test_example(client: Client, kdc: KDC):
         pass

docs/topology.rst

@@ -54,7 +54,7 @@ The marker is used as:
 
 .. code-block:: python
 
-    @pytest.mark.topology(name, topology, *, fixtures ...)
+    @pytest.mark.topology(name, topology, *, fixtures=dict(...))
     def test_example():
         assert True
 
@@ -114,17 +114,17 @@ domains (:class:`~pytest_mh.MultihostDomain`) and hosts (as
 
 To access the hosts through the :func:`~pytest_mh.mh` fixture use:
 
-* ``mh.<domain-id>.<role>`` to access a list of all hosts that implements given role
-* ``mh.<domain-id>.<role>[<index>]`` to access a specific host through index starting from 0
+* ``mh.ns.<domain-id>.<role>`` to access a list of all hosts that implements given role
+* ``mh.ns.<domain-id>.<role>[<index>]`` to access a specific host through index starting from 0
 
 The following snippet shows how to access hosts from our topology:
 
 .. code-block:: python
 
     @pytest.mark.topology('ldap', Topology(TopologyDomain('test', client=1, ldap=1)))
     def test_example(mh: MultihostFixture):
-        assert mh.test.client[0].role == 'client'
-        assert mh.test.ldap[0].role == 'ldap'
+        assert mh.ns.test.client[0].role == 'client'
+        assert mh.ns.test.ldap[0].role == 'ldap'
 
 Since the role objects are instances of your own classes (``LDAP`` and
 ``Client`` for our example), you can also set the type to get the advantage of
@@ -134,17 +134,17 @@ Python type hinting.
 
     @pytest.mark.topology('ldap', Topology(TopologyDomain('test', client=1, ldap=1)))
     def test_example(mh: MultihostFixture):
-        client: Client = mh.test.client[0]
-        ldap: LDAP = mh.test.ldap[0]
+        client: Client = mh.ns.test.client[0]
+        ldap: LDAP = mh.ns.test.ldap[0]
 
         assert client.role == 'client'
         assert ldap.role == 'ldap'
 
 
     @pytest.mark.topology('ldap', Topology(TopologyDomain('test', client=1, ldap=1)))
     def test_example2(mh: MultihostFixture):
-        clients: list[Client] = mh.test.client
-        ldaps: list[LDAP] = mh.test.ldap
+        clients: list[Client] = mh.ns.test.client
+        ldaps: list[LDAP] = mh.ns.test.ldap
 
         for client in clients:
             assert client.role == 'client'
@@ -179,7 +179,7 @@ The example above can be rewritten as:
 
     @pytest.mark.topology(
         'ldap', Topology(TopologyDomain('test', client=1, ldap=1)),
-        client='test.client[0]', ldap='test.ldap[0]'
+        fixtures=dict(client='test.client[0]', ldap='test.ldap[0]')
     )
     def test_example(client: Client, ldap: LDAP):
         assert client.role == 'client'
@@ -198,7 +198,7 @@ benefit from it.
 
     @pytest.mark.topology(
         'ldap', Topology(TopologyDomain('test', client=1, ldap=1)),
-        clients='test.client', ldap='test.ldap[0]'
+        fixtures=dict(clients='test.client', ldap='test.ldap[0]')
     )
     def test_example(clients: list[Client], ldap: LDAP):
         for client in clients:
@@ -217,7 +217,7 @@ benefit from it.
 
         @pytest.mark.topology(
             'ldap', Topology(TopologyDomain('test', client=1, ldap=1)),
-            clients='test.client'
+            fixtures=dict(clients='test.client')
         )
         def test_example(mh: MultihostFixture, clients: list[Client]):
             pass
@@ -231,7 +231,7 @@ benefit from it.
 
         @pytest.mark.topology(
             'ldap', Topology(TopologyDomain('test', client=1, ldap=1)),
-            client='test.client[0]', ldap='test.ldap[0]', provider='test.ldap[0]'
+            fixtures=dict(client='test.client[0]', ldap='test.ldap[0]', provider='test.ldap[0]')
         )
         def test_example(client: Client, provider: GenericProvider):
             pass

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