Quick Start Guide

This guide will show you how to setup and extend the pytest-mh plugin. We will write a simple test of Kerberos authentication that spans over two separate hosts - one host has the Kerberos KDC running and the other host will be used as a client machine.

Note

The complete code is located in the example directory of pytest-mh repository.

See also

A real life example of how pytest-mh can help to test your code can be seen in the SSSD project.

All projects are different, therefore pytest_mh plugin provides only the most basic functionality like ssh access to hosts and building blocks to build your own tools and API. It is expected that you implement required functionality in host, role and utility classes by extending MultihostHost, MultihostRole and MultihostUtility.

Since pytest_mh plugin is fully extensible, it is possible to also add your own configuration options and different domain types by extending MultihostConfig and MultihostDomain. This step is actually required as the base classes are abstract and you have to overwrite specific methods and properties in order to give a list of your own domain, host and role classes that will be automatically be instantiated by the plugin.

Note

The difference between host, roles, and utility classes:

• Host classes are created only once before the first test is executed and exist during the whole pytest session. They can be used to setup everything that should live for the whole session.

• Role classes are the main objects that are directly accessible from individual tests. They are created just before the test execution and destroyed once the test is finished. They can perform setup required to run the tests and proper clean up after the test is finished. Roles should also define and implement proper API to access required resources.

• Utility classes are instantiated inside individual roles. They represent functionality that can be shared between roles. They are also responsible to clean up every change that is done through their API. The pytest_mh plugin already has some utility classes bundled within, see pytest_mh.utils.

Create configuration and domain classes

First of all, we need to extend MultihostConfig and tell it how to create our own domain object. Additionally, we need to extend MultihostDomain and define a mapping between role name and host classes and also a mapping between role name and role classes. This tells the plugin which host and role classes should be instantiated for given role.

In the example below, we define two roles: “client” and “kdc”. Each role has its own role (client, KDC) and host class (ClientHost, KDCHost).

/lib/config.py

from __future__ import annotations

from typing import Type

from pytest_mh import MultihostConfig, MultihostDomain, MultihostHost, MultihostRole


class ExampleMultihostConfig(MultihostConfig):
    @property
    def id_to_domain_class(self) -> dict[str, Type[MultihostDomain]]:
        """
        Map domain id to domain class. Asterisk ``*`` can be used as fallback
        value.

        :rtype: Class name.
        """
        return {"*": ExampleMultihostDomain}


class ExampleMultihostDomain(MultihostDomain[ExampleMultihostConfig]):
    @property
    def role_to_host_class(self) -> dict[str, Type[MultihostHost]]:
        """
        Map role to host class. Asterisk ``*`` can be used as fallback value.

        :rtype: Class name.
        """
        from .hosts.client import ClientHost
        from .hosts.kdc import KDCHost

        return {
            "client": ClientHost,
            "kdc": KDCHost,
        }

    @property
    def role_to_role_class(self) -> dict[str, Type[MultihostRole]]:
        """
        Map role to role class. Asterisk ``*`` can be used as fallback value.

        :rtype: Class name.
        """
        from .roles.client import Client
        from .roles.kdc import KDC

        return {
            "client": Client,
            "kdc": KDC,
        }

Note

It is not necessary to create distinct role and host class for every role. The classes can be shared for multiple roles if it makes sense for your project.

Create host classes

KDC Host

The KDC host takes care of backup and restore of the KDC data. It create backup of KDC database when pytest is started and restores it to the original state every time a test is finished. This ensures that the database is always the same for each test execution. It also removes the backup file when pytest is terminated.

/lib/hosts/kdc.py

from __future__ import annotations

from pytest_mh import MultihostHost
from pytest_mh.ssh import SSHPowerShellProcess

from ..config import ExampleMultihostDomain


