100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
"""Pytest fixtures."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from functools import partial
|
|
|
|
import pytest
|
|
from pytest_mh import MultihostItemData, Topology
|
|
|
|
from .misc import to_list_of_strings
|
|
from .roles.base import BaseRole
|
|
from .topology import KnownTopology, KnownTopologyGroup
|
|
|
|
|
|
def pytest_configure(config: pytest.Config):
|
|
"""
|
|
Pytest hook: register multihost plugin.
|
|
"""
|
|
|
|
# register additional markers
|
|
config.addinivalue_line(
|
|
"markers",
|
|
"builtwith(feature): Run test only if shadow was built with given feature",
|
|
)
|
|
|
|
|
|
def builtwith(item: pytest.Function, requirements: dict[str, str], **kwargs: BaseRole):
|
|
def value_error(msg: str) -> ValueError:
|
|
return ValueError(f"{item.nodeid}::{item.originalname}: @pytest.mark.builtwith: {msg}")
|
|
|
|
errors: list[str] = []
|
|
for role, features in requirements.items():
|
|
if role not in kwargs:
|
|
raise value_error(f"unknown fixture '{role}'")
|
|
|
|
if not isinstance(kwargs[role], BaseRole):
|
|
raise value_error(f"fixture '{role}' is not instance of BaseRole")
|
|
|
|
obj = kwargs[role]
|
|
for feature in to_list_of_strings(features):
|
|
if feature not in obj.features:
|
|
raise value_error(f"unknown feature '{feature}' in '{role}'")
|
|
|
|
if not obj.features[feature]:
|
|
errors.append(f'{role} does not support "{feature}"')
|
|
|
|
if len(errors) == 1:
|
|
return (False, errors[0])
|
|
elif len(errors) > 1:
|
|
return (False, str(errors))
|
|
|
|
# All requirements were passed
|
|
return True
|
|
|
|
|
|
@pytest.hookimpl(tryfirst=True)
|
|
def pytest_runtest_setup(item: pytest.Item) -> None:
|
|
if not isinstance(item, pytest.Function):
|
|
raise TypeError(f"Unexpected item type: {type(item)}")
|
|
|
|
topology: list[Topology] = []
|
|
mh_item_data: MultihostItemData | None = MultihostItemData.GetData(item)
|
|
for mark in item.iter_markers("builtwith"):
|
|
requirements: dict[str, str] = {}
|
|
|
|
if len(mark.args) == 1 and not mark.kwargs:
|
|
# @pytest.mark.builtwith("feature_x")
|
|
# -> check if "feature_x" is supported by shadow
|
|
requirements["shadow"] = mark.args[0]
|
|
topology = []
|
|
elif not mark.args and mark.kwargs:
|
|
# @pytest.mark.builtwith(shadow="feature_x", another_host="feature_x") ->
|
|
# -> check if "feature_x" is supported by both shadow and another_host
|
|
requirements = dict(mark.kwargs)
|
|
topology = []
|
|
elif (
|
|
len(mark.args) == 1
|
|
and isinstance(mark.args[0], (Topology, KnownTopology, KnownTopologyGroup))
|
|
and mark.kwargs
|
|
):
|
|
# @pytest.mark.builtwith(KnownTopology.Shadow, shadow="feature_x") ->
|
|
# -> check if "feature_x" is supported by shadow only if the test runs on shadow topology
|
|
requirements = dict(mark.kwargs)
|
|
if isinstance(mark.args[0], Topology):
|
|
topology = [mark.args[0]]
|
|
elif isinstance(mark.args[0], KnownTopology):
|
|
topology = [mark.args[0].value.topology]
|
|
elif isinstance(mark.args[0], KnownTopologyGroup):
|
|
topology = [x.value.topology for x in mark.args[0].value]
|
|
else:
|
|
raise ValueError(f"{item.nodeid}::{item.originalname}: invalid arguments for @pytest.mark.builtwith")
|
|
|
|
if mh_item_data is None:
|
|
raise ValueError(f"{item.nodeid}::{item.originalname}: multihost item data is not set")
|
|
|
|
if mh_item_data.topology_mark is None:
|
|
raise ValueError(f"{item.nodeid}::{item.originalname}: multihost topology mark is not set")
|
|
|
|
if not topology or mh_item_data.topology_mark.topology in topology:
|
|
item.add_marker(pytest.mark.require(partial(builtwith, item=item, requirements=requirements)))
|