pytest_mh

Functions

`mh`(request)	Pytest multihost fixture.
`mh_config`(mh)	Multihost configuration.
`mh_logger`(mh)	Multihost logger.
`mh_topology`(mh)	Current topology.
`mh_topology_mark`(mh)	Current topology mark.
`mh_topology_name`(mh)	Current topology name.
`mh_utility`(util)	On-demand use of a multihost utility object with a context manager.
`mh_utility_postpone_setup`(cls)	Class decorator that will postpone calling setup of `MultihostUtility`.
`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.
`mh_fixture`([fixture_function, scope])	This creates a function-scoped pytest fixture that can access MultihostRole objects that are available to the test directly.
`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.
`MultihostLogger`(args, *kwargs)	Multihost logger class.
`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`(args, *kwargs)	Base multihost host class.
`MultihostBackupHost`(args, *kwargs)	Abstract class implementing automatic backup and restore for a host.
`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`(args, *kwargs)	Base role class.
`MultihostTopologyControllerArtifacts`()	Manage set of artifacts that are collected at specific places.
`MultihostUtility`(args, *kwargs)	Base class for utility functions that operate on remote hosts, such as writing a file or managing SSSD.
`MultihostReentrantUtility`(args, *kwargs)	Reentrant multihost utility.
`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.
`BackupTopologyController`()	Implements automatic backup and restore of all topology hosts that inherit from `MultihostBackupHost`.
`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

pytest_mh.mh_config(mh: MultihostFixture) → MultihostConfig

Multihost configuration.

Parameters:: mh (MultihostFixture) – mh fixture
Returns:: Multihost configuration
Return type:: MultihostConfig

pytest_mh.mh_logger(mh: MultihostFixture) → MultihostLogger

Multihost logger.

Can be used to log messages into the test log.

Parameters:: mh (MultihostFixture) – mh fixture
Returns:: Multihost logger.
Return type:: MultihostLogger

pytest_mh.mh_topology(mh: MultihostFixture) → Topology

Current topology.

Parameters:: mh (MultihostFixture) – mh fixture
Returns:: Current topology
Return type:: Topology

pytest_mh.mh_topology_mark(mh: MultihostFixture) → TopologyMark

Current topology mark.

Parameters:: mh (MultihostFixture) – mh fixture
Returns:: Current topology mark
Return type:: TopologyMark

pytest_mh.mh_topology_name(mh: MultihostFixture) → str

Current topology name.

Parameters:: mh (MultihostFixture) – mh fixture
Returns:: Current topology name
Return type:: str

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 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]'),
    )


@final
@unique
class KnownTopologyGroup(KnownTopologyGroupBase):
    All = [
        KnownTopology.A,
        KnownTopology.B,
    ]


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

class pytest_mh.MultihostLogger(*args, **kwargs)

Bases: Logger

Multihost logger class.

It extends the standard logger with additional colorize() method that can be used to put some colors into the log message.

It also allows to log extra data, that are printed in a formatted way together with the message.

logger.info(
    'Main message'',
    extra={'data': {
        'Field1': 'value1',
        'Field2': 'value2',
        ...
    }}
)

Initialize the logger with a name and an optional level.

classmethod GetLogger(*, loggercls: Type[MultihostLogger] | None = None, suffix: str | None = None) → MultihostLogger

Returns the multihost logger.

Parameters:

loggercls (Type[MultihostLogger] | None) – Logger class, defaults to None (= cls).
suffix (str | None) – Logger name suffix, defaults to None.

Returns:

Logger.

Return type:

MultihostLogger

colorize(text: str | Any, *colors: str) → str

Make the text colored with ANSI colors.

Parameters:

text (str | Any) – Text to format. str(text) is called on the parameter.
*colors (colorama.Fore | colorama.Back | colorama.Style) – Colors to apply on the text.

Returns:

Text with colors, if colors are allowed. Unchanged text otherwise.

Return type:

str

