From ddb1cafe977eaff4fd9c8517ab1519c34119c1c2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 24 Jan 2024 12:16:29 +0100
Subject: [PATCH 1/7] mh: move role object mapping to ns namespace

To avoid potentional conflict in attributes names.
---
 docs/topology.rst              | 16 ++++++++--------
 pytest_mh/_private/fixtures.py | 13 +++++++++----
 2 files changed, 17 insertions(+), 12 deletions(-)

diff --git a/docs/topology.rst b/docs/topology.rst
index d6b1079..23e2d60 100644
--- a/docs/topology.rst
+++ b/docs/topology.rst
@@ -114,8 +114,8 @@ 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:
 
@@ -123,8 +123,8 @@ The following snippet shows how to access hosts from our topology:
 
     @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,8 +134,8 @@ 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'
@@ -143,8 +143,8 @@ Python type hinting.
 
     @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'
diff --git a/pytest_mh/_private/fixtures.py b/pytest_mh/_private/fixtures.py
index 85b5fa9..e1e8db2 100644
--- a/pytest_mh/_private/fixtures.py
+++ b/pytest_mh/_private/fixtures.py
@@ -46,9 +46,9 @@ class MultihostFixture(object):
         :caption: Example of the MultihostFixture object
 
         def test_example(mh: MultihostFixture):
-            mh.test            # -> namespace containing roles as properties
-            mh.test.client     # -> list of hosts providing given role
-            mh.test.client[0]  # -> host object, instance of specific role
+            mh.ns.test            # -> namespace containing roles as properties
+            mh.ns.test.client     # -> list of hosts providing given role
+            mh.ns.test.client[0]  # -> host object, instance of specific role
     """
 
     def __init__(
@@ -100,6 +100,11 @@ def __init__(
         Available MultihostHost objects.
         """
 
+        self.ns: SimpleNamespace = SimpleNamespace()
+        """
+        Roles as object accessible through topology path, e.g. ``mh.ns.domain_id.role_name``.
+        """
+
         self._opt_artifacts_dir: str = self.request.config.getoption("mh_artifacts_dir")
         self._opt_artifacts_mode: str = self.request.config.getoption("mh_collect_artifacts")
         self._opt_artifacts_compression: bool = self.request.config.getoption("mh_compress_artifacts")
@@ -109,7 +114,7 @@ def __init__(
 
         for domain in self.multihost.domains:
             if domain.id in topology:
-                setattr(self, domain.id, self._domain_to_namespace(domain, topology.get(domain.id)))
+                setattr(self.ns, domain.id, self._domain_to_namespace(domain, 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)

From fe00420de2f4173ca936995d89a0a73de5e93bb7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 24 Jan 2024 13:29:08 +0100
Subject: [PATCH 2/7] misc: add invoke_callback

---
 pytest_mh/_private/fixtures.py | 25 ++-----------
 pytest_mh/_private/misc.py     | 44 +++++++++++++++++++++++
 tests/test_misc.py             | 66 ++++++++++++++++++++++++++++++++++
 3 files changed, 112 insertions(+), 23 deletions(-)
 create mode 100644 tests/test_misc.py

diff --git a/pytest_mh/_private/fixtures.py b/pytest_mh/_private/fixtures.py
index e1e8db2..712073d 100644
--- a/pytest_mh/_private/fixtures.py
+++ b/pytest_mh/_private/fixtures.py
@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-import inspect
-from functools import partial
 from types import SimpleNamespace
 from typing import Any, Generator
 
@@ -10,6 +8,7 @@
 
 from .data import MultihostItemData
 from .logging import MultihostLogger
+from .misc import invoke_callback
 from .multihost import MultihostConfig, MultihostDomain, MultihostHost, MultihostRole
 from .topology import Topology, TopologyDomain
 
@@ -176,27 +175,7 @@ def _skip(self) -> bool:
             condition = mark.args[0]
             reason = "Required condition was not met" if len(mark.args) != 2 else mark.args[1]
 
-            args: list[str] = []
-            if isinstance(condition, partial):
-                spec = inspect.getfullargspec(condition.func)
-
-                # Remove bound positional parameters
-                args = spec.args[len(condition.args) :]
-
-                # Remove bound keyword parameters
-                args = [x for x in args if x not in condition.keywords]
-            else:
-                spec = inspect.getfullargspec(condition)
-                args = spec.args
-
-            if spec.varkw is None:
-                # No **kwargs is present, just pick selected arguments
-                callspec = {k: v for k, v in fixtures.items() if k in args}
-            else:
-                # **kwargs is present, pass everything
-                callspec = fixtures
-
-            callresult = condition(**callspec)
+            callresult = invoke_callback(condition, **fixtures)
             if isinstance(callresult, tuple):
                 if len(callresult) != 2:
                     raise ValueError(
diff --git a/pytest_mh/_private/misc.py b/pytest_mh/_private/misc.py
index 35e6474..3397cf9 100644
--- a/pytest_mh/_private/misc.py
+++ b/pytest_mh/_private/misc.py
@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 from copy import deepcopy
+from functools import partial
+from inspect import getfullargspec
+from typing import Any, Callable
 
 
 def merge_dict(d1, d2):
@@ -18,3 +21,44 @@ def merge_dict(d1, d2):
         dest[key] = value
 
     return dest
+
+
+def invoke_callback(cb: Callable, /, **kwargs: Any) -> Any:
+    """
+    Invoke callback with given arguments.
+
+    The callback may take all or only selected subset of given arguments.
+
+    :param cb: Callback to call.
+    :type cb: Callable
+    :param `*kwargs`: Callback parameters.
+    :type `*kwargs`: dict[str, Any]
+    :param
+    :return: Return value of the callabck.
+    :rtype: Any
+    """
+    cb_args: list[str] = []
+
+    # Get list of parameters required by the callback
+    if isinstance(cb, partial):
+        spec = getfullargspec(cb.func)
+
+        cb_args = spec.args + spec.kwonlyargs
+
+        # Remove bound positional parameters
+        cb_args = cb_args[len(cb.args) :]
+
+        # Remove bound keyword parameters
+        cb_args = [x for x in cb_args if x not in cb.keywords]
+    else:
+        spec = getfullargspec(cb)
+        cb_args = spec.args + spec.kwonlyargs
+
+    if spec.varkw is None:
+        # No **kwargs is present, just pick selected arguments
+        callspec = {k: v for k, v in kwargs.items() if k in cb_args}
+    else:
+        # **kwargs is present, pass everything
+        callspec = kwargs
+
+    return cb(**callspec)
diff --git a/tests/test_misc.py b/tests/test_misc.py
new file mode 100644
index 0000000..dea2441
--- /dev/null
+++ b/tests/test_misc.py
@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from functools import partial
+
+from pytest_mh._private.misc import invoke_callback
+
+
+def test_invoke_callback__exact():
+    def _cb(a, b, c) -> None:
+        assert a == 1
+        assert b == 2
+        assert c == 3
+
+    invoke_callback(_cb, a=1, b=2, c=3)
+    invoke_callback(partial(_cb, 1), b=2, c=3)
+    invoke_callback(partial(_cb, 1, 2), c=3)
+    invoke_callback(partial(_cb, a=1), b=2, c=3)
+
+
+def test_invoke_callback__subset():
+    def _cb(a, b, c) -> None:
+        assert a == 1
+        assert b == 2
+        assert c == 3
+
+    invoke_callback(_cb, a=1, b=2, c=3, d=4, e=5)
+    invoke_callback(partial(_cb, 1), b=2, c=3, d=4, e=5)
+    invoke_callback(partial(_cb, 1, 2), c=3, d=4, e=5)
+    invoke_callback(partial(_cb, a=1), b=2, c=3, d=4, e=5)
+
+
+def test_invoke_callback__kwonly():
+    def _cb(a, *, b, c) -> None:
+        assert a == 1
+        assert b == 2
+        assert c == 3
+
+    invoke_callback(_cb, a=1, b=2, c=3)
+    invoke_callback(partial(_cb, 1), b=2, c=3)
+    invoke_callback(partial(_cb, a=1), b=2, c=3)
+    invoke_callback(partial(_cb, b=2), a=1, c=3)
+
+
+def test_invoke_callback__kwargs():
+    def _cb(**kwargs) -> None:
+        assert kwargs["a"] == 1
+        assert kwargs["b"] == 2
+        assert kwargs["c"] == 3
+
+    invoke_callback(_cb, a=1, b=2, c=3)
+    invoke_callback(partial(_cb, a=1), b=2, c=3)
+    invoke_callback(partial(_cb, b=2), a=1, c=3)
+    invoke_callback(partial(_cb, b=2, c=3), a=1)
+
+
+def test_invoke_callback__kwargs_mixed():
+    def _cb(d, **kwargs) -> None:
+        assert kwargs["a"] == 1
+        assert kwargs["b"] == 2
+        assert kwargs["c"] == 3
+        assert d == 4
+
+    invoke_callback(_cb, a=1, b=2, c=3, d=4)
+    invoke_callback(partial(_cb, 4), a=1, b=2, c=3)
+    invoke_callback(partial(_cb, a=1), b=2, c=3, d=4)
+    invoke_callback(partial(_cb, d=4), a=1, b=2, c=3)

From 29a5ed8e2b0532191b6e094137cf2a244e1e66af Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 24 Jan 2024 13:50:44 +0100
Subject: [PATCH 3/7] misc: make merge_dict more generic

This is not strictly needed, but I wanted to add unit tests so why
not make it future proof when at it.
---
 pytest_mh/_private/misc.py | 22 +++++++-----
 tests/test_misc.py         | 70 +++++++++++++++++++++++++++++++++++++-
 2 files changed, 83 insertions(+), 9 deletions(-)

diff --git a/pytest_mh/_private/misc.py b/pytest_mh/_private/misc.py
index 3397cf9..6e42266 100644
--- a/pytest_mh/_private/misc.py
+++ b/pytest_mh/_private/misc.py
@@ -6,19 +6,25 @@
 from typing import Any, Callable
 
 
-def merge_dict(d1, d2):
+def merge_dict(*args: dict | None):
     """
-    Merge two nested dictionaries together.
+    Merge two or more nested dictionaries together.
 
     Nested dictionaries are not overwritten but combined.
     """
-    dest = deepcopy(d1)
-    for key, value in d2.items():
-        if isinstance(value, dict):
-            dest[key] = merge_dict(dest.get(key, {}), value)
-            continue
+    filtered_args = [x for x in args if x is not None]
+    if not filtered_args:
+        return {}
 
-        dest[key] = value
+    dest = deepcopy(filtered_args[0])
+
+    for source in filtered_args[1:]:
+        for key, value in source.items():
+            if isinstance(value, dict):
+                dest[key] = merge_dict(dest.get(key, {}), value)
+                continue
+
+            dest[key] = value
 
     return dest
 
diff --git a/tests/test_misc.py b/tests/test_misc.py
index dea2441..1b98f17 100644
--- a/tests/test_misc.py
+++ b/tests/test_misc.py
@@ -2,7 +2,75 @@
 
 from functools import partial
 
-from pytest_mh._private.misc import invoke_callback
+import pytest
+
+from pytest_mh._private.misc import invoke_callback, merge_dict
+
+
+@pytest.mark.parametrize(
+    "dicts, expected",
+    [
+        ([None], dict()),
+        ([dict()], dict()),
+        ([None, None], dict()),
+        ([dict(), dict()], dict()),
+        ([dict(), dict(), dict()], dict()),
+        ([None, dict(a="a", b="b")], dict(a="a", b="b")),
+        ([None, dict(a="a", b="b")], dict(a="a", b="b")),
+        ([dict(a="a", b="b"), None], dict(a="a", b="b")),
+        ([dict(), dict(a="a", b="b")], dict(a="a", b="b")),
+        ([dict(a="a", b="b"), dict()], dict(a="a", b="b")),
+        ([dict(a="a"), dict(b="b"), dict(c="c")], dict(a="a", b="b", c="c")),
+        ([dict(a="a", b="b"), dict(c="c")], dict(a="a", b="b", c="c")),
+        (
+            [
+                dict(a="a", b=dict(bb="bb")),
+                dict(c="c"),
+            ],
+            dict(a="a", b=dict(bb="bb"), c="c"),
+        ),
+        (
+            [
+                dict(a="a", b=dict(bb="bb")),
+                dict(b="b"),
+            ],
+            dict(a="a", b="b"),
+        ),
+        (
+            [
+                dict(a="a", b=dict(bb="bb")),
+                dict(b=dict(bb="1")),
+            ],
+            dict(a="a", b=dict(bb="1")),
+        ),
+        (
+            [
+                dict(a="a", b=dict(bb="bb")),
+                dict(b=dict(bc="bc")),
+            ],
+            dict(a="a", b=dict(bb="bb", bc="bc")),
+        ),
+        (
+            [
+                dict(a="a", b=dict(bb="bb")),
+                dict(b=dict(bc="bc")),
+                dict(b=dict(bd="bd")),
+            ],
+            dict(a="a", b=dict(bb="bb", bc="bc", bd="bd")),
+        ),
+        (
+            [
+                dict(a="a", b=dict(bb=dict(bbb="bbb"))),
+                dict(b=dict(bc="bc")),
+                dict(b=dict(bb=dict(bbc="bbc"))),
+            ],
+            dict(a="a", b=dict(bb=dict(bbb="bbb", bbc="bbc"), bc="bc")),
+        ),
+    ],
+)
+def test_merge_dict(dicts, expected):
+    result = merge_dict(*dicts)
+    assert result == expected
 
 
 def test_invoke_callback__exact():

From 9e5b5a44dce0fe0fbfabb006d13eda6c683f2153 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 24 Jan 2024 14:29:20 +0100
Subject: [PATCH 4/7] marks: turn fixtures into keyword argument for topology
 marker

The current state is not possible to extend very easily without
breaking existing setups. This change allows full flexibility
when adding new functionality to the marker.
---
 docs/quick-start.rst        |  2 +-
 docs/topology.rst           | 10 +++++-----
 pytest_mh/_private/marks.py |  9 +++++----
 3 files changed, 11 insertions(+), 10 deletions(-)

diff --git a/docs/quick-start.rst b/docs/quick-start.rst
index 769825c..116bc5c 100644
--- a/docs/quick-start.rst
+++ b/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
diff --git a/docs/topology.rst b/docs/topology.rst
index 23e2d60..b0c4a4c 100644
--- a/docs/topology.rst
+++ b/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
 
@@ -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
diff --git a/pytest_mh/_private/marks.py b/pytest_mh/_private/marks.py
index c3bdbd3..c94e63c 100644
--- a/pytest_mh/_private/marks.py
+++ b/pytest_mh/_private/marks.py
@@ -21,7 +21,7 @@ class TopologyMark(object):
     .. code-block:: python
         :caption: Example usage
 
-        @pytest.mark.topology(name, topology, fixture1='path1', fixture2='path2', ...)
+        @pytest.mark.topology(name, topology, fixture=dict(fixture1='path1', fixture2='path2', ...))
         def test_fixture_name(fixture1: BaseRole, fixture2: BaseRole, ...):
             assert True
 
@@ -40,6 +40,7 @@ def __init__(
         self,
         name: str,
         topology: Topology,
+        *,
         fixtures: dict[str, str] | None = None,
     ) -> None:
         """
@@ -176,7 +177,7 @@ def _CreateFromArgs(cls, item: pytest.Function, args: Tuple, kwargs: Mapping[str
         :return: Instance of TopologyMark.
         :rtype: TopologyMark
         """
-        # First three parameters are positional, the rest are keyword arguments.
+        # First two parameters are positional, the rest are keyword arguments.
         if len(args) != 2:
             nodeid = item.parent.nodeid if item.parent is not None else ""
             error = f"{nodeid}::{item.originalname}: invalid arguments for @pytest.mark.topology"
@@ -184,9 +185,9 @@ def _CreateFromArgs(cls, item: pytest.Function, args: Tuple, kwargs: Mapping[str
 
         name = args[0]
         topology = args[1]
-        fixtures = {k: str(v) for k, v in kwargs.items()}
+        fixtures = {k: str(v) for k, v in kwargs.get("fixtures", {}).items()}
 
-        return cls(name, topology, fixtures)
+        return cls(name, topology, fixtures=fixtures)
 
 
 class KnownTopologyBase(Enum):

From 5753aa1749737646a91f0ec9833d8d9be1808e86 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Tue, 23 Jan 2024 13:44:17 +0100
Subject: [PATCH 5/7] plugin: sort tests by topology name

This is needed in order to allow one-time topology setup and teardown.

If the tests would not be sorted by topology, we would have to run
topology setup and teardown for each test, which has its use cases
but in most majority of cases you want to setup environment for
given topology only once and then run all tests for this topology
in order to speed things up.

We will provide hooks to do both - one-time setup and per-test setup.
---
 pytest_mh/_private/plugin.py | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/pytest_mh/_private/plugin.py b/pytest_mh/_private/plugin.py
index 3a02560..549ce0b 100644
--- a/pytest_mh/_private/plugin.py
+++ b/pytest_mh/_private/plugin.py
@@ -218,8 +218,9 @@ def pytest_collection_modifyitems(self, config: pytest.Config, items: list[pytes
         :meta private:
         """
 
-        selected = []
-        deselected = []
+        selected: list[pytest.Item] = []
+        deselected: list[pytest.Item] = []
+        mapping: dict[str, list[pytest.Item]] = {}
 
         for item in items:
             data = MultihostItemData(self.multihost, item.stash[MarkStashKey]) if self.multihost else None
@@ -229,8 +230,16 @@ def pytest_collection_modifyitems(self, config: pytest.Config, items: list[pytes
                 deselected.append(item)
                 continue
 
-            selected.append(item)
+            # Map test items by topology name so we can sort them later
+            if data is None or data.topology_mark is None:
+                mapping.setdefault("", []).append(item)
+            else:
+                mapping.setdefault(data.topology_mark.name, []).append(item)
+
+        # Sort test by topology name
+        selected = sum([y for _, y in sorted(mapping.items())], [])
 
+        # Yield result to pytest
         config.hook.pytest_deselected(items=deselected)
         items[:] = selected
 

From 57492220bb6f4877f8284495888b7e848d3728fb Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 24 Jan 2024 14:32:59 +0100
Subject: [PATCH 6/7] 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                          | 103 +++++++-
 pytest_mh/__init__.py                     |   2 +
 pytest_mh/_private/data.py                |  15 ++
 pytest_mh/_private/fixtures.py            | 128 +++++++---
 pytest_mh/_private/marks.py               |  22 +-
 pytest_mh/_private/misc.py                |  95 ++++++-
 pytest_mh/_private/plugin.py              |  81 +++++-
 pytest_mh/_private/topology_controller.py | 287 ++++++++++++++++++++++
 8 files changed, 677 insertions(+), 56 deletions(-)
 create mode 100644 pytest_mh/_private/topology_controller.py

diff --git a/docs/classes.rst b/docs/classes.rst
index a18cdd2..a6081e4 100644
--- a/docs/classes.rst
+++ b/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,7 +319,7 @@ 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
 
@@ -235,14 +327,15 @@ role, and utility objects are executed.
 
             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
\ No newline at end of file
+        style teardown fill:#FDD,stroke-width:2px,stroke:#FAA
diff --git a/pytest_mh/__init__.py b/pytest_mh/__init__.py
index 70285b6..82c08db 100644
--- a/pytest_mh/__init__.py
+++ b/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",
diff --git a/pytest_mh/_private/data.py b/pytest_mh/_private/data.py
index 3caa8f7..2e834e3 100644
--- a/pytest_mh/_private/data.py
+++ b/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
diff --git a/pytest_mh/_private/fixtures.py b/pytest_mh/_private/fixtures.py
index 712073d..bb20a54 100644
--- a/pytest_mh/_private/fixtures.py
+++ b/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")
diff --git a/pytest_mh/_private/marks.py b/pytest_mh/_private/marks.py
index c94e63c..2f3703c 100644
--- a/pytest_mh/_private/marks.py
+++ b/pytest_mh/_private/marks.py
@@ -5,9 +5,11 @@
 
 import pytest
 
+from .topology import Topology
+from .topology_controller import TopologyController
+
 if TYPE_CHECKING:
     from .fixtures import MultihostFixture
-    from .topology import Topology
 
 
 class TopologyMark(object):
@@ -16,12 +18,17 @@ class TopologyMark(object):
 
     * **name**, that is used to identify topology in pytest output
     * **topology** (:class:Topology) that is required to run the test
-    * **fixtures** that are available during the test run
+    * **controller** (:class:TopologyController) to provide per-topology hooks, optional
+    * **fixtures** that are available during the test run, optional
 
     .. code-block:: python
         :caption: Example usage
 
-        @pytest.mark.topology(name, topology, fixture=dict(fixture1='path1', fixture2='path2', ...))
+        @pytest.mark.topology(
+            name, topology,
+            controller=controller,
+            fixture=dict(fixture1='path1', fixture2='path2', ...)
+        )
         def test_fixture_name(fixture1: BaseRole, fixture2: BaseRole, ...):
             assert True
 
@@ -41,6 +48,7 @@ def __init__(
         name: str,
         topology: Topology,
         *,
+        controller: TopologyController | None = None,
         fixtures: dict[str, str] | None = None,
     ) -> None:
         """
@@ -48,6 +56,8 @@ def __init__(
         :type name: str
         :param topology: Topology required to run the test.
         :type topology: Topology
+        :param controller: Topology controller, defaults to None
+        :type controller: TopologyController | None, optional
         :param fixtures: Dynamically created fixtures available during the test run, defaults to None
         :type fixtures: dict[str, str] | None, optional
         """
@@ -57,6 +67,9 @@ def __init__(
         self.topology: Topology = topology
         """Multihost topology."""
 
+        self.controller: TopologyController = controller if controller is not None else TopologyController()
+        """Multihost topology controller."""
+
         self.fixtures: dict[str, str] = fixtures if fixtures is not None else {}
         """Dynamic fixtures mapping."""
 
@@ -185,9 +198,10 @@ def _CreateFromArgs(cls, item: pytest.Function, args: Tuple, kwargs: Mapping[str
 
         name = args[0]
         topology = args[1]
+        controller = kwargs.get("controller", None)
         fixtures = {k: str(v) for k, v in kwargs.get("fixtures", {}).items()}
 
-        return cls(name, topology, fixtures=fixtures)
+        return cls(name, topology, controller=controller, fixtures=fixtures)
 
 
 class KnownTopologyBase(Enum):
diff --git a/pytest_mh/_private/misc.py b/pytest_mh/_private/misc.py
index 6e42266..59ebbbe 100644
--- a/pytest_mh/_private/misc.py
+++ b/pytest_mh/_private/misc.py
@@ -3,7 +3,13 @@
 from copy import deepcopy
 from functools import partial
 from inspect import getfullargspec
-from typing import Any, Callable
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, Callable
+
+from .topology import Topology, TopologyDomain
+
+if TYPE_CHECKING:
+    from .multihost import MultihostConfig, MultihostDomain, MultihostHost
 
 
 def merge_dict(*args: dict | None):
@@ -68,3 +74,90 @@ def invoke_callback(cb: Callable, /, **kwargs: Any) -> Any:
         callspec = kwargs
 
     return cb(**callspec)
+
+
+def topology_domain_to_host_namespace(
+    topology_domain: TopologyDomain, mh_domain: MultihostDomain
+) -> tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]:
+    """
+    Convert topology domain into a namespace of MultihostHost objects accessible
+    by roles names.
+
+    :param topology_domain: Topology domain.
+    :type topology_domain: TopologyDomain
+    :param mh_domain: MultihostDomain
+    :type mh_domain: MultihostDomain
+    :return: Pair of namespace and path to host mapping.
+    :rtype: tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]
+    """
+    ns = SimpleNamespace()
+    paths: dict[str, MultihostHost | list[MultihostHost]] = {}
+
+    for role_name in mh_domain.roles:
+        if role_name not in topology_domain:
+            continue
+
+        count = topology_domain.get(role_name)
+        hosts = [host for host in mh_domain.hosts_by_role(role_name)[:count]]
+        setattr(ns, role_name, hosts)
+
+        paths[f"{topology_domain.id}.{role_name}"] = hosts
+        for index, host in enumerate(hosts):
+            paths[f"{topology_domain.id}.{role_name}[{index}]"] = host
+
+    return (ns, paths)
+
+
+def topology_to_host_namespace(
+    topology: Topology, mh_domains: list[MultihostDomain]
+) -> tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]:
+    """
+    Convert topology into a namespace of MultihostHost objects accessible
+    by domain id and roles names.
+
+    :param topology: Topology.
+    :type topology: Topology
+    :param mh_domains: List of MultihostDomain
+    :type mh_domains: list[MultihostDomain]
+    :return: Pair of namespace and path to host mapping.
+    :rtype: tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]
+    """
+    root = SimpleNamespace()
+    paths: dict[str, MultihostHost | list[MultihostHost]] = {}
+
+    for mh_domain in mh_domains:
+        if mh_domain.id in topology:
+            ns, nspaths = topology_domain_to_host_namespace(topology.get(mh_domain.id), mh_domain)
+            setattr(root, mh_domain.id, ns)
+            paths.update(**nspaths)
+
+    return root, paths
+
+
+def topology_controller_parameters(
+    mh_config: MultihostConfig, topology: Topology, fixtures: dict[str, str]
+) -> dict[str, Any]:
+    """
+    Create dictionary of parameters for topology controller hooks.
+
+    :param mh_config: MultihostConfig object.
+    :type mh_config: MultihostConfig
+    :param topology: Topology.
+    :type topology: Topology
+    :param fixtures: Topology fixtures.
+    :type fixtures: dict[str, str]
+    :return: Parameters.
+    :rtype: dict[str, Any]
+    """
+    ns, paths = topology_to_host_namespace(topology, mh_config.domains)
+
+    args = {}
+    for name, path in fixtures.items():
+        args[name] = paths[path]
+
+    return {
+        "mhc": mh_config,
+        "logger": mh_config.logger,
+        "ns": ns,
+        **args,
+    }
diff --git a/pytest_mh/_private/plugin.py b/pytest_mh/_private/plugin.py
index 549ce0b..189e2ab 100644
--- a/pytest_mh/_private/plugin.py
+++ b/pytest_mh/_private/plugin.py
@@ -30,6 +30,7 @@ def __init__(self, pytest_config: pytest.Config) -> None:
         self.multihost: MultihostConfig | None = None
         self.topology: Topology | None = None
         self.confdict: dict | None = None
+        self.current_topology: str | None = None
 
         # CLI options
         self.mh_config: str = pytest_config.getoption("mh_config")
@@ -230,6 +231,10 @@ def pytest_collection_modifyitems(self, config: pytest.Config, items: list[pytes
                 deselected.append(item)
                 continue
 
+            # This test can be run, perform delayed initialization of data.
+            if data is not None:
+                data._init()
+
             # Map test items by topology name so we can sort them later
             if data is None or data.topology_mark is None:
                 mapping.setdefault("", []).append(item)
@@ -259,20 +264,28 @@ def pytest_runtest_setup(self, item: pytest.Item) -> None:
             return
 
         data: MultihostItemData | None = MultihostItemData.GetData(item)
-        if data is None:
+        if self.multihost is None or data is None or data.topology_mark is None:
             return
+        mark: TopologyMark = data.topology_mark
+
+        # Execute per-topology setup if topology is switched.
+        if self._topology_switch(None, item):
+            try:
+                mark.controller._invoke_with_args(mark.controller.topology_setup)
+            finally:
+                self.current_topology = mark.name
+                self.multihost.logger.write_to_file(f"{self.mh_artifacts_dir}/setup_topology_{mark.name}.log")
+
+        # Make mh fixture always available
+        if "mh" not in item.fixturenames:
+            item.fixturenames.append("mh")
 
         # Fill in parameters that will be set later in pytest_runtest_call hook,
         # otherwise pytest will raise unknown fixture error.
-        if data.topology_mark is not None:
-            # Make mh fixture always available
-            if "mh" not in item.fixturenames:
-                item.fixturenames.append("mh")
-
-            spec = inspect.getfullargspec(item.obj)
-            for arg in data.topology_mark.args:
-                if arg in spec.args:
-                    item.funcargs[arg] = None
+        spec = inspect.getfullargspec(item.obj)
+        for arg in mark.args:
+            if arg in spec.args:
+                item.funcargs[arg] = None
 
     @pytest.hookimpl(tryfirst=True)
     def pytest_runtest_call(self, item: pytest.Item) -> None:
@@ -295,6 +308,24 @@ def pytest_runtest_call(self, item: pytest.Item) -> None:
 
             data.topology_mark.apply(mh, item.funcargs)
 
+    @pytest.hookimpl(trylast=True)
+    def pytest_runtest_teardown(self, item: pytest.Item, nextitem: pytest.Item | None) -> None:
+        if self.multihost is None:
+            return
+
+        data: MultihostItemData | None = MultihostItemData.GetData(item)
+        if data is None or data.topology_mark is None:
+            return
+        mark: TopologyMark = data.topology_mark
+
+        # Execute per-topology teardown if topology changed.
+        if self._topology_switch(item, nextitem):
+            try:
+                mark.controller._invoke_with_args(mark.controller.topology_teardown)
+            finally:
+                self.current_topology = None
+                self.multihost.logger.write_to_file(f"{self.mh_artifacts_dir}/teardown_topology_{mark.name}.log")
+
     @pytest.hookimpl(hookwrapper=True)
     def pytest_runtest_makereport(
         self, item: pytest.Item, call: pytest.CallInfo[None]
@@ -398,6 +429,36 @@ def _clone_function(self, name: str, f: pytest.Function) -> pytest.Function:
             originalname=f.originalname,
         )
 
+    def _topology_switch(self, curitem: pytest.Item | None, nextitem: pytest.Item | None) -> bool:
+        # No more items means topology switch for our usecase
+        if nextitem is None:
+            return True
+
+        # If current item is None, we need to check current topology
+        if curitem is None:
+            # This is a first test in the new topology
+            if self.current_topology is None:
+                return True
+            # We always set current_topology to None when switching topologies
+            else:
+                return False
+
+        curdata: MultihostItemData | None = MultihostItemData.GetData(curitem)
+        nextdata: MultihostItemData | None = MultihostItemData.GetData(nextitem)
+
+        if curdata is None or nextdata is None:
+            raise RuntimeError("Data can not be None")
+
+        # If the test does not have topology marker, we consider it a switch
+        if curdata.topology_mark is None or nextdata.topology_mark is None:
+            return True
+
+        # Different topology name is a switch
+        if curdata.topology_mark.name != nextdata.topology_mark.name:
+            return True
+
+        return False
+
 
 # These pytest hooks must be available outside of the plugin's class because
 # they are executed before the plugin is registered.
diff --git a/pytest_mh/_private/topology_controller.py b/pytest_mh/_private/topology_controller.py
new file mode 100644
index 0000000..93cfb3e
--- /dev/null
+++ b/pytest_mh/_private/topology_controller.py
@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import TYPE_CHECKING, Any, Callable
+
+from .logging import MultihostLogger
+from .misc import invoke_callback
+from .topology import Topology, TopologyDomain
+
+if TYPE_CHECKING:
+    from .multihost import MultihostConfig, MultihostDomain, MultihostHost
+
+
+class TopologyController(object):
+    """
+    Topology controller can be associated with a topology via TopologyMark
+    to provide additional per-topology hooks such as per-topology setup
+    and teardown.
+
+    When inheriting from this class, keep it mind that there is postpone
+    initialization of all present properties therefore you can not access
+    them inside the constructor. The properties are initialized a test is
+    collected.
+
+    Each method can take MultihostHost object as parameters as defined in
+    topology fixtures.
+
+    .. 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
+
+        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
+    """
+
+    def __init__(self) -> None:
+        self.__name: str | None = None
+        self.__multihost: MultihostConfig | None = None
+        self.__logger: MultihostLogger | None = None
+        self.__topology: Topology | None = None
+        self.__ns: SimpleNamespace | None = None
+        self.__args: dict[str, MultihostHost | list[MultihostHost]] | None = None
+        self.__initialized: bool = False
+
+    def _init(
+        self,
+        name: str,
+        multihost: MultihostConfig,
+        logger: MultihostLogger,
+        topology: Topology,
+        mapping: dict[str, str],
+    ):
+        # This is called for each testcase but the controller may be shared with
+        # multiple testcases therefore we want to avoid multiple initialization.
+        if self.__initialized:
+            return
+
+        self.__name = name
+        self.__multihost = multihost
+        self.__logger = logger
+        self.__topology = topology
+        self.__ns, self.__args = self._build_namespace_and_args(multihost.domains, topology, mapping)
+
+        self.__initialized = True
+
+    def _build_namespace_and_args(
+        self,
+        mh_domains: list[MultihostDomain],
+        topology: Topology,
+        mapping: dict[str, str],
+    ) -> tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]:
+        root = SimpleNamespace()
+        paths: dict[str, MultihostHost | list[MultihostHost]] = {}
+
+        for mh_domain in mh_domains:
+            if mh_domain.id in topology:
+                ns, nspaths = self._build_domain_namespace_and_paths(topology.get(mh_domain.id), mh_domain)
+                setattr(root, mh_domain.id, ns)
+                paths.update(**nspaths)
+
+        args = {}
+        for name, path in mapping.items():
+            args[name] = paths[path]
+
+        return (root, args)
+
+    def _build_domain_namespace_and_paths(
+        self,
+        topology_domain: TopologyDomain,
+        mh_domain: MultihostDomain,
+    ) -> tuple[SimpleNamespace, dict[str, MultihostHost | list[MultihostHost]]]:
+        ns = SimpleNamespace()
+        paths: dict[str, MultihostHost | list[MultihostHost]] = {}
+
+        for role_name in mh_domain.roles:
+            if role_name not in topology_domain:
+                continue
+
+            count = topology_domain.get(role_name)
+            hosts = [host for host in mh_domain.hosts_by_role(role_name)[:count]]
+            setattr(ns, role_name, hosts)
+
+            paths[f"{topology_domain.id}.{role_name}"] = hosts
+            for index, host in enumerate(hosts):
+                paths[f"{topology_domain.id}.{role_name}[{index}]"] = host
+
+        return (ns, paths)
+
+    def _invoke_with_args(self, cb: Callable) -> Any:
+        if self.__args is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return invoke_callback(cb, **self.__args)
+
+    @property
+    def name(self) -> str:
+        """
+        Topology name.
+
+        This property cannot be accessed from the constructor.
+
+        :return: Topology name.
+        :rtype: str
+        """
+        if self.__name is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return self.__name
+
+    @property
+    def topology(self) -> Topology:
+        """
+        Multihost topology.
+
+        This property cannot be accessed from the constructor.
+
+        :return: Topology.
+        :rtype: Topology
+        """
+        if self.__topology is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return self.__topology
+
+    @property
+    def multihost(self) -> MultihostConfig:
+        """
+        Multihost configuration.
+
+        This property cannot be accessed from the constructor.
+
+        :return: Multihost configuration.
+        :rtype: MultihostConfig
+        """
+        if self.__multihost is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return self.__multihost
+
+    @property
+    def logger(self) -> MultihostLogger:
+        """
+        Multihost logger.
+
+        This property cannot be accessed from the constructor.
+
+        :return: Multihost logger.
+        :rtype: MultihostLogger
+        """
+        if self.__logger is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return self.__logger
+
+    @property
+    def ns(self) -> SimpleNamespace:
+        """
+        Namespace of MultihostHost objects accessible by domain id and roles names.
+
+        This property cannot be accessed from the constructor.
+
+        :return: Namespace.
+        :rtype: SimpleNamespace
+        """
+        if self.__ns is None:
+            raise RuntimeError("TopologyController has not been initialized yet")
+
+        return self.__ns
+
+    def skip(self) -> str | None:
+        """
+        Called before a test is executed.
+
+        If a non-None value is returned the test is skipped, using the returned
+        value as a skip reason.
+
+        :rtype: str | None
+        """
+        return None
+
+    def topology_setup(self) -> None:
+        """
+        Called once before executing the first test of given topology.
+        """
+        pass
+
+    def topology_teardown(self) -> None:
+        """
+        Called once after all tests for given topology were run.
+        """
+        pass
+
+    def setup(self) -> None:
+        """
+        Called before execution of each test.
+        """
+        pass
+
+    def teardown(self) -> None:
+        """
+        Called after execution of each test.
+        """
+        pass

From 32123b511748247046186abaf5d959c3a318a773 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Pavel=20B=C5=99ezina?= <pbrezina@redhat.com>
Date: Wed, 31 Jan 2024 20:41:40 +0100
Subject: [PATCH 7/7] marks: put generic error message in correct scope

---
 pytest_mh/_private/marks.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/pytest_mh/_private/marks.py b/pytest_mh/_private/marks.py
index 2f3703c..6aa99c9 100644
--- a/pytest_mh/_private/marks.py
+++ b/pytest_mh/_private/marks.py
@@ -151,9 +151,10 @@ def Create(cls, item: pytest.Function, mark: pytest.Mark) -> TopologyMark:
         :raises ValueError:
         :rtype: TopologyMark
         """
+        nodeid = item.parent.nodeid if item.parent is not None else ""
+        error = f"{nodeid}::{item.originalname}: invalid arguments for @pytest.mark.topology"
+
         if not mark.args or len(mark.args) > 3:
-            nodeid = item.parent.nodeid if item.parent is not None else ""
-            error = f"{nodeid}::{item.originalname}: invalid arguments for @pytest.mark.topology"
             raise ValueError(error)
 
         # Constructor for KnownTopologyBase