class KDCHost(MultihostHost[ExampleMultihostDomain]):
    """
    Kerberos KDC server host object.

    Provides features specific to Kerberos KDC.

    .. note::

        Full backup and restore is supported.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        self.__backup_location: str | None = None
        """Backup file or folder location."""

    def pytest_setup(self) -> None:
        """
        Called once before execution of any tests.
        """
        super().setup()

        # Backup KDC data
        self.ssh.run('kdb5_util dump /tmp/mh.kdc.kdb.backup && rm -f "/tmp/mh.kdc.kdb.backup.dump_ok"')
        self.__backup_location = "/tmp/mh.kdc.kdb.backup"

    def pytest_teardown(self) -> None:
        """
        Called once after all tests are finished.
        """
        # Remove backup file
        if self.__backup_location is not None:
            if self.ssh.shell is SSHPowerShellProcess:
                self.ssh.exec(["Remove-Item", "-Force", "-Recurse", self.__backup_location])
            else:
                self.ssh.exec(["rm", "-fr", self.__backup_location])

        super().teardown()

    def teardown(self) -> None:
        """
        Called after execution of each test.
        """
        # Restore KDC data to its original state
        self.ssh.run(f'kdb5_util load "{self.__backup_location}"')
        super().teardown()

Client Host

The client host does not perform any backup and restore as it is not needed, but it reads additional configuration values from the multihost configuration (mhc.yaml) file.

Note

The additional configuration is read from the standard config field which is there for this very reason. But if it makes sense, you can of course extend any section.

/lib/hosts/client.py

from __future__ import annotations

from pytest_mh import MultihostHost

from ..config import ExampleMultihostDomain


class ClientHost(MultihostHost[ExampleMultihostDomain]):
    """
    Kerberos client host object.

    Provides features specific to Kerberos client.

    This class adds ``config.realm``, ``config.krbdomain`` and ``config.kdc``
    multihost configuration options to set the default kerberos realm,
    domain and the kdc hostname.

    .. code-block:: yaml
        :caption: Example multihost configuration
        :emphasize-lines: 6-8

        - hostname: client.test
          role: client
          config:
            realm: TEST
            krbdomain: test
            kdc: kdc.test

    .. note::

        Full backup and restore is supported.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        self.realm: str = self.config.get("realm", "TEST")
        self.krbdomain: str = self.config.get("krbdomain", "test")
        self.kdc: str = self.config.get("kdc", "kdc.test")

Create role classes

Unlike hosts, the role classes are the right place to provide all functionality that will help you write good tests so they are usually quite complex.

KDC Role

The KDC class implements the functionality desired for “kdc” role. In this example, we focus on adding the Kerberos principal (or Kerberos user if you are not familiar with Kerberos terminology) and querying the kadmin tool to get some additional information.

/lib/roles/kdc.py

from __future__ import annotations

from pytest_mh import MultihostRole
from pytest_mh.ssh import SSHProcessResult

from ..hosts.kdc import KDCHost