critical(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘CRITICAL’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.critical(“Houston, we have a %s”, “major disaster”, exc_info=True)

debug(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘DEBUG’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.debug(“Houston, we have a %s”, “thorny problem”, exc_info=True)

error(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘ERROR’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.error(“Houston, we have a %s”, “major problem”, exc_info=True)

exception(msg, *args, exc_info=True, **kwargs): Convenience method for logging an ERROR with exception information.

fatal(msg, *args, **kwargs): Don’t use this method, use critical() instead.

flush(outcome: Literal['passed', 'failed', 'skipped', 'error', 'unknown'], path: str | Path | None = None) → None

Either write logger content to a file or clear it, depending on the outcome of the test or operation and selected artifacts mode.

Parameters:

outcome (MultihostOutcome) – Test or operation outcome.
path (str | Path | None) – Destination file path, if None split() should be called first, defaults to None.

info(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘INFO’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.info(“Houston, we have a %s”, “interesting problem”, exc_info=True)

log(level, msg, *args, **kwargs)

Log ‘msg % args’ with the integer severity ‘level’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.log(level, “We have a %s”, “mysterious problem”, exc_info=True)

phase(phase: str) → None

Log current phase.

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

setup(**kwargs) → None

Setup multihost logging facility.

Colors are allowed if log_path is /dev/stdout or /dev/stderr.

Parameters:: log_path (str) – Path to the log file.

split(path: str | Path) → None

Move current buffer to a file that will be written later.

The files can be written by write_files()

Parameters:: path (str | Path) – Destination file path.

subclass(cls: Type[MultihostLogger], suffix: str, **kwargs) → MultihostLogger

warn(msg, *args, **kwargs)

warning(msg, *args, **kwargs)

Log ‘msg % args’ with severity ‘WARNING’.

To pass exception information, use the keyword argument exc_info with a true value, e.g.

logger.warning(“Houston, we have a %s”, “bit of a problem”, exc_info=True)

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.

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

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.

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.

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]

confdict: dict[str, Any]: Multihost configuration dictionary given to the constructor.

config: dict[str, Any]: Custom 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

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

Bases: ABC, Generic[ConfigType]

Multihost domain class.

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

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]

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.

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.

property roles: list[str]

All roles available in this domain.

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

confdict: dict[str, Any]: Multihost domain configuration dictionary given to the constructor.

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

mh_config: ConfigType: Multihost configuration

logger: MultihostLogger: Multihost logger

id: str: Domain id

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

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.

log_phase(phase: str) → None

Log current test phase.

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

split_log_file(name: str) → None

Split current log records into a log file.

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

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.

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

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

fixtures: dict[str, MultihostRole | list[MultihostRole]]: All dynamic fixtures defined in the topology mapped from name to MultihostRole.

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

class pytest_mh.MultihostHost(*args, **kwargs)

Bases: Generic[DomainType]

Base multihost host class.

Note

Host objects may contain MultihostReentrantUtility objects. These utilities are automatically setup, entered, exited and teared down.

It may also contain MultihostUtility objects, but setup and teardown of these utilities must be handled manually by in the host or topology setup/teardown methods to create the required scope.

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.

get_artifacts_list(host: MultihostHost, artifacts_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.
artifacts_type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

get_connection() → Connection

Get connection object to the host with given shell.

This creates a connection object using the information from the multihost configuration. The caller should not make any assumptions about the connection mechanism.

Returns:: Generic connection to the host.
Return type:: Connection

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

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

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.

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

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

confdict: dict[str, Any]: Multihost host configuration dictionary given to the constructor.

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.

os_family: MultihostOSFamily: Host operating system os_family.

shell: Shell: Shell used to run commands over host connection.

conn: Connection[Process[ProcessResult, ProcessInputBuffer, ProcessTimeoutError], ProcessResult[ProcessError]]: Connection to the host.

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.

class pytest_mh.MultihostBackupHost(*args, **kwargs)

Bases: MultihostHost[DomainType], ABC

Abstract class implementing automatic backup and restore for a host.

A backup of the host is created once when pytest starts and the host is restored automatically (unless disabled) when a test run is finished.

If the backup data is stored as PurePath or a sequence of PurePath, the file is automatically removed from the host when all tests are finished. Otherwise no action is done – it is possible to overwrite remove_backup() to clean up your data if needed.

It is required to implement start(), stop(), backup() and restore(). The start() method is called in pytest_setup() unless auto_start is set to False and the implementation of this method may raise NotImplementedError which will be ignored.

By default, the host is reverted when each test run is finished. This may not always be desirable and can be disabled via auto_restore parameter of the constructor.

Parameters:

auto_start – Automatically start service before taking the first backup.
auto_restore (bool, optional) – If True, the host is automatically restored to the backup state when a test is finished in teardown(), defaults to True

abstractmethod backup() → PurePath | Sequence[PurePath] | Any | None

Backup backend data.

Returns directory or file path where the backup is stored (as PurePath or sequence of PurePath) or any Python data relevant for the backup. This data is passed to restore() which will use this information to restore the host to its original state.

Returns:: Backup data.
Return type:: PurePath | Sequence[PurePath] | Any | None

pytest_setup() → None: Start the services via start() and take a backup by calling backup().

pytest_teardown() → None: Remove backup files from the host (calls remove_backup()).

remove_backup(backup_data: PurePath | Sequence[PurePath] | Any | None) → None

Remove backup data from the host.

If backup_data is not PurePath or a sequence of PurePath, this will not have any effect. Otherwise, the paths are removed from the host.

Parameters:: backup_data (PurePath | Sequence[PurePath] | Any | None) – Backup data.

abstractmethod restore(backup_data: Any | None) → None

Restore data from the backup.

Parameters:: backup_data (PurePath | Sequence[PurePath] | Any | None) – Backup data.

abstractmethod start() → None

Start required services.

Raises:: NotImplementedError – If start operation is not supported.

abstractmethod stop() → None

Stop required services.

Raises:: NotImplementedError – If stop operation is not supported.

teardown() → None: Restore the host from the backup by calling restore().

backup_data: PurePath | Sequence[PurePath] | Any | None: Backup data of vanilla state of this host.

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.

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]

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.

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

Bases: object

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

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

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

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.

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_output_item_collected(config: Config, item) → None

pytest_report_teststatus(report: CollectReport | TestReport, config: Config) → tuple[str, str, str | tuple[str, dict[str, bool]]] | None

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’.

sigint_handler(sig, frame) → None

class pytest_mh.MultihostRole(*args, **kwargs)

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.

Note

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

get_artifacts_list(host: MultihostHost, artifacts_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.
artifacts_type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

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

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

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.

class pytest_mh.MultihostTopologyControllerArtifacts

Bases: object

Manage set of artifacts that are collected at specific places.

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]

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.

class pytest_mh.MultihostUtility(*args, **kwargs)

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.

Find all MultihostUtility objects in the constructor.

get_artifacts_list(host: MultihostHost, artifacts_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.
artifacts_type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

postpone_setup() → Self

Postpone setup on this instance of MultihostUtility.

Example usage

class MyRole(MultihostRole):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.firewall: Firewalld = Firewalld(self.host).postpone_setup()

Returns:: Self.
Return type:: Self

pytest_report_teststatus(report: CollectReport | TestReport, config: Config) → tuple[str, str, str | tuple[str, dict[str, bool]]] | None

See pytest built-in hook pytest_report_teststatus() for more information.

Warning

This hook is currently called only if report.when == 'call', that is only after the test is run. This may however change in the future, therefore it is recommended to add a test to your code as well.

class Example(MultihostUtility):
    def pytest_report_teststatus(self, report, config):
        if report.when != 'call':
            return None

        return ("error", "X", "MYERROR")

Parameters:

report (pytest.CollectReport | pytest.TestReport) – Pytest report.
config (pytest.Config) – Pytest config.

Returns:

Test status.

Return type:

tuple[str, str, str | tuple[str, Mapping[str, bool]]] | None

setup() → None: Setup object.

teardown() → None: Teardown object.

host: HostType: Multihost host.

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.

class pytest_mh.MultihostReentrantUtility(*args, **kwargs)

Bases: MultihostUtility[HostType]

Reentrant multihost utility.

It provides the __enter__ and __exit__ abstract methods that can be called multiple times in order to create nested states. The implementation of __enter__ should save current state and __exit__ should restore hosts into this state.

The utility can be used as a context manager, leaving the context will restore the system to the state during the context enter.

Find all MultihostUtility objects in the constructor.

pytest_mh.mh_utility(util: MultihostUtility) → Generator[MultihostUtility, None, None]

On-demand use of a multihost utility object with a context manager.

This can be used to automatically setup and teardown a multihost utility object that is created on-demand inside a test.

with mh_utility(LinuxFileSystem(role.host)) as fs:
    fs.write("/root/test", "content")
    with fs:
        fs.write("/root/test", "new_content")
        assert fs.read("/root/test") == "new_content"

    assert fs.read("/root/test") == "content"

pytest_mh.mh_utility_postpone_setup(cls)

Class decorator that will postpone calling setup of MultihostUtility.

Decorated class will not invoke setup() before each test immediately but it will be postponed to the point when the utility is actually used for the first time in the test. This can be used to avoid costly utility setup on utilities that are used only sporadically.

If the utility is not used then setup and teardown method are ignored.

Example

@mh_utility_postpone_setup
class ExampleUtility(MultihostUtility):
    def setup(self):
        pass

    def teardown(self):
        pass

See also

There are other decorators that can affect the behavior of postponed setup.

mh_utility_used()

mh_utility_ignore_use()

Parameters:: cls (type) – Class to decorate.
Returns:: Decorated class.
Return type:: type

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().

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(), 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.mh_fixture(fixture_function: Callable | None = None, *, scope: Literal['function'] = 'function')

This creates a function-scoped pytest fixture that can access MultihostRole objects that are available to the test directly.

Note

For this to work correctly, all multihost fixtures have to be correctly typed. It will not work without the type hints.

At this moment, only function scope is supported.

@mh_fixture()
def my_fixture(client: Client, request: pytest.FixtureRequest):
    pass

@pytest.mark.topology(KnownTopology.LDAP)
def test_example(client: Client, ldap: LDAP, my_fixture):
    pass

Parameters:: scope (Literal["function"], optional) – Fixture scope, defaults to “function”

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.

classmethod FromMultihostConfig(mhc: dict) → Topology

Create Topology from multihost configuration object.

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

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

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

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

class pytest_mh.TopologyController

Bases: Generic[ConfigType]

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 postponed initialization of all present properties therefore you can not access them inside the constructor. The properties are initialized when a test is collected. Override init() instead of the constructor if you need to access these properties from constructor.

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.conn.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

get_artifacts_list(host: MultihostHost, artifacts_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.
artifacts_type (MultihostArtifactsType) – Type of artifacts that are being collected.

Returns:

List of artifacts to collect.

Return type:

set[str]

property hosts: list[MultihostHost]

List of MultihostHost objects available in this topology.

This property cannot be accessed from the constructor.

Returns:: List of MultihostHost objects.
Return type:: list[MultihostHost]

init(name: str, multihost: ConfigType, logger: MultihostLogger, topology: Topology, mapping: dict[str, str])

Postponed initialization of the topology controller, called by the plugin.

All properties are set and accessible after this method is finished.

Parameters:

name (str) – Topology name.
multihost (ConfigType) – MultihostConfig instance.
logger (MultihostLogger) – Multihost logger.
topology (Topology) – Topology.
mapping (dict[str, str]) – Host to fixtures mapping.

property logger: MultihostLogger

Multihost logger.

This property cannot be accessed from the constructor.

Returns:: Multihost logger.
Return type:: MultihostLogger

property multihost: ConfigType

Multihost configuration.

This property cannot be accessed from the constructor.

Returns:: Multihost configuration.
Return type:: MultihostConfig

property name: str

Topology name.

This property cannot be accessed from the constructor.

Returns:: Topology name.
Return type:: str

property ns: SimpleNamespace

Namespace of MultihostHost objects accessible by domain id and roles names.

This property cannot be accessed from the constructor.

Returns:: Namespace.
Return type:: SimpleNamespace

set_artifacts(*args, **kwargs) → None

Called before topology_setup() to set topology artifacts.

Note that the artifacts can be set in any other method as well. This dedicated method is just for your convenience.

setup(*args, **kwargs) → None: Called before execution of each test.

skip(*args, **kwargs) → 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.

Return type:: str | None

teardown(*args, **kwargs) → None: Called after execution of each test.

property topology: Topology

Multihost topology.

This property cannot be accessed from the constructor.

Returns:: Topology.
Return type:: Topology

topology_setup(*args, **kwargs) → None: Called once before executing the first test of given topology.

topology_teardown(*args, **kwargs) → None: Called once after all tests for given topology were run.

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

class pytest_mh.BackupTopologyController

Bases: TopologyController[ConfigType]

Implements automatic backup and restore of all topology hosts that inherit from MultihostBackupHost.

The backup of all hosts is taken in topology_setup(). It is expected that this method is overridden by the user to setup the topology environment. In such case, it is possible to call super().topology_setup(**kwargs) at the end of the overridden function or omit this call and store the backup in backup_data manually.

teardown() restores the hosts to the backup taken in topology_setup(). This is done after each test, so each test starts with clear topology environment.

When all tests for this topology are run, topology_teardown() is called and the hosts are restored to the original state which backup was taken in MultihostBackupHost.pytest_setup() so the environment is fresh for the next topology.

Note

It is possible to decorate methods, usually the custom implementation of topology_setup() with restore_vanilla_on_error(). This makes sure that the hosts are reverted to the original state if any of the setup calls fail.

@BackupTopologyController.restore_vanilla_on_error
def topology_setup(self, *kwargs) -> None:
    raise Exception("Hosts are automatically restored now.")

restore(hosts: dict[MultihostBackupHost, Any | None]) → None

Restore given hosts to their given backup.

Parameters:: hosts (dict[MultihostBackupHost, Any | None]) – Dictionary (host, backup)
Raises:: ExceptionGroup – If some hosts fail to restore.

restore_vanilla() → None

Restore to the original host state that is stored in the host object.

This backup was taken when pytest started and we want to revert to this state when this topology is finished.

static restore_vanilla_on_error(method)

Decorator. Restore all hosts to its original state if an exception occurs during method execution.

Parameters:: method (Any setup or teardown callback.) – Method to decorate.
Returns:: Decorated method.
Return type:: Callback

teardown(*args, **kwargs) → None: Restore the host to the state created by this topology in topology_setup() after each test is finished.

topology_setup(*args, **kwargs) → None: Take backup of all topology hosts.

topology_teardown(*args, **kwargs) → None: Remove all topology backups from the hosts and restore the hosts to the original state before this topology.

backup_data: dict[MultihostBackupHost, Any | None]: Backup data. Dictionary with host as a key and backup as a value.

class pytest_mh.TopologyDomain(id: str, **kwargs: int)

Bases: object

Create a new topology domain.

Topology domain specifies domain id required by the topology as well as required roles and number of hosts that must implement these roles. See Topology for more information.

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

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

Parameters:

id (str) – Domain id.
*kwargs (dict[str, int]) – Required roles.

export() → dict

Export the topology domain into a dictionary object that can be easily converted to JSON, YAML or other formats.

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

Return type:: dict

get(role: str) → int

Find role and return the number of hosts that must implement this role.

Parameters:: role – Host role to lookup.
Raises:: KeyError – The domain was not found.
Return type:: int

satisfies(other: TopologyDomain) → bool

Check if the topology domain satisfies the other domain.

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

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

class pytest_mh.TopologyMark(name: str, topology: Topology, *, controller: TopologyController | None = None, fixtures: dict[str, str] | None = None)

Bases: object

Topology mark is used to describe test case requirements. It defines:

name, that is used to identify topology in pytest output
topology (:class:Topology) that is required to run the test
controller (:class:TopologyController) to provide per-topology hooks, optional
fixtures that are available during the test run, optional

Example usage

@pytest.mark.topology(
    name, topology,
    controller=controller,
    fixtures=dict(fixture1='path1', fixture2='path2', ...)
)
def test_fixture_name(fixture1: BaseRole, fixture2: BaseRole, ...):
    assert True

Fixture path points to a host in the multihost configuration and can be either in the form of $domain-id.$role (all host of given role) or $domain-id.$role[$index] (specific host on given index).

The name is visible in verbose pytest output after the test name, for example:

tests/test_basic.py::test_case (topology-name) PASSED

Parameters:

name (str) – Topology name used in pytest output.
topology (Topology) – Topology required to run the test.
controller (TopologyController | None, optional) – Topology controller, defaults to None
fixtures (dict[str, str] | None, optional) – Dynamically created fixtures available during the test run, defaults to None

classmethod Create(item: Function, mark: Mark) → Self

Create instance of TopologyMark from @pytest.mark.topology.

Raises:: ValueError –
Return type:: Self

classmethod CreateFromArgs(item: Function, args: Tuple, kwargs: Mapping[str, Any]) → Self

Create TopologyMark from pytest.mark.topology arguments.

Warning

This should only be called internally. You can inherit from TopologyMark and override this in order to add additional attributes to the marker.

Parameters:

item (pytest.Function) – Pytest item.
args (Any) – Pytest mark positional arguments.
kwargs (Mapping[str, Any]) – Pytest mark keyword arguments arguments.

Raises:

ValueError – If the marker is invalid.

Returns:

Instance of TopologyMark.

Return type:

Self

classmethod ExpandMarkers(item: Item) → list[Mark]

apply(mh: MultihostFixture, funcargs: dict[str, Any]) → None

Create required fixtures by modifying pytest.Item.funcargs.

Parameters:

mh (MultihostFixture) – Multihost fixture.
funcargs (dict[str, Any]) – Pytest test item funcargs that will be modified.

property args: set[str]: Names of all dynamically created fixtures.

export() → dict

Export the topology mark into a dictionary object that can be easily converted to JSON, YAML or other formats.

{
    'name': 'client',
    'fixtures': { 'client': 'test.client[0]' },
    'topology': [
        {
            'id': 'test',
            'hosts': { 'client': 1 }
        }
    ]
}

Return type:: dict

map_fixtures_to_roles(mh: MultihostFixture) → dict[str, MultihostRole | list[MultihostRole]]

Return all topology fixtures mapped to role object(s).

Parameters:: mh (MultihostFixture) – Multihost fixture.
Returns:: Dynamic fixtures mapped to roles.
Return type:: dict[str, MultihostRole | list[MultihostRole]]

name: str: Topology name.

topology: Topology: Multihost topology.

controller: TopologyController: Multihost topology controller.

fixtures: dict[str, str]: Dynamic fixtures mapping.