Multihost Utilities

MultihostUtility can be used to share code between different MultihostRole classes, in addition MultihostReentrantUtility can be used to share code between roles but also between MultihostHost classes.

See also

Pytest-mh already provides several general-purpose utility classes that are ready to use in order to test your project. See Ready to Use Utilities for more information.

MultihostUtility

All instances of MultihostUtility that are available within MultihostRole classes are automatically setup and teardown before and after the test. This can be used to provide high-level API that also cleans up after itself and to share this code between multiple roles.

Example utility to manage local users
 1from typing import Self
 2
 3from pytest_mh import MultihostHost, MultihostUtility
 4from pytest_mh.cli import CLIBuilder, CLIBuilderArgs
 5from pytest_mh.conn import ProcessLogLevel
 6from pytest_mh.utils.fs import LinuxFileSystem
 7
 8
 9class LocalUsersUtils(MultihostUtility[MultihostHost]):
10    """
11    Management of local users.
12
13    .. note::
14
15        All changes are automatically reverted when a test is finished.
16    """
17
18    def __init__(self, host: MultihostHost, fs: LinuxFileSystem) -> None:
19        """
20        :param host: Remote host instance.
21        :type host: MultihostHost
22        """
23        super().__init__(host)
24
25        self.cli: CLIBuilder = host.cli
26        """
27        CLI builder helper.
28        """
29
30        self.fs: LinuxFileSystem = fs
31        """
32        File system manipulation.
33        """
34
35        self._users: list[str] = []
36        """
37        List of local users that were created during the test.
38        """
39
40    def teardown(self) -> None:
41        """
42        Delete any added user and group.
43        """
44        cmd = ""
45
46        if self._users:
47            cmd = "\n".join([f"userdel '{x}' --force --remove" for x in self._users]) + "\n"
48            self.host.conn.run("set -e\n\n" + cmd)
49
50        super().teardown()
51
52    def add_local_user(
53        self,
54        *,
55        name: str,
56        uid: int | None = None,
57        gid: int | None = None,
58        password: str | None = "Secret123",
59        home: str | None = None,
60        gecos: str | None = None,
61        shell: str | None = None,
62    ) -> Self:
63        """
64        Create new local user.
65
66        :param uid: User id, defaults to None
67        :type uid: int | None, optional
68        :param gid: Primary group id, defaults to None
69        :type gid: int | None, optional
70        :param password: Password, defaults to 'Secret123'
71        :type password: str, optional
72        :param home: Home directory, defaults to None
73        :type home: str | None, optional
74        :param gecos: GECOS, defaults to None
75        :type gecos: str | None, optional
76        :param shell: Login shell, defaults to None
77        :type shell: str | None, optional
78        :return: Self.
79        :rtype: Self
80        """
81        if home is not None:
82            self.fs.backup(home)
83
84        args: CLIBuilderArgs = {
85            "name": (self.cli.option.POSITIONAL, name),
86            "uid": (self.cli.option.VALUE, uid),
87            "gid": (self.cli.option.VALUE, gid),
88            "home": (self.cli.option.VALUE, home),
89            "gecos": (self.cli.option.VALUE, gecos),
90            "shell": (self.cli.option.VALUE, shell),
91        }
92
93        passwd = f" && passwd --stdin '{name}'" if password else ""
94        self.logger.info(f'Creating local user "{name}" on {self.host.hostname}')
95        self.host.conn.run(self.cli.command("useradd", args) + passwd, input=password, log_level=ProcessLogLevel.Error)
96
97        self._users.append(name)
98
99        return self

Note

Before a test is run, the hosts are setup multiple times at different scopes and later teardown in the same order (see Setup and Teardown Hooks). For this reason, it is not possible to use MultihostUtility objects in MultihostHost because it can not guarantee that its setup() and teardown() methods are called at proper places.

In theory, it is possible, if you know what you are doing and call setup and teardown manually at desired place. However, it is not possible to call these methods multiple times, so you can only use it within a single setup scope (e.g. only in MultihostHost.pytest_setup). It is therefore highly recommended to use only MultihostReentrantUtility in host objects.

MultihostReentrantUtility

MultihostReentrantUtility objects are designed to work with multiple setup scopes and therefore can be safely used inside MultihostHost. You can understand a setup scope as a pair of setup and teardown hooks, every code that is executed between these calls is a setup scope. pytest-mh currently defines the following scopes:

Setup scopes
  | MultihostHost.pytest_setup
  |      | TopologyController.topology_setup