class KDC(MultihostRole[KDCHost]):
    """
    Kerberos KDC role.

    Provides unified Python API for managing objects in the Kerberos KDC.

    .. code-block:: python
        :caption: Creating user and group

        @pytest.mark.topology(KnownTopology.KDC)
        def test_example(kdc: KDC):
            kdc.principal('tuser').add()

    .. note::

        The role object is instantiated automatically as a dynamic pytest
        fixture by the multihost plugin. You should not create the object
        manually.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

    def kadmin(self, command: str) -> SSHProcessResult:
        """
        Run kadmin command on the KDC.

        :param command: kadmin command
        :type command: str
        """
        result = self.host.ssh.exec(["kadmin.local", "-q", command])

        # Remove "Authenticating as principal root/admin@TEST with password."
        # from the output and keep only output of the command itself.
        result.stdout_lines = result.stdout_lines[1:]
        result.stdout = "\n".join(result.stdout_lines)

        return result

    def list_principals(self) -> list[str]:
        """
        List existing Kerberos principals.

        :return: List of Kerberos principals.
        :rtype: list[str]
        """
        result = self.kadmin("listprincs")
        return result.stdout_lines

    def principal(self, name: str) -> KDCPrincipal:
        """
        Get Kerberos principal object.

        .. code-block:: python
            :caption: Example usage

            @pytest.mark.topology(KnownTopology.KDC)
            def test_example(client: Client, kdc: KDC):
                kdc.principal('tuser').add()

        :param name: Principal name.
        :type name: str
        :return: New principal object.
        :rtype: KDCPrincipal
        """
        return KDCPrincipal(self, name)


class KDCPrincipal(object):
    """
    Kerberos principals management.
    """

    def __init__(self, role: KDC, name: str) -> None:
        """
        :param role: KDC role object.
        :type role: KDC
        :param name: Principal name.
        :type name: str
        """
        self.role: KDC = role
        """KDC role."""

        self.name: str = name
        """Principal name."""

    def add(self, *, password: str | None = "Secret123") -> KDCPrincipal:
        """
        Add a new Kerberos principal.

        Random password is generated if ``password`` is ``None``.

        :param password: Principal's password, defaults to 'Secret123'
        :type password: str | None
        :return: Self.
        :rtype: KDCPrincipal
        """
        if password is not None:
            self.role.kadmin(f'addprinc -pw "{password}" "{self.name}"')
        else:
            self.role.kadmin(f'addprinc -randkey "{self.name}"')

        return self

    def get(self) -> dict[str, str]:
        """
        Retrieve principal information.

        :return: Principal information.
        :rtype: dict[str, str]
        """
        result = self.role.kadmin(f'getprinc "{self.name}"')
        out = {}
        for line in result.stdout_lines:
            (key, value) = line.split(":", maxsplit=1)
            out[key] = value.strip()

        return out

    def delete(self) -> None:
        """
        Delete existing Kerberos principal.
        """
        self.role.kadmin(f'delprinc -force "{self.name}"')

    def set_string(self, key: str, value: str) -> KDCPrincipal:
        """
        Set principal's string attribute.

        :param key: Attribute name.
        :type key: str
        :param value: Atribute value.
        :type value: str
        :return: Self.
        :rtype: KDCPrincipal
        """
        self.role.kadmin(f'setstr "{self.name}" "{key}" "{value}"')
        return self

    def get_strings(self) -> dict[str, str]:
        """
        Get all principal's string attributes.

        :return: String attributes.
        :rtype: dict[str, str]
        """
        result = self.role.kadmin(f'getstrs "{self.name}"')
        out = {}
        for line in result.stdout_lines:
            (key, value) = line.split(":", maxsplit=1)
            out[key] = value.strip()

        return out

    def get_string(self, key: str) -> str | None:
        """
        Set principal's string attribute.

        :param key: Attribute name.
        :type key: str
        :return: Attribute's value or None if not found.
        :rtype: str | None
        """
        attrs = self.get_strings()

        return attrs.get(key, None)

Client Role

The client role first creates /etc/krb5.conf so the Kerberos client knows what KDC we want to use. For this, it uses the bundle LinuxFileSystem utility class, which writes the file to the remote path and when a test is finished, it makes sure to restore the original content or remove the file if it was not present before.

/lib/roles/client.py

"""Client multihost role."""

from __future__ import annotations

import textwrap

from pytest_mh import MultihostRole
from pytest_mh.ssh import SSHProcessError, SSHProcessResult
from pytest_mh.utils.fs import LinuxFileSystem

from ..hosts.client import ClientHost


class Client(MultihostRole[ClientHost]):
    """
    Kerberos client role.

    Provides unified Python API for managing and testing Kerberos client.

    .. note::

        The role object is instantiated automatically as a dynamic pytest
        fixture by the multihost plugin. You should not create the object
        manually.
    """

    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)

        self.realm: str = self.host.realm
        """
        Kerberos realm.
        """

        self.fs: LinuxFileSystem = LinuxFileSystem(self.host)
        """
        File system manipulation.
        """

    def setup(self) -> None:
        """
        Called before execution of each test.

        Setup client host:

        #. Create krb5.conf

        .. note::

            Original krb5.conf is automatically restored when the test is finished.
        """
        super().setup()
        config = textwrap.dedent(
            f"""
            [logging]
            default = FILE:/var/log/krb5libs.log
            kdc = FILE:/var/log/krb5kdc.log
            admin_server = FILE:/var/log/kadmind.log

            [libdefaults]
            default_realm = {self.host.realm}
            default_ccache_name = KCM:
            dns_lookup_realm = false
            dns_lookup_kdc = false
            ticket_lifetime = 24h
            renew_lifetime = 7d
            forwardable = yes

            [realms]
            {self.host.realm} = {{
              kdc = {self.host.kdc}:88
              admin_server = {self.host.kdc}:749
              max_life = 7d
              max_renewable_life = 14d
            }}

            [domain_realm]
            .{self.host.krbdomain} = {self.host.realm}
            {self.host.krbdomain} = {self.host.realm}
        """
        ).lstrip()
        self.fs.write("/etc/krb5.conf", config, user="root", group="root", mode="0644")

    def kinit(
        self, principal: str, *, password: str, realm: str | None = None, args: list[str] | None = None
    ) -> SSHProcessResult:
        """
        Run ``kinit`` command.

        Principal can be without the realm part. The realm can be given in
        separate parameter ``realm``, in such case the principal name is
        constructed as ``$principal@$realm``. If the principal does not contain
        realm specification and ``realm`` parameter is not set then the default
        realm is used.

        :param principal: Kerberos principal.
        :type principal: str
        :param password: Principal's password.
        :type password: str
        :param realm: Kerberos realm that is appended to the principal (``$principal@$realm``), defaults to None
        :type realm: str | None, optional
        :param args: Additional parameters to ``klist``, defaults to None
        :type args: list[str] | None, optional
        :return: Command result.
        :rtype: SSHProcessResult
        """
        if args is None:
            args = []

        if realm is not None:
            principal = f"{principal}@{realm}"

        return self.host.ssh.exec(["kinit", *args, principal], input=password)

    def kvno(self, principal: str, *, realm: str | None = None, args: list[str] | None = None) -> SSHProcessResult:
        """
        Run ``kvno`` command.

        Principal can be without the realm part. The realm can be given in
        separate parameter ``realm``, in such case the principal name is
        constructed as ``$principal@$realm``. If the principal does not contain
        realm specification and ``realm`` parameter is not set then the default
        realm is used.

        :param principal: Kerberos principal.
        :type principal: str
        :param realm: Kerberos realm that is appended to the principal (``$principal@$realm``), defaults to None
        :type realm: str | None, optional
        :param args: Additional parameters to ``klist``, defaults to None
        :type args: list[str] | None, optional
        :return: Command result.
        :rtype: SSHProcessResult
        """
        if args is None:
            args = []

        if realm is not None:
            principal = f"{principal}@{realm}"

        return self.host.ssh.exec(["kvno", *args, principal])

    def klist(self, *, args: list[str] | None = None) -> SSHProcessResult:
        """
        Run ``klist`` command.

        :param args: Additional parameters to ``klist``, defaults to None
        :type args: list[str] | None, optional
        :return: Command result.
        :rtype: SSHProcessResult
        """
        if args is None:
            args = []

        return self.host.ssh.exec(["klist", *args])

    def kswitch(self, principal: str, realm: str) -> SSHProcessResult:
        """
        Run ``kswitch -p principal@realm`` command.

        :param principal: Kerberos principal.
        :type principal: str
        :param realm: Kerberos realm that is appended to the principal (``$principal@$realm``)
        :type realm: str
        :return: Command result.
        :rtype: SSHProcessResult
        """
        if "@" not in principal:
            principal = f"{principal}@{realm}"

        return self.host.ssh.exec(["kswitch", "-p", principal])

    def kdestroy(
        self, *, all: bool = False, ccache: str | None = None, principal: str | None = None, realm: str | None = None
    ) -> SSHProcessResult:
        """
        Run ``kdestroy`` command.

        Principal can be without the realm part. The realm can be given in
        separate parameter ``realm``, in such case the principal name is
        constructed as ``$principal@$realm``. If the principal does not contain
        realm specification and ``realm`` parameter is not set then the default
        realm is used.

        :param all: Destroy all ccaches (``kdestroy -A``), defaults to False
        :type all: bool, optional
        :param ccache: Destroy specific ccache (``kdestroy -c $cache``), defaults to None
        :type ccache: str | None, optional
        :param principal: Destroy ccache for given principal (``kdestroy -p $princ``), defaults to None
        :type principal: str | None, optional
        :param realm: Kerberos realm that is appended to the principal (``$principal@$realm``), defaults to None
        :type realm: str | None, optional
        :return: Command result.
        :rtype: SSHProcessResult
        """
        args = []

        if all:
            args.append("-A")

        if ccache is not None:
            args.append("-c")
            args.append(ccache)

        if realm is not None and principal is not None:
            principal = f"{principal}@{realm}"

        if principal is not None:
            args.append("-p")
            args.append(principal)

        return self.host.ssh.exec(["kdestroy", *args])

    def has_tgt(self, realm: str) -> bool:
        """
        Check that the user has obtained Kerberos Ticket Granting Ticket.

        :param realm: Expected realm for which the TGT was obtained.
        :type realm: str
        :return: True if TGT is available, False otherwise.
        :rtype: bool
        """
        try:
            result = self.klist()
        except SSHProcessError:
            return False

        return f"krbtgt/{realm}@{realm}" in result.stdout

Define multihost topology

Each test is associated with one or more topologies. A topology defines multihost requirements that must be met in order to run the test. If the requirements are not met, the test will not run. These requirements are:

• What domains are available

• What roles and how many roles inside each domain are available

To assign a topology to a test case, we use @pytest.mark.topology(...). The next example defines a topology with one domain that contains one client and one kdc role. Hosts that implements these roles are then available as pytest fixtures.

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

However, this can be little bit cumbersome, therefore it is good practice to define a list of known topologies first.

/lib/topology.py

from __future__ import annotations

from enum import unique
from typing import final

from pytest_mh import KnownTopologyBase, Topology, TopologyDomain, TopologyMark


@final
@unique
class KnownTopology(KnownTopologyBase):
    """
    Well-known topologies that can be given to ``pytest.mark.topology``
    directly. It is expected to use these values in favor of providing
    custom marker values.

    .. code-block:: python
        :caption: Example usage

        @pytest.mark.topology(KnownTopology.KDC)
        def test_kdc(client: Client, kdc: KDC):
            assert True
    """

    KDC = TopologyMark(
        name="kdc",
        topology=Topology(TopologyDomain("test", client=1, kdc=1)),
        fixtures=dict(client="test.client[0]", kdc="test.kdc[0]"),
    )

Now we can shorten the topology marker like this:

@pytest.mark.topology(KnownTopology.KDC)
def test_example(client: Client, kdc: KDC):
    pass

See also

There is also KnownTopologyGroupBase to define a list of topologies that should be assigned to the test case and thus create topology parameterization.

Create multihost configuration

Now, our test framework is ready to use. We just need to provide multihost configuration file that defines available hosts.

We set custom fields that are required by ClientHost and we also define list of artifacts that are automatically fetched from the remote host.

/mhc.yaml

domains:
- id: test
  hosts:
  - hostname: client.test
    role: client
    config:
      realm: TEST
      krbdomain: test
      kdc: kdc.test

  - hostname: kdc.test
    role: kdc
    artifacts:
    - /var/log/krb5kdc.log

Note

The example configuration assumes running containers from sssd-ci-containers project.

Enable pytest-mh in pytest

The pytest-mh plugin needs to be manually enabled in conftest.py and it needs to know the configuration class that should be instantiated.

/conftest.py

# Configuration file for multihost tests.

from __future__ import annotations

from lib.config import ExampleMultihostConfig

from pytest_mh import MultihostPlugin

# Load additional plugins
pytest_plugins = ("pytest_mh",)


# Setup pytest-mh
def pytest_plugin_registered(plugin) -> None:
    if isinstance(plugin, MultihostPlugin):
        plugin.config_class = ExampleMultihostConfig

Write and run a simple test

All the pieces are now available. We have successfully setup the pytest-mh plugin, created our own test framework API. Now it is time to write some tests.

/tests/test_kdc.py

from __future__ import annotations

import pytest
from lib.roles.client import Client
from lib.roles.kdc import KDC
from lib.topology import KnownTopology


@pytest.mark.topology(KnownTopology.KDC)
def test_kinit(client: Client, kdc: KDC):
    kdc.principal("user-1").add(password="Secret123")

    client.kinit("user-1", realm=client.realm, password="Secret123")
    assert client.has_tgt(client.realm)

    client.kdestroy()
    assert not client.has_tgt(client.realm)


@pytest.mark.topology(KnownTopology.KDC)
def test_kvno(client: Client, kdc: KDC):
    kdc.principal("user-1").add(password="Secret123")
    kdc.principal("host/myhost").add()

    client.kinit("user-1", realm=client.realm, password="Secret123")
    assert client.has_tgt(client.realm)

    client.kvno("host/myhost", realm=client.realm)
    assert "host/myhost" in client.klist().stdout

Now we can run them. Notice how the topology name is mentioned in the test name.

$ pytest --mh-config=./mhc.yaml -vv
Multihost configuration:
domains:
- id: test
    hosts:
    - hostname: client.test
      role: client
      config:
        realm: TEST
        krbdomain: test
        kdc: kdc.test
    - hostname: kdc.test
      role: kdc
      artifacts:
      - /var/log/krb5kdc.log

Detected topology:
- id: test
    hosts:
    client: 1
    kdc: 1

Additional settings:
config file: ./mhc.yaml
log path: None
lazy ssh: False
topology filter: None
require exact topology: False
collect artifacts: on-failure
artifacts directory: ./artifacts

============================================================================================================ test session starts =============================================================================================================
platform linux -- Python 3.10.8, pytest-7.2.1, pluggy-1.0.0 -- /home/pbrezina/workspace/pytest-mh/.venv/bin/python3
cachedir: .pytest_cache
rootdir: /home/pbrezina/workspace/pytest-mh, configfile: pytest.ini
collected 2 items

tests/test_kdc.py::test_kinit (kdc) PASSED                                                                                                                                                                                             [ 50%]
tests/test_kdc.py::test_kvno (kdc) PASSED