pytest_mh

Functions

`mh`(request)	Pytest multihost fixture.
`mh_utility_ignore_use`(method)	Decorator for `MultihostUtility` methods defined in inherited classes.
`mh_utility_used`(method)	Decorator for `MultihostUtility` methods defined in inherited classes.
`pytest_addoption`(parser)	Pytest hook: add command line options.
`pytest_configure`(config)	Pytest hook: register multihost plugin.

Classes

`KnownTopologyBase`(value)	Base class for a predefined set of topologies.
`KnownTopologyGroupBase`(value)	Base class for a predefined set of list of topologies.
`MultihostConfig`(confdict, *, logger, ...)	Multihost configuration.
`MultihostDomain`(config, confdict)	Multihost domain class.
`MultihostFixture`(request, data, multihost, ...)	Multihost object provides access to underlaying multihost configuration, individual domains and hosts.
`MultihostHost`(domain, confdict)	Base multihost host class.
`MultihostHostArtifacts`([config])	Manage set of artifacts that are collected at specific places.
`MultihostItemData`(multihost, topology_mark)	Multihost internal pytest data, stored in `pytest.Item.multihost`
`MultihostOSFamily`(value)	Host operating system family.
`MultihostPlugin`(pytest_config)	Pytest multihost plugin.
`MultihostRole`(mh, role, host)	Base role class.
`MultihostTopologyControllerArtifacts`()	Manage set of artifacts that are collected at specific places.
`MultihostUtility`(host)	Base class for utility functions that operate on remote hosts, such as writing a file or managing SSSD.
`Topology`(*domains)	A topology specifies requirements that a multihost configuration must fulfil in order to run a test.
`TopologyController`()	Topology controller can be associated with a topology via TopologyMark to provide additional per-topology hooks such as per-topology setup and teardown.
`TopologyDomain`(id, **kwargs)	Create a new topology domain.
`TopologyMark`(name, topology, *[, ...])	Topology mark is used to describe test case requirements.

pytest_mh.mh(request: FixtureRequest) → Generator[MultihostFixture, None, None]

Pytest multihost fixture. Returns instance of MultihostFixture. When a pytest test is finished, this fixture takes care of tearing down the MultihostFixture object automatically in order to clean up after the test run.

Note

It is preferred that the test case does not use this fixture directly but rather access the hosts through dynamically created role fixtures that are defined in @pytest.mark.topology.

Parameters:: request (pytest.FixtureRequest) – Pytest’s request fixture.
Raises:: ValueError – If not multihost configuration was given.
Yield:: MultihostFixture

class pytest_mh.KnownTopologyBase(value)

Bases: Enum

Base class for a predefined set of topologies.

Users of this plugin may inherit from this class in order to created a predefined, well-known set of topology markers.

Example usage

@final
@unique
class KnownTopology(KnownTopologyBase):
    A = TopologyMark(
        name='A',
        topology=Topology(TopologyDomain('test', a=1)),
        fixtures=dict(a='test.a[0]'),
    )

    B = TopologyMark(
        name='B',
        topology=Topology(TopologyDomain('test', b=1)),
        fixtures=dict(b='test.b[0]'),
    )


@pytest.mark.topology(KnownTopology.A)
def test_a(a: ARole):
    pass

@pytest.mark.topology(KnownTopology.B)
def test_b(b: BRole):
    pass

class pytest_mh.KnownTopologyGroupBase(value)

Bases: Enum

Base class for a predefined set of list of topologies.

Users of this plugin may inherit from this class in order to create a predefined, well-known set of list of topology markers that can be used directly in @pytest.mark.topology to enable topology parametrization for a test case.

Example usage

@final
@unique
class KnownTopologyGroup(KnownTopologyGroupBase):
    All = [
        TopologyMark(
            name='A',
            topology=Topology(TopologyDomain('test', a=1)),
            fixtures=dict(a='test.a[0]', generic='test.a[0]'),
        ),

        B = TopologyMark(
            name='B',
            topology=Topology(TopologyDomain('test', b=1)),
            fixtures=dict(b='test.b[0]', generic='test.a[0]'),
        )
    ]


# Will run once for A, once for B
@pytest.mark.topology(KnownTopologyGroup.All)
def test_all(generic: GenericRole):
    pass

class pytest_mh.MultihostConfig(confdict: dict[str, Any], *, logger: MultihostLogger, lazy_ssh: bool, artifacts_dir: Path, artifacts_mode: Literal['never', 'on-failure', 'always'], artifacts_compression: bool)