S |    T |      |
E |    O |      |
S |    P |    T | MultihostHost.setup
S |    O |    E | TopologyController.setup
I |    L |    S | TopologyController.teardown
O |    O |    T | MultihostHost.teardown
N |    G |      |
  |    Y |      |
  |      | TopologyController.topology_teardown
  | MultihostHost.pytest_teardown

All instances of MultihostReentrantUtility are “entered” (MultihostReentrantUtility.__enter__) when entering a new setup scope and “exited” (MultihostReentrantUtility.__exit__) when the setup scope is leaved. The implementation of the utility is expected to save its state in __enter__ and restore to this state in __exit__ – revert all changes that where done inside the setup scope when the scope is leaved.

A typical use case is to use the LinuxFileSystem utility to write or modify a configuration file. Since it is a reentrant utility, it is possible to write a common configuration of your service in MultihostHost.pytest_setup (state=A) and then further modify it in TopologyController.pytest_setup (state=B). The configuration is in state B for all tests for given topology. Once all tests for this topology are finished, the configuration is restored to state A and ready for next topology to be run.

State changes
MultihostHost.pytest_setup (None -> state A)

    TopologyController_1.topology_setup (state A -> state B)
          | test_for_topology_1__a
        B | test_for_topology_1__b
          | test_for_topology_1__c
    TopologyController_1.topology_teardown (state B -> state A)

    TopologyController_2.topology_setup (state A -> state C)
          | test_for_topology_2__a
        C | test_for_topology_2__b
          | test_for_topology_2__c
    TopologyController_2.topology_teardown (state C -> state A)

MultihostHost.pytest_teardown (state A -> None)

The setup and teardown methods of MultihostReentrantUtility are still being called, although it is expected that they will not be used in most implementations. They are, however, called only once: before MultihostHost.pytest_setup and after MultihostHost.pytest_teardown. The following snippet illustrates when the methods are called:

Reentrant utilities callstack
setup host utilities
enter host utilities
MultihostHost.pytest_setup

    enter host utilities
    TopologyController.topology_setup

        enter host utilities
        MultihostHost.setup
        TopologyController.setup

            test_a
            test_b
            ...

        TopologyController.teardown
        MultihostHost.teardown
        exit host utilities

    TopologyController.topology_teardown
    exit host utilities

MultihostHost.pytest_teardown
exit host utilities
teardown host utilities

We can modify the LocalUsersUtils and convert it into a reentrant version so we can safely add users even inside host and topology setup, see the following example.

Reentrant version of user management
  1from collections import deque
  2from typing import Self
  3
  4from pytest_mh import MultihostHost, MultihostReentrantUtility
  5from pytest_mh.cli import CLIBuilder, CLIBuilderArgs
  6from pytest_mh.conn import ProcessLogLevel
  7from pytest_mh.utils.fs import LinuxFileSystem
  8
  9
 10class LocalUsersUtils(MultihostReentrantUtility[MultihostHost]):
 11    """
 12    Management of local users.
 13
 14    .. note::
 15
 16        All changes are automatically reverted when a test is finished.
 17    """
 18
 19    def __init__(self, host: MultihostHost, fs: LinuxFileSystem) -> None:
 20        """
 21        :param host: Remote host instance.
 22        :type host: MultihostHost
 23        """
 24        super().__init__(host)
 25
 26        self.cli: CLIBuilder = host.cli
 27        """
 28        CLI builder helper.
 29        """
 30
 31        self.fs: LinuxFileSystem = fs
 32        """
 33        File system manipulation.
 34        """
 35
 36        self._states: deque[list[str]] = deque()
 37        """
 38        Stored state for each setup scope.
 39        """
 40
 41        self._users: list[str] = []
 42        """
 43        List of local users that were created during the test.
 44        """
 45
 46    def __enter__(self) -> Self:
 47        """
 48        Save current state.
 49
 50        :return: Self.
 51        :rtype: Self
 52        """
 53        self._states.append(self._users)
 54        self._users = []
 55
 56        return self
 57
 58    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
 59        """
 60        Revert all changes done during current context.
 61        """
 62        if self._users:
 63            cmd = "\n".join([f"userdel '{x}' --force --remove" for x in self._users]) + "\n"
 64            self.host.conn.run("set -e\n\n" + cmd)
 65
 66        self._users = self._states.pop()
 67
 68    def add_local_user(
 69        self,
 70        *,
 71        name: str,
 72        uid: int | None = None,
 73        gid: int | None = None,
 74        password: str | None = "Secret123",
 75        home: str | None = None,
 76        gecos: str | None = None,
 77        shell: str | None = None,
 78    ) -> Self:
 79        """
 80        Create new local user.
 81
 82        :param uid: User id, defaults to None
 83        :type uid: int | None, optional
 84        :param gid: Primary group id, defaults to None
 85        :type gid: int | None, optional
 86        :param password: Password, defaults to 'Secret123'
 87        :type password: str, optional
 88        :param home: Home directory, defaults to None
 89        :type home: str | None, optional
 90        :param gecos: GECOS, defaults to None
 91        :type gecos: str | None, optional
 92        :param shell: Login shell, defaults to None
 93        :type shell: str | None, optional
 94        :return: Self.
 95        :rtype: Self
 96        """
 97        if home is not None:
 98            self.fs.backup(home)
 99
100        args: CLIBuilderArgs = {
101            "name": (self.cli.option.POSITIONAL, name),
102            "uid": (self.cli.option.VALUE, uid),
103            "gid": (self.cli.option.VALUE, gid),
104            "home": (self.cli.option.VALUE, home),
105            "gecos": (self.cli.option.VALUE, gecos),
106            "shell": (self.cli.option.VALUE, shell),
107        }
108
109        passwd = f" && passwd --stdin '{name}'" if password else ""
110        self.logger.info(f'Creating local user "{name}" on {self.host.hostname}')
111        self.host.conn.run(self.cli.command("useradd", args) + passwd, input=password, log_level=ProcessLogLevel.Error)
112
113        self._users.append(name)
114
115        return self

Creating more setup-scopes in tests

The main purpose of MultihostUtility is to share code between roles; the main purpose of MultihostReentrantUtility is to share code between hosts. However, they are implemented using Python’s context management functions and therefore it is also possible to pass them into the with statement. This can be used to create additional scopes within a test, if needed.

The following example illustrates how you can change a file multiple times and keep reverting it to its previous state every time the context manager is destroyed.

Reentrant utility in tests
 1    @pytest.mark.topology(...)
 2    def test_ad_hoc_util(example: ExampleRole) -> None:
 3        with example.fs as fs_a:
 4            fs_a.write("/root/test", "content_a")
 5
 6            with fs_a as fs_b:
 7                fs_b.write("/root/test", "content_b")
 8
 9                with fs_b as fs_c:
10                    fs_c.write("/root/test", "content_c")
11                    assert fs_c.read("/root/test") == "content_c"
12
13                # content is restored to "content_b" here since fs_c.__exit__ was called
14
15                assert fs_b.read("/root/test") == "content_b"
16
17            # content is restored to "content_a" here since fs_b.__exit__ was called
18
19            assert fs_a.read("/root/test") == "content_a"
20
21         # content is restored to original content (probably file was deleted) here since fs_a.__exit__ was called

Postponing utility setup

Some utilities may require a complex setup method that consumes some time, but at the same time these utilities can be used in your tests only sporadically, therefore it does not make sense to run the setup for tests that do not actually use it. For this purpose, it is possible to postpone setup of the utility to a place when it is used for the first time.

It is possible to mark the utility with a decorator mh_utility_postpone_setup() or run MultihostUtility.postpone_setup when it is instantiated. Either way, the result is the same but calling the method gives you more control if you want to see different behavior in different roles or hosts.

Examples of postpone utility
1from pytest_mh import mh_utility_postpone_setup
2
3@mh_utility_postpone_setup
4class ExampleUtility(MultihostUtility):
5    def setup(self):
6        pass
7
8    def teardown(self):
9        pass

1class MyRole(MultihostRole):
2    def __init__(self, *args, **kwargs):
3        super().__init__(*args, **kwargs)
4
5        self.firewall: Firewalld = Firewalld(self.host).postpone_setup()

Creating ad-hoc utilities

Sometimes, the utility is used so rarely that it does not make sense to include it in the role object at all. At such time, it is possible to create it directly in the test. The setup and teardown, enter and exit methods are called automatically.

Ad-hoc utility usage
1    from pytest_mh.utils.fs import LinuxFileSystem
2
3    @pytest.mark.topology(...)
4    def test_ad_hoc_util(example: ExampleRole) -> None:
5        with mh_utility(LinuxFileSystem(role.host)) as fs:
6            fs.write("/root/test", "content")
7            assert fs.read("/root/test") == "content"