Bases: ABC

Multihost configuration.

logger: MultihostLogger: Multihost logger

lazy_ssh: bool: If True, hosts postpone connecting to ssh when the connection is first required

artifacts_dir: Path: Artifacts output directory.

artifacts_mode: Literal['never', 'on-failure', 'always']: Artifacts collection mode.

artifacts_compression: bool: Store artifacts in compressed archive?

domains: list[MultihostDomain]: Available domains

property required_fields: list[str]

Fields that must be set in the host configuration. An error is raised if any field is missing.

The field name may contain a . to check nested fields.

property TopologyMarkClass: Type[TopologyMark]: Class name of the type or subtype of TopologyMark.

create_domain(domain: dict[str, Any]) → MultihostDomain

Create new multihost domain from dictionary.

It maps the role name to a Python class using id_to_domain_class. If the role is not found in the property, it fallbacks to *. If even asterisk is not found, it raises ValueError.

Parameters:: domain (dict[str, Any]) – Domain in dictionary form.
Raises:: ValueError – If domain does not have id or mapping to Python class is not found.
Returns:: New multihost domain.
Return type:: MultihostDomain

topology_hosts(topology: Topology) → list[MultihostHost]

Return all hosts required by the topology as list.

Parameters:: topology (Multihost topology) – Topology.
Returns:: List of MultihostHost.
Return type:: list[MultihostHost]

abstract property id_to_domain_class: dict[str, Type[MultihostDomain]]

Map domain id to domain class. Asterisk * can be used as fallback value.

Return type:: Class name.

class pytest_mh.MultihostDomain(config: ConfigType, confdict: dict[str, Any])

Bases: ABC, Generic[ConfigType]

Multihost domain class.

mh_config: ConfigType: Multihost configuration

logger: MultihostLogger: Multihost logger

id: str: Domain id

hosts: list[MultihostHost]: Available hosts in this domain

property required_fields: list[str]

Fields that must be set in the domain configuration. An error is raised if any field is missing.

The field name may contain a . to check nested fields.

property roles: list[str]

All roles available in this domain.

Returns:: Role names.
Return type:: list[str]

create_host(confdict: dict[str, Any]) → MultihostHost

Create host object from role.

It maps the role name to a Python class using role_to_host_class. If the role is not found in the property, it fallbacks to *. If even asterisk is not found, it fallbacks to MultiHost.

Parameters:: confdict (dict[str, Any]) – Host configuration as a dictionary.
Raises:: ValueError – If role property is missing in the host configuration.
Returns:: Host instance.
Return type:: MultihostHost

create_role(mh: MultihostFixture, host: MultihostHost) → MultihostRole

Create role object from given host.

It maps the role name to a Python class using role_to_role_class. If the role is not found in the property, it fallbacks to *. If even asterisk is not found, it raises ValueError.

Parameters:

mh (Multihost) – Multihost instance.
host (MultihostHost) – Multihost host instance.

Raises:

ValueError – If unexpected role name is given.

Returns:

Role instance.

Return type:

MultihostRole

abstract property role_to_host_class: dict[str, Type[MultihostHost]]

Map role to host class. Asterisk * can be used as fallback value.

Return type:: Class name.

abstract property role_to_role_class: dict[str, Type[MultihostRole]]

Map role to role class. Asterisk * can be used as fallback value.

Return type:: Class name.

hosts_by_role(role: str) → list[MultihostHost]

Return all hosts of the given role.

Parameters:: role (str) – Role name.
Returns:: List of hosts of given role.
Return type:: list[MultihostHost]

class pytest_mh.MultihostFixture(request: FixtureRequest, data: MultihostItemData, multihost: MultihostConfig, topology_mark: TopologyMark)

Bases: object

Multihost object provides access to underlaying multihost configuration, individual domains and hosts. This object should be used only in tests as the mh() pytest fixture.

Domains are accessible as dynamically created properties of this object, hosts are accessible by roles as dynamically created properties of each domain. Each host object is instance of specific role class based on MultihostRole.

Example multihost configuration

domains:
- id: test
  hosts:
  - name: client
    hostname: client.test
    role: client

  - name: ldap
    hostname: master.ldap.test
    role: ldap

The configuration above creates one domain of id test with two hosts. The following example shows how to access the hosts:

Example of the MultihostFixture object

def test_example(mh: MultihostFixture):
    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

Parameters:

request (pytest.FixtureRequest) – Pytest request.
data (MultihostItemData) – Multihost item data.
multihost (MultihostConfig) – Multihost configuration.
topology_mark (TopologyMark) – Multihost topology mark.

data: MultihostItemData: Multihost item data.

request: FixtureRequest: Pytest request.

multihost: MultihostConfig: Multihost configuration.

topology_mark: TopologyMark: Topology mark.

topology: Topology: Topology data.

topology_controller: TopologyController: Topology controller.

logger: MultihostLogger: Multihost logger.

ns: SimpleNamespace: Roles as object accessible through topology path, e.g. mh.ns.domain_id.role_name.

roles: list[MultihostRole]: Available MultihostRole objects.

hosts: list[MultihostHost]: Available MultihostHost objects.

split_log_file(name: str) → None

Split current log records into a log file.

Parameters:: name (str) – Log file name.

log_phase(phase: str) → None

Log current test phase.

Parameters:: phase (str) – Phase name or description.

class pytest_mh.MultihostHost(domain: DomainType, confdict: dict[str, Any])

Bases: Generic[DomainType]

Base multihost host class.

Example configuration in YAML format

- hostname: dc.ad.test
  role: ad
  os:
    family: linux
  ssh:
    host: 1.2.3.4
    username: root
    password: Secret123
  config:
    binddn: Administrator@ad.test
    bindpw: vagrant
    client:
      ad_domain: ad.test
      krb5_keytab: /enrollment/ad.keytab
      ldap_krb5_keytab: /enrollment/ad.keytab

Required fields: hostname, role
Optional fields: artifacts, config, os, ssh

Parameters:

domain (DomainType) – Multihost domain object.
confdict (dict[str, Any]) – Host configuration as a dictionary.
shell (str) – Shell used in SSH connection, defaults to ‘/usr/bin/bash -c’.

mh_domain: DomainType: Multihost domain.

role: str: Host role.

hostname: str: Host hostname.

logger: MultihostLogger: Multihost logger.

config: dict[str, Any]: Custom configuration.

configured_artifacts: MultihostHostArtifacts: Host artifacts produced during tests, configured by the user.

ssh_host: str: SSH host (resolvable hostname or IP address), defaults to hostname.

ssh_port: int: SSH port, defaults to 22.

ssh_username: str: SSH username, defaults to root.

ssh_password: str: SSH password, defaults to Secret123.

os_family: MultihostOSFamily: Host operating system os_family.

shell: Type[SSHProcess]: Shell used in SSH session.

ssh: SSHClient: SSH client.

cli: CLIBuilder: Command line builder.

artifacts: MultihostHostArtifacts: List of artifacts that will be automatically collected at specific places. This list can be dynamically extended. Values may contain wildcard character.

artifacts_collector: MultihostArtifactsCollector: Artifacts collector.

property required_fields: list[str]

Fields that must be set in the host configuration. An error is raised if any field is missing.

The field name may contain a . to check nested fields.

pytest_setup() → None: Called once before execution of any tests.

pytest_teardown() → None: Called once after all tests are finished.

setup() → None: Called before execution of each test.

teardown() → None: Called after execution of each test.

get_artifacts_list(host: MultihostHost, type: Literal['pytest_setup', 'pytest_teardown', 'topology_setup', 'topology_teardown', 'test']) → set[str]

Return the list of artifacts to collect.

This just returns artifacts, but it is possible to override this method in order to generate additional artifacts that were not created by the test, or detect which artifacts were created and update the artifacts list.

Parameters:

host (MultihostHost) – Host where the artifacts are being collected.
type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

class pytest_mh.MultihostHostArtifacts(config: list[str] | dict[str, list[str]] | None = None)

Bases: object

Manage set of artifacts that are collected at specific places.

pytest_setup: set[str]

List of artifacts collected for host after initial pytest_setup.

See MultihostHost.pytest_setup().

pytest_teardown: set[str]

List of artifacts collected for host after final pytest_teardown.

See MultihostHost.pytest_teardown().

test: set[str]: List of artifacts collected for a test when the test run is finished.

get(artifacts_type: Literal['pytest_setup', 'pytest_teardown', 'topology_setup', 'topology_teardown', 'test']) → set[str]

Get list of artifacts by type.

Parameters:: artifacts_type (MultihostArtifactsType) – Type to retrieve.
Raises:: ValueError – If invalid artifacts type is given.
Returns:: List of artifacts.
Return type:: set[str]

class pytest_mh.MultihostItemData(multihost: MultihostConfig | None, topology_mark: TopologyMark | None)

Bases: object

Multihost internal pytest data, stored in pytest.Item.multihost

multihost: MultihostConfig | None: Multihost object.

topology_mark: TopologyMark | None: Topology mark for the test run.

outcome: Literal['passed', 'failed', 'skipped', 'error', 'unknown']: Test run outcome, available in fixture finalizers.

result: TestReport | None: Pytest test result.

static SetData(item: Item, data: MultihostItemData | None) → None

static GetData(item: Item) → MultihostItemData | None

class pytest_mh.MultihostOSFamily(value)

Bases: Enum

Host operating system family.

Linux = 'linux'

Windows = 'windows'

class pytest_mh.MultihostPlugin(pytest_config: Config)

Bases: object

Pytest multihost plugin.

classmethod GetLogger() → Logger: Get plugin’s logger.

pytest_collection_finish(session: Session) → Generator

pytest_runtest_makereport(item: Item, call: CallInfo[None]) → Generator[None, TestReport, None]: Store test outcome in multihost data: item.multihost.outcome. The outcome can be ‘passed’, ‘failed’ or ‘skipped’.

pytest_output_item_collected(config: Config, item) → None

class pytest_mh.MultihostRole(mh: MultihostFixture, role: str, host: HostType)

Bases: Generic[HostType]

Base role class. Roles are the main interface to the remote hosts that can be directly accessed in test cases as fixtures.

All changes to the remote host that were done through the role object API are automatically reverted when a test is finished.

logger: MultihostLogger: Multihost logger.

artifacts: set[str]: List of artifacts that will be automatically collected at specific places. This list can be dynamically extended. Values may contain wildcard character.

setup() → None: Setup all MultihostUtility objects that are attributes of this class.

teardown() → None: Teardown all MultihostUtility objects that are attributes of this class.

get_artifacts_list(host: MultihostHost, type: Literal['pytest_setup', 'pytest_teardown', 'topology_setup', 'topology_teardown', 'test']) → set[str]

Return the list of artifacts to collect.

This just returns artifacts, but it is possible to override this method in order to generate additional artifacts that were not created by the test, or detect which artifacts were created and update the artifacts list.

Parameters:

host (MultihostHost) – Host where the artifacts are being collected.
type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

ssh(user: str, password: str, *, shell=<class 'pytest_mh.ssh.SSHBashProcess'>) → SSHClient

Open SSH connection to the host as given user.

Parameters:

user (str) – Username.
password (str) – User password.
shell (str, optional) – Shell that will run the commands, defaults to SSHBashProcess

Returns:

SSH client connection.

Return type:

SSHClient

class pytest_mh.MultihostTopologyControllerArtifacts

Bases: object

Manage set of artifacts that are collected at specific places.

topology_setup: dict[MultihostHost, set[str]]

List of artifacts collected for host after initial topology_setup.

See TopologyController.topology_setup().

topology_teardown: dict[MultihostHost, set[str]]

List of artifacts collected for host after final topology_teardown.

See TopologyController.topology_teardown().

test: dict[MultihostHost, set[str]]: List of artifacts collected for host when a test run is finished.

get(host: MultihostHost, artifacts_type: MultihostArtifactsType) → set[str]

Get list of artifacts by host and type.

Parameters:: artifacts_type (MultihostArtifactsType) – Type to retrieve.
Raises:: ValueError – If invalid artifacts type is given.
Returns:: List of artifacts.
Return type:: set[str]

class pytest_mh.MultihostUtility(host: HostType)

Bases: Generic[HostType]

Base class for utility functions that operate on remote hosts, such as writing a file or managing SSSD.

Instances of MultihostUtility can be used in any role class which is a subclass of MultihostRole. In this case, setup() and teardown() methods are called automatically when the object is created and destroyed to ensure proper setup and clean up on the remote host.

Note

MultihostUtility uses custom metaclass that inherits from ABCMeta. Therefore all subclasses can use @abstractmethod any other abc decorators without directly inheriting ABCMeta class from ABC.

Parameters:: host (HostType) – Remote host instance.

host: HostType: Multihost host.

logger: MultihostLogger: Multihost logger.

used: bool: Indicate if this utility instance was already used or not within current test.

artifacts: set[str]: List of artifacts that will be automatically collected at specific places. This list can be dynamically extended. Values may contain wildcard character.

setup() → None: Setup object.

teardown() → None: Teardown object.

setup_when_used() → None: Setup the object when it is used for the first time.

teardown_when_used() → None: Teardown the object only if it was used.

get_artifacts_list(host: MultihostHost, type: Literal['pytest_setup', 'pytest_teardown', 'topology_setup', 'topology_teardown', 'test']) → set[str]

Return the list of artifacts to collect.

This just returns artifacts, but it is possible to override this method in order to generate additional artifacts that were not created by the test, or detect which artifacts were created and update the artifacts list.

Parameters:

host (MultihostHost) – Host where the artifacts are being collected.
host – Host where the artifacts are being collected.
type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

static GetUtilityAttributes(o: object) → dict[str, MultihostUtility]

Get all attributes of the o that are instance of MultihostUtility.

Parameters:: o (object) – Any object.
Returns:: Dictionary {attribute name: value}
Return type:: dict[str, MultihostUtility]

classmethod SetupUtilityAttributes(o: object) → None

Setup all MultihostUtility objects attributes of the given object.

Parameters:: o (object) – Any object.

classmethod TeardownUtilityAttributes(o: object) → None

Teardown all MultihostUtility objects attributes of the given object.

Parameters:: o (object) – Any object.

pytest_mh.mh_utility_ignore_use(method)

Decorator for MultihostUtility methods defined in inherited classes.

Decorated method will not count as “using the class” and therefore it will not invoke MultihostUtility.setup_when_called().

Note

This is the opposite of mh_utility_used().

Parameters:: method (Callable) – Method to decorate.
Returns:: Decorated method.
Return type:: Callable

pytest_mh.mh_utility_used(method)

Decorator for MultihostUtility methods defined in inherited classes.

Calling decorated method will first invoke MultihostUtility.setup_when_called(), unless other methods were already called.

Note

Callables and methods decorated with @property are decorated automatically.

This decorator can be used to decorate fields declared in __init__ or descriptors not handled by pytest-mh.

Parameters:: method (Callable) – Method to decorate.
Returns:: Decorated method.
Return type:: Callable

pytest_mh.pytest_addoption(parser): Pytest hook: add command line options.

pytest_mh.pytest_configure(config: Config): Pytest hook: register multihost plugin.

class pytest_mh.Topology(*domains: TopologyDomain)

Bases: object

A topology specifies requirements that a multihost configuration must fulfil in order to run a test.

Each topology consist of one or more domains (TopologyDomain) that defines how many hosts are available inside the domain and what roles are implemented.

The following example defines an ldap topology that consist of one domain of id test and requires two roles: client and ldap each provided by one host.

Topology(
    TopologyDomain(
        'test',
         client=1, ldap=1
    )
)

This topology can be satisfied for example by the following multihost configuration:

domains:
- name: ldap.test
  id: test
  hosts:
  - name: client
    hostname: client.test
    role: client

  - name: ldap
    hostname: master.ldap.test
    role: ldap

Parameters:: *args (TopologyDomain) – Domains that are included in this topology.

get(id: str) → TopologyDomain

Find topology domain of the given id and return it.

Parameters:: id (str) – Topology domain id to lookup.
Raises:: KeyError – The domain was not found.
Return type:: TopologyDomain

export() → list[dict]

Export the topology into a list of dictionaries that can be easily converted to JSON, YAML or other formats.

[
    {
        'id': 'test',
        'roles': {
            'client': 1,
            'ldap': 1
    }
]

Return type:: dict

satisfies(other: Topology) → bool

Check if the topology satisfies the other topology.

Returns True if this topology contains all domains and required roles defined in the other topology and False otherwise.

Parameters:: other (Topology) – The other topology.
Return type:: bool

classmethod FromMultihostConfig(mhc: dict) → Topology

Create Topology from multihost configuration object.

Parameters:: mhc (dict) – Multihost configuration object (dictionary)
Returns:: Inferred topology.
Return type:: Topology

class pytest_mh.TopologyController

Bases: 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.

Example topology controller

class ExampleController(TopologyController):
    def set_artifacts(self, client: ClientHost) -> None:
        self.artifacts.topology_setup[client] = {"/etc/issue"}

    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

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

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

artifacts: MultihostTopologyControllerArtifacts: List of artifacts that will be automatically collected at specific places. This list can be dynamically extended. Values may contain wildcard character.

property name: str

Topology name.