diff --git a/pyproject.toml b/pyproject.toml index e627b1016f..d6a72e8c4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -286,7 +286,6 @@ ignore = [ "tmt/queue.py", "tmt/utils/__init__.py", "tmt/utils/structured_field.py", - "tmt/hardware.py", # pyright does not pick up pint's _typing.py or something :/ ] pythonVersion = "3.9" diff --git a/tests/unit/provision/mrack/test_hw.py b/tests/unit/provision/mrack/test_hw.py index e5d97b93f0..590db3147c 100644 --- a/tests/unit/provision/mrack/test_hw.py +++ b/tests/unit/provision/mrack/test_hw.py @@ -4,9 +4,9 @@ import tmt.utils from tests.unit.test_hardware import FULL_HARDWARE_REQUIREMENTS, OR_HARDWARE_REQUIREMENTS -from tmt.hardware import ( +from tmt.hardware.constraints import Operator +from tmt.hardware.requirements import ( Hardware, - Operator, _parse_cpu, _parse_device, _parse_disk, diff --git a/tests/unit/provision/testcloud/test_hw.py b/tests/unit/provision/testcloud/test_hw.py index cab4894073..b4bf918b7d 100644 --- a/tests/unit/provision/testcloud/test_hw.py +++ b/tests/unit/provision/testcloud/test_hw.py @@ -7,7 +7,8 @@ from testcloud.domain_configuration import DomainConfiguration, TPMConfiguration from tests.unit import MATCH, assert_log -from tmt.hardware import TPM_VERSION_ALLOWED_OPERATORS, Hardware, Operator +from tmt.hardware.constraints import Operator +from tmt.hardware.requirements import TPM_VERSION_ALLOWED_OPERATORS, Hardware from tmt.log import Logger from tmt.steps.provision.testcloud import ( TPM_VERSION_ALLOWED_OPERATORS as virtual_TPM_VERSION_ALLOWED_OPERATORS, # noqa: N811 diff --git a/tests/unit/test_hardware.py b/tests/unit/test_hardware.py index 3daa7340da..ab450cc7a2 100644 --- a/tests/unit/test_hardware.py +++ b/tests/unit/test_hardware.py @@ -10,6 +10,7 @@ import tmt.guest import tmt.hardware +import tmt.hardware.constraints import tmt.utils from tmt.hardware import Hardware from tmt.log import Logger @@ -31,15 +32,15 @@ def parse_hw(text: str) -> Hardware: ] + [ (f'{operator.value} 10', (operator.value, '10')) - for operator in tmt.hardware.INPUTABLE_OPERATORS + for operator in tmt.hardware.constraints.INPUTABLE_OPERATORS ] + [ (f'{operator.value} 10 GiB', (operator.value, '10 GiB')) - for operator in tmt.hardware.INPUTABLE_OPERATORS + for operator in tmt.hardware.constraints.INPUTABLE_OPERATORS ] + [ (f'{operator.value}10GiB', (operator.value, '10GiB')) - for operator in tmt.hardware.INPUTABLE_OPERATORS + for operator in tmt.hardware.constraints.INPUTABLE_OPERATORS ] ) @@ -53,7 +54,7 @@ def parse_hw(text: str) -> Hardware: ], ) def test_constraint_value_pattern(value: str, expected: tuple[Any, Any]) -> None: - match = tmt.hardware.CONSTRAINT_VALUE_PATTERN.match(value) + match = tmt.hardware.constraints.CONSTRAINT_VALUE_PATTERN.match(value) assert match is not None assert match.groups() == expected @@ -75,7 +76,7 @@ def test_constraint_value_pattern(value: str, expected: tuple[Any, Any]) -> None ], ) def test_constraint_name_pattern(value: str, expected: tuple[Any, Any]) -> None: - match = tmt.hardware.CONSTRAINT_NAME_PATTERN.match(value) + match = tmt.hardware.constraints.CONSTRAINT_NAME_PATTERN.match(value) assert match is not None assert match.groups() == expected @@ -128,7 +129,7 @@ def test_constraint_default_unit(value: dict, expected: tuple[Any, Any]) -> None ], ) def test_constraint_components_pattern(value: str, expected: tuple[Any, Any]) -> None: - match = tmt.hardware.CONSTRAINT_COMPONENTS_PATTERN.match(value) + match = tmt.hardware.constraints.CONSTRAINT_COMPONENTS_PATTERN.match(value) assert match is not None assert match.groups() == expected @@ -180,7 +181,10 @@ def test_normalize_hardware(root_logger) -> None: ], ) def test_normalize_invalid_hardware( - spec: tmt.hardware.Spec, expected_exc: type[Exception], expected_message: str, root_logger + spec: tmt.hardware.constraints.Spec, + expected_exc: type[Exception], + expected_message: str, + root_logger, ) -> None: with pytest.raises(expected_exc, match=expected_message): tmt.guest.normalize_hardware('', spec, root_logger) @@ -441,15 +445,15 @@ def _test_check(constraint: tmt.hardware.Constraint) -> bool: @pytest.mark.parametrize( ('operator', 'left', 'right', 'expected'), [ - (tmt.hardware.not_contains, ['foo'], 'foo', False), - (tmt.hardware.not_contains, ['foo', 'bar'], 'foo', False), - (tmt.hardware.not_contains, ['foo'], 'bar', True), - (tmt.hardware.not_contains_exclusive, ['foo'], 'foo', False), - (tmt.hardware.not_contains_exclusive, ['foo', 'bar'], 'foo', True), - (tmt.hardware.not_contains_exclusive, ['foo'], 'bar', True), + (tmt.hardware.constraints.not_contains, ['foo'], 'foo', False), + (tmt.hardware.constraints.not_contains, ['foo', 'bar'], 'foo', False), + (tmt.hardware.constraints.not_contains, ['foo'], 'bar', True), + (tmt.hardware.constraints.not_contains_exclusive, ['foo'], 'foo', False), + (tmt.hardware.constraints.not_contains_exclusive, ['foo', 'bar'], 'foo', True), + (tmt.hardware.constraints.not_contains_exclusive, ['foo'], 'bar', True), ], ) def test_operators( - operator: tmt.hardware.OperatorHandlerType, left: Any, right: Any, expected: bool + operator: tmt.hardware.constraints.OperatorHandlerType, left: Any, right: Any, expected: bool ) -> None: assert operator(left, right) is expected diff --git a/tmt/guest/__init__.py b/tmt/guest/__init__.py index fa636c4772..a85111e87a 100644 --- a/tmt/guest/__init__.py +++ b/tmt/guest/__init__.py @@ -33,6 +33,7 @@ import tmt import tmt.hardware +import tmt.hardware.constraints import tmt.log import tmt.package_managers import tmt.steps @@ -1108,7 +1109,7 @@ def _flag(field: str, label: str) -> tuple[str, str, str]: def normalize_hardware( key_address: str, - raw_hardware: Union[None, tmt.hardware.Spec, tmt.hardware.Hardware], + raw_hardware: Union[None, tmt.hardware.constraints.Spec, tmt.hardware.Hardware], logger: tmt.log.Logger, ) -> Optional[tmt.hardware.Hardware]: """ @@ -1130,10 +1131,10 @@ def normalize_hardware( merged: dict[str, Any] = {} for raw_datum in raw_hardware: - components = tmt.hardware.ConstraintComponents.from_spec(raw_datum) + components = tmt.hardware.constraints.ConstraintComponents.from_spec(raw_datum) if ( - components.name not in tmt.hardware.CHILDLESS_CONSTRAINTS + components.name not in tmt.hardware.constraints.CHILDLESS_CONSTRAINTS and components.child_name is None ): raise tmt.utils.SpecificationError( @@ -1142,7 +1143,7 @@ def normalize_hardware( ) if ( - components.name in tmt.hardware.INDEXABLE_CONSTRAINTS + components.name in tmt.hardware.constraints.INDEXABLE_CONSTRAINTS and components.peer_index is None ): raise tmt.utils.SpecificationError( diff --git a/tmt/hardware/__init__.py b/tmt/hardware/__init__.py new file mode 100644 index 0000000000..dec7380eec --- /dev/null +++ b/tmt/hardware/__init__.py @@ -0,0 +1,51 @@ +""" +Guest hardware requirements specification and helpers. + +tmt metadata allow to describe various HW requirements a guest needs to satisfy. +This package provides useful functions and classes for core functionality and +shared across provision plugins. + +Parsing of HW requirements +========================== + +Set of HW requirements, as given by test or plan metadata, is represented by +Python structures - lists, mappings, primitive types - when loaded from fmf +files. Part of the code below converts this representation to a tree of objects +that provide helpful operations for easier evaluation and processing of HW +requirements. + +Each HW requirement "rule" in original metadata is a constraint, a condition +the eventual guest HW must satisfy. Each node of the tree created from HW +requirements is therefore called "a constraint", and represents either a single +condition ("trivial" constraints), or a set of such conditions plus a function +reducing their individual outcomes to one final answer for the whole set (think +:py:func:`any` and :py:func:`all` built-in functions) ("compound" constraints). +Components of each constraint - dimension, operator, value, units - are +decoupled from the rest, and made available for inspection. + +[1] https://tmt.readthedocs.io/en/stable/spec/hardware.html +""" + +from tmt.hardware.constraints import ( + UNITS, + Constraint, + FlagConstraint, + IntegerConstraint, + NumberConstraint, + Operator, + SizeConstraint, + TextConstraint, +) +from tmt.hardware.requirements import Hardware + +__all__ = [ + 'UNITS', + 'Constraint', + 'FlagConstraint', + 'Hardware', + 'IntegerConstraint', + 'NumberConstraint', + 'Operator', + 'SizeConstraint', + 'TextConstraint', +] diff --git a/tmt/hardware.py b/tmt/hardware/constraints.py similarity index 54% rename from tmt/hardware.py rename to tmt/hardware/constraints.py index 1379f995c7..3882d9d32c 100644 --- a/tmt/hardware.py +++ b/tmt/hardware/constraints.py @@ -1,34 +1,5 @@ -""" -Guest hardware requirements specification and helpers. - -tmt metadata allow to describe various HW requirements a guest needs to satisfy. -This package provides useful functions and classes for core functionality and -shared across provision plugins. - -Parsing of HW requirements -========================== - -Set of HW requirements, as given by test or plan metadata, is represented by -Python structures - lists, mappings, primitive types - when loaded from fmf -files. Part of the code below converts this representation to a tree of objects -that provide helpful operations for easier evaluation and processing of HW -requirements. - -Each HW requirement "rule" in original metadata is a constraint, a condition -the eventual guest HW must satisfy. Each node of the tree created from HW -requirements is therefore called "a constraint", and represents either a single -condition ("trivial" constraints), or a set of such conditions plus a function -reducing their individual outcomes to one final answer for the whole set (think -:py:func:`any` and :py:func:`all` built-in functions) ("compound" constraints). -Components of each constraint - dimension, operator, value, units - are -decoupled from the rest, and made available for inspection. - -[1] https://tmt.readthedocs.io/en/stable/spec/hardware.html -""" - import abc import enum -import functools import itertools import operator import re @@ -64,7 +35,8 @@ UNITS = pint.UnitRegistry() # The default formatting should use unit symbols rather than full names. -UNITS.default_format = '~' +# reportDeprecated: in some Pint versions, this method is deprecated. +UNITS.default_format = '~' # type: ignore[reportDeprecated,unused-ignore] class Operator(enum.Enum): @@ -370,7 +342,9 @@ class BaseConstraint(SpecBasedContainer[Spec, Spec]): @classmethod def from_spec(cls, spec: Any) -> 'BaseConstraint': - return parse_hw_requirements(spec) + import tmt.hardware.requirements + + return tmt.hardware.requirements.parse_hw_requirements(spec) @abc.abstractmethod def to_spec(self) -> Spec: @@ -518,7 +492,7 @@ class Constraint(BaseConstraint): operator_handler: OperatorHandlerType # Constraint value. - value: ConstraintValue # Subclasses will specialize further + value: Any # Subclasses will specialize further # Stored for possible inspection by more advanced processing. raw_value: str @@ -964,893 +938,3 @@ def variants( for constraint in self.constraints: for variant in constraint.variants(): yield members + variant - - -# -# Constraint parsing -# - - -def ungroupify(fn: Callable[[Spec], BaseConstraint]) -> Callable[[Spec], BaseConstraint]: - """ - Swap returned single-child compound constraint and that child. - - Helps reduce the number of levels in the constraint tree: if the return value - is a compound constraint which contains just a single child, return the - child instead of the compound constraint. - - Meant for constraints that do not have an index, e.g. ``memory`` or ``cpu``. - For indexable constraints, see :py:func:`ungroupify_indexed`. - """ - - @functools.wraps(fn) - def wrapper(spec: Spec) -> BaseConstraint: - constraint = fn(spec) - - if isinstance(constraint, CompoundConstraint) and len(constraint.constraints) == 1: - return constraint.constraints[0] - - return constraint - - return wrapper - - -def ungroupify_indexed( - fn: Callable[[Spec, int], BaseConstraint], -) -> Callable[[Spec, int], BaseConstraint]: - """ - Swap returned single-child compound constraint and that child. - - Helps reduce the number of levels in the constraint tree: if the return value - is a compound constraint which contains just a single child, return the - child instead of the compound constraint. - - Meant for constraints that have an index, e.g. ``disk`` or ``network``. For - non-indexable constraints, see :py:func:`ungroupify`. - """ - - @functools.wraps(fn) - def wrapper(spec: Spec, index: int) -> BaseConstraint: - constraint = fn(spec, index) - - if isinstance(constraint, CompoundConstraint) and len(constraint.constraints) == 1: - return constraint.constraints[0] - - return constraint - - return wrapper - - -def _parse_int_constraints( - spec: Spec, - prefix: str, - constraint_keys: tuple[str, ...], -) -> list[BaseConstraint]: - """ - Parse number-like constraints defined by a given set of keys, to int - """ - - return [ - IntegerConstraint.from_specification( - f'{prefix}.{constraint_name.replace("-", "_")}', - str(spec[constraint_name]), - allowed_operators=[ - Operator.EQ, - Operator.NEQ, - Operator.LT, - Operator.LTE, - Operator.GT, - Operator.GTE, - ], - ) - for constraint_name in constraint_keys - if constraint_name in spec - ] - - -def _parse_number_constraints( - spec: Spec, - prefix: str, - constraint_keys: tuple[str, ...], - default_unit: Optional[Any] = None, -) -> list[BaseConstraint]: - """ - Parse number-like constraints defined by a given set of keys, to float - """ - - return [ - NumberConstraint.from_specification( - f'{prefix}.{constraint_name.replace("-", "_")}', - str(spec[constraint_name]), - allowed_operators=[ - Operator.EQ, - Operator.NEQ, - Operator.LT, - Operator.LTE, - Operator.GT, - Operator.GTE, - ], - default_unit=default_unit, - ) - for constraint_name in constraint_keys - if constraint_name in spec - ] - - -def _parse_size_constraints( - spec: Spec, - prefix: str, - constraint_keys: tuple[str, ...], -) -> list[BaseConstraint]: - """ - Parse size-like constraints defined by a given set of keys - """ - - return [ - SizeConstraint.from_specification( - f'{prefix}.{constraint_name.replace("-", "_")}', - str(spec[constraint_name]), - allowed_operators=[ - Operator.EQ, - Operator.NEQ, - Operator.LT, - Operator.LTE, - Operator.GT, - Operator.GTE, - ], - ) - for constraint_name in constraint_keys - if constraint_name in spec - ] - - -def _parse_text_constraints( - spec: Spec, - prefix: str, - constraint_keys: tuple[str, ...], - allowed_operators: Optional[tuple[Operator, ...]] = None, -) -> list[BaseConstraint]: - """ - Parse text-like constraints defined by a given set of keys - """ - - allowed_operators = allowed_operators or ( - Operator.EQ, - Operator.NEQ, - Operator.MATCH, - Operator.NOTMATCH, - ) - - return [ - TextConstraint.from_specification( - f'{prefix}.{constraint_name.replace("-", "_")}', - str(spec[constraint_name]), - allowed_operators=list(allowed_operators), - ) - for constraint_name in constraint_keys - if constraint_name in spec - ] - - -def _parse_flag_constraints( - spec: Spec, - prefix: str, - constraint_keys: tuple[str, ...], -) -> list[BaseConstraint]: - """ - Parse flag-like constraints defined by a given set of keys - """ - - return [ - FlagConstraint.from_specification( - f'{prefix}.{constraint_name.replace("-", "_")}', - spec[constraint_name], - allowed_operators=[Operator.EQ, Operator.NEQ], - ) - for constraint_name in constraint_keys - if constraint_name in spec - ] - - -def _parse_device_core( - spec: Spec, - device_prefix: str = 'device', - include_driver: bool = True, - include_device: bool = True, -) -> And: - """ - Parse constraints shared across device classes. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - number_constraints: tuple[str, ...] = ('vendor',) - text_constraints: tuple[str, ...] = ('vendor-name',) - - if include_device: - number_constraints = (*number_constraints, 'device') - text_constraints = (*text_constraints, 'device-name') - - if include_driver: - text_constraints = (*text_constraints, 'driver') - - group.constraints += _parse_int_constraints( - spec, - device_prefix, - number_constraints, - ) - group.constraints += _parse_text_constraints( - spec, - device_prefix, - text_constraints, - ) - - return group - - -@ungroupify -def _parse_boot(spec: Spec) -> BaseConstraint: - """ - Parse a boot-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - if 'method' in spec: - constraint = TextConstraint.from_specification( - 'boot.method', spec["method"], allowed_operators=[Operator.EQ, Operator.NEQ] - ) - - if constraint.operator == Operator.EQ: - constraint.change_operator(Operator.CONTAINS) - - elif constraint.operator == Operator.NEQ: - constraint.change_operator(Operator.NOTCONTAINS_EXCLUSIVE) - - group.constraints += [constraint] - - return group - - -@ungroupify -def _parse_virtualization(spec: Spec) -> BaseConstraint: - """ - Parse a virtualization-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_flag_constraints( - spec, - 'virtualization', - ('is-virtualized', 'is-supported', 'confidential'), - ) - group.constraints += _parse_text_constraints( - spec, - 'virtualization', - ('hypervisor',), - ) - - return group - - -@ungroupify -def _parse_compatible(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the compatible distro parameter. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - for distro in spec.get('distro', []): - constraint = TextConstraint.from_specification('compatible.distro', distro) - - constraint.change_operator(Operator.CONTAINS) - - group.constraints += [constraint] - - return group - - -@ungroupify -def _parse_cpu(spec: Spec) -> BaseConstraint: - """ - Parse a cpu-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_int_constraints( - spec, - 'cpu', - ( - 'processors', - 'sockets', - 'cores', - 'threads', - 'cores-per-socket', - 'threads-per-core', - 'model', - 'family', - 'vendor', - 'stepping', - ), - ) - - group.constraints += _parse_number_constraints( - spec, - 'cpu', - ('frequency',), - default_unit='MHz', - ) - - group.constraints += _parse_text_constraints( - spec, - 'cpu', - ( - 'family-name', - 'model-name', - 'vendor-name', - ), - ) - - if 'flag' in spec: - flag_group = And() - - for flag_spec in spec['flag']: - constraint = TextConstraint.from_specification('cpu.flag', flag_spec) - - if constraint.operator == Operator.EQ: - constraint.change_operator(Operator.CONTAINS) - - elif constraint.operator == Operator.NEQ: - constraint.change_operator(Operator.NOTCONTAINS) - - flag_group.constraints += [constraint] - - group.constraints += [flag_group] - - if 'hyper-threading' in spec: - group.constraints += [ - FlagConstraint.from_specification( - 'cpu.hyper_threading', - spec['hyper-threading'], - allowed_operators=[Operator.EQ, Operator.NEQ], - ) - ] - - return group - - -@ungroupify -def _parse_device(spec: Spec) -> BaseConstraint: - """ - Parse a device-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - return _parse_device_core(spec) - - -@ungroupify_indexed -def _parse_disk(spec: Spec, disk_index: int) -> BaseConstraint: - """ - Parse a disk-related constraints. - - :param spec: raw constraint block specification. - :param disk_index: index of this disk among its peers in specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_size_constraints( - spec, - f'disk[{disk_index}]', - ('size', 'physical-sector-size', 'logical-sector-size'), - ) - group.constraints += _parse_text_constraints( - spec, - f'disk[{disk_index}]', - ('model-name', 'driver'), - ) - - return group - - -@ungroupify -def _parse_disks(spec: Spec) -> BaseConstraint: - """ - Parse a storage-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - # The old-style constraint when `disk` was a mapping. Remove once v0.0.26 is gone. - if isinstance(spec, dict): - return _parse_disk(spec, 0) - - group = And() - - group.constraints += [ - _parse_disk(disk_spec, disk_index) for disk_index, disk_spec in enumerate(spec) - ] - - return group - - -@ungroupify -def _parse_gpu(spec: Spec) -> BaseConstraint: - """ - Parse a gpu-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - return _parse_device_core(spec, device_prefix='gpu') - - -@ungroupify_indexed -def _parse_network(spec: Spec, network_index: int) -> BaseConstraint: - """ - Parse a network-related constraints. - - :param spec: raw constraint block specification. - :param network_index: index of this network among its peers in specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = _parse_device_core(spec, f'network[{network_index}]') - group.constraints += _parse_text_constraints( - spec, - f'network[{network_index}]', - ('type',), - ) - - return group - - -@ungroupify -def _parse_networks(spec: Spec) -> BaseConstraint: - """ - Parse a network-related constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += [ - _parse_network(network_spec, network_index) - for network_index, network_spec in enumerate(spec) - ] - - return group - - -@ungroupify -def _parse_system(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``system`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = _parse_device_core( - spec, device_prefix='system', include_driver=False, include_device=False - ) - - group.constraints += _parse_int_constraints( - spec, - 'system', - ('model', 'numa-nodes'), - ) - group.constraints += _parse_text_constraints( - spec, - 'system', - ('model-name', 'type'), - ) - - return group - - -TPM_VERSION_ALLOWED_OPERATORS: tuple[Operator, ...] = ( - Operator.EQ, - Operator.NEQ, - Operator.LT, - Operator.LTE, - Operator.GT, - Operator.GTE, -) - - -@ungroupify -def _parse_tpm(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``tpm`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_text_constraints( - spec, - 'tpm', - ('version',), - allowed_operators=TPM_VERSION_ALLOWED_OPERATORS, - ) - - return group - - -def _parse_memory(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``memory`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - return SizeConstraint.from_specification( - 'memory', - str(spec['memory']), - allowed_operators=[ - Operator.EQ, - Operator.NEQ, - Operator.LT, - Operator.LTE, - Operator.GT, - Operator.GTE, - ], - default_unit='MiB', - ) - - -def _parse_hostname(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``hostname`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - return TextConstraint.from_specification( - 'hostname', - spec['hostname'], - allowed_operators=[Operator.EQ, Operator.NEQ, Operator.MATCH, Operator.NOTMATCH], - ) - - -@ungroupify -def _parse_zcrypt(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``zcrypt`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_text_constraints( - spec, - 'zcrypt', - ('adapter', 'mode'), - ) - - return group - - -@ungroupify -def _parse_iommu(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``iommu`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_flag_constraints( - spec, - 'iommu', - ('is-supported',), - ) - group.constraints += _parse_text_constraints( - spec, - 'iommu', - ('model-name',), - ) - - return group - - -@ungroupify -def _parse_location(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``location`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_text_constraints( - spec, - 'location', - ('lab-controller',), - ) - - return group - - -@ungroupify -def _parse_beaker(spec: Spec) -> BaseConstraint: - """ - Parse constraints related to the ``beaker`` HW requirement. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += _parse_text_constraints( - spec, - 'beaker', - ('pool',), - allowed_operators=(Operator.EQ, Operator.NEQ), - ) - - group.constraints += _parse_flag_constraints( - spec, - 'beaker', - ('panic-watchdog',), - ) - - return group - - -@ungroupify -def _parse_generic_spec(spec: Spec) -> BaseConstraint: - """ - Parse actual constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - if 'arch' in spec: - group.constraints += [TextConstraint.from_specification('arch', spec['arch'])] - - if 'beaker' in spec: - group.constraints += [_parse_beaker(spec['beaker'])] - - if 'boot' in spec: - group.constraints += [_parse_boot(spec['boot'])] - - if 'compatible' in spec: - group.constraints += [_parse_compatible(spec['compatible'])] - - if 'cpu' in spec: - group.constraints += [_parse_cpu(spec['cpu'])] - - if 'device' in spec: - group.constraints += [_parse_device(spec['device'])] - - if 'gpu' in spec: - group.constraints += [_parse_gpu(spec['gpu'])] - - if 'memory' in spec: - group.constraints += [_parse_memory(spec)] - - if 'disk' in spec: - group.constraints += [_parse_disks(spec['disk'])] - - if 'network' in spec: - group.constraints += [_parse_networks(spec['network'])] - - if 'hostname' in spec: - group.constraints += [_parse_hostname(spec)] - - if 'location' in spec: - group.constraints += [_parse_location(spec['location'])] - - if 'system' in spec: - group.constraints += [_parse_system(spec['system'])] - - if 'tpm' in spec: - group.constraints += [_parse_tpm(spec['tpm'])] - - if 'virtualization' in spec: - group.constraints += [_parse_virtualization(spec['virtualization'])] - - if 'zcrypt' in spec: - group.constraints += [_parse_zcrypt(spec['zcrypt'])] - - if 'iommu' in spec: - group.constraints += [_parse_iommu(spec['iommu'])] - - return group - - -@ungroupify -def _parse_and(spec: Spec) -> BaseConstraint: - """ - Parse an ``and`` clause holding one or more subblocks or constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = And() - - group.constraints += [_parse_block(member) for member in spec] - - return group - - -@ungroupify -def _parse_or(spec: Spec) -> BaseConstraint: - """ - Parse an ``or`` clause holding one or more subblocks or constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. - """ - - group = Or() - - group.constraints += [_parse_block(member) for member in spec] - - return group - - -@ungroupify -def _parse_block(spec: Spec) -> BaseConstraint: - """ - Parse a generic block of HW constraints - may contain ``and`` and ``or`` - subblocks and actual constraints. - - :param spec: raw constraint block specification. - :returns: block representation as :py:class:`BaseConstraint` or one of its - subclasses. - """ - - if 'and' in spec: - return _parse_and(spec['and']) - - if 'or' in spec: - return _parse_or(spec['or']) - - return _parse_generic_spec(spec) - - -def parse_hw_requirements(spec: Spec) -> BaseConstraint: - """ - Convert raw specification of HW constraints to our internal representation. - - :param spec: raw constraints specification as stored in an environment. - :returns: root of HW constraints tree. - """ - - return _parse_block(spec) - - -@container -class Hardware(SpecBasedContainer[Spec, Spec]): - constraint: Optional[BaseConstraint] - spec: Spec - - @classmethod - def from_spec(cls: type['Hardware'], spec: Spec) -> 'Hardware': - if not spec: - return Hardware(constraint=None, spec=spec) - - return Hardware(constraint=parse_hw_requirements(spec), spec=spec) - - def to_spec(self) -> Spec: - return self.spec - - def to_minimal_spec(self) -> Spec: - return self.spec - - def and_(self, constraint: BaseConstraint) -> None: - if self.constraint: - group = And() - - group.constraints = [self.constraint, constraint] - - self.constraint = group - - else: - self.constraint = constraint - - self.spec = self.constraint.to_spec() - - def report_support( - self, - *, - names: Optional[list[str]] = None, - check: Optional[Callable[['Constraint'], bool]] = None, - logger: tmt.log.Logger, - ) -> None: - """ - Report all unsupported constraints. - - A helper method for plugins: plugin provides a callback which checks - whether a given constraint is or is not supported by the plugin, and - method calls the callback for each constraint stored in this container. - - Both ``names`` and ``check`` are optional, and both can be used and - combined. First, the ``names`` list is checked, if a constraint is not - found, ``check`` is called if it's defined. - - :param names: a list of constraint names. If a constraint name is on - this list, it is considered to be supported by the ``report_support`` - caller. Caller may list both full constraint name, e.g. ``cpu.cores``, - or just the subsystem name, ``cpu``, indicating all child constraints - are supported. - :param check: a callback to call for each constraint in this container. - Accepts a single parameter, a constraint to check, and if its return - value is true-ish, the constraint is considered to be supported - by the ``report_support`` caller. - """ - - if not self.constraint: - return - - names = names or [] - check = check or (lambda _: False) - - for variant in self.constraint.variants(): - for constraint in variant: - name, _, child_name = constraint.expand_name() - - if name in names or f'{name}.{child_name}' in names or check(constraint): - continue - - logger.warning(f"Hardware requirement '{constraint}' is not supported.") - - def format_variants(self) -> Iterator[str]: - """ - Format variants of constraints. - - :yields: for each variant, which is nothing but a list of constraints, - method yields a string variant's serial number and formatted - constraints. - """ - - if self.constraint is None: - return - - for i, constraints in enumerate(self.constraint.variants(), start=1): - for constraint in constraints: - yield f'variant #{i}: {constraint!s}' diff --git a/tmt/hardware/requirements.py b/tmt/hardware/requirements.py new file mode 100644 index 0000000000..708a308cff --- /dev/null +++ b/tmt/hardware/requirements.py @@ -0,0 +1,905 @@ +import functools +from collections.abc import Iterator +from typing import Any, Callable, Optional + +import tmt.log +from tmt.container import SpecBasedContainer, container +from tmt.hardware.constraints import ( + And, + BaseConstraint, + CompoundConstraint, + Constraint, + FlagConstraint, + IntegerConstraint, + NumberConstraint, + Operator, + Or, + SizeConstraint, + Spec, + TextConstraint, +) + + +def ungroupify(fn: Callable[[Spec], BaseConstraint]) -> Callable[[Spec], BaseConstraint]: + """ + Swap returned single-child compound constraint and that child. + + Helps reduce the number of levels in the constraint tree: if the return value + is a compound constraint which contains just a single child, return the + child instead of the compound constraint. + + Meant for constraints that do not have an index, e.g. ``memory`` or ``cpu``. + For indexable constraints, see :py:func:`ungroupify_indexed`. + """ + + @functools.wraps(fn) + def wrapper(spec: Spec) -> BaseConstraint: + constraint = fn(spec) + + if isinstance(constraint, CompoundConstraint) and len(constraint.constraints) == 1: + return constraint.constraints[0] + + return constraint + + return wrapper + + +def ungroupify_indexed( + fn: Callable[[Spec, int], BaseConstraint], +) -> Callable[[Spec, int], BaseConstraint]: + """ + Swap returned single-child compound constraint and that child. + + Helps reduce the number of levels in the constraint tree: if the return value + is a compound constraint which contains just a single child, return the + child instead of the compound constraint. + + Meant for constraints that have an index, e.g. ``disk`` or ``network``. For + non-indexable constraints, see :py:func:`ungroupify`. + """ + + @functools.wraps(fn) + def wrapper(spec: Spec, index: int) -> BaseConstraint: + constraint = fn(spec, index) + + if isinstance(constraint, CompoundConstraint) and len(constraint.constraints) == 1: + return constraint.constraints[0] + + return constraint + + return wrapper + + +def _parse_int_constraints( + spec: Spec, + prefix: str, + constraint_keys: tuple[str, ...], +) -> list[BaseConstraint]: + """ + Parse number-like constraints defined by a given set of keys, to int + """ + + return [ + IntegerConstraint.from_specification( + f'{prefix}.{constraint_name.replace("-", "_")}', + str(spec[constraint_name]), + allowed_operators=[ + Operator.EQ, + Operator.NEQ, + Operator.LT, + Operator.LTE, + Operator.GT, + Operator.GTE, + ], + ) + for constraint_name in constraint_keys + if constraint_name in spec + ] + + +def _parse_number_constraints( + spec: Spec, + prefix: str, + constraint_keys: tuple[str, ...], + default_unit: Optional[Any] = None, +) -> list[BaseConstraint]: + """ + Parse number-like constraints defined by a given set of keys, to float + """ + + return [ + NumberConstraint.from_specification( + f'{prefix}.{constraint_name.replace("-", "_")}', + str(spec[constraint_name]), + allowed_operators=[ + Operator.EQ, + Operator.NEQ, + Operator.LT, + Operator.LTE, + Operator.GT, + Operator.GTE, + ], + default_unit=default_unit, + ) + for constraint_name in constraint_keys + if constraint_name in spec + ] + + +def _parse_size_constraints( + spec: Spec, + prefix: str, + constraint_keys: tuple[str, ...], +) -> list[BaseConstraint]: + """ + Parse size-like constraints defined by a given set of keys + """ + + return [ + SizeConstraint.from_specification( + f'{prefix}.{constraint_name.replace("-", "_")}', + str(spec[constraint_name]), + allowed_operators=[ + Operator.EQ, + Operator.NEQ, + Operator.LT, + Operator.LTE, + Operator.GT, + Operator.GTE, + ], + ) + for constraint_name in constraint_keys + if constraint_name in spec + ] + + +def _parse_text_constraints( + spec: Spec, + prefix: str, + constraint_keys: tuple[str, ...], + allowed_operators: Optional[tuple[Operator, ...]] = None, +) -> list[BaseConstraint]: + """ + Parse text-like constraints defined by a given set of keys + """ + + allowed_operators = allowed_operators or ( + Operator.EQ, + Operator.NEQ, + Operator.MATCH, + Operator.NOTMATCH, + ) + + return [ + TextConstraint.from_specification( + f'{prefix}.{constraint_name.replace("-", "_")}', + str(spec[constraint_name]), + allowed_operators=list(allowed_operators), + ) + for constraint_name in constraint_keys + if constraint_name in spec + ] + + +def _parse_flag_constraints( + spec: Spec, + prefix: str, + constraint_keys: tuple[str, ...], +) -> list[BaseConstraint]: + """ + Parse flag-like constraints defined by a given set of keys + """ + + return [ + FlagConstraint.from_specification( + f'{prefix}.{constraint_name.replace("-", "_")}', + spec[constraint_name], + allowed_operators=[Operator.EQ, Operator.NEQ], + ) + for constraint_name in constraint_keys + if constraint_name in spec + ] + + +def _parse_device_core( + spec: Spec, + device_prefix: str = 'device', + include_driver: bool = True, + include_device: bool = True, +) -> And: + """ + Parse constraints shared across device classes. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + number_constraints: tuple[str, ...] = ('vendor',) + text_constraints: tuple[str, ...] = ('vendor-name',) + + if include_device: + number_constraints = (*number_constraints, 'device') + text_constraints = (*text_constraints, 'device-name') + + if include_driver: + text_constraints = (*text_constraints, 'driver') + + group.constraints += _parse_int_constraints( + spec, + device_prefix, + number_constraints, + ) + group.constraints += _parse_text_constraints( + spec, + device_prefix, + text_constraints, + ) + + return group + + +@ungroupify +def _parse_boot(spec: Spec) -> BaseConstraint: + """ + Parse a boot-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + if 'method' in spec: + constraint = TextConstraint.from_specification( + 'boot.method', spec["method"], allowed_operators=[Operator.EQ, Operator.NEQ] + ) + + if constraint.operator == Operator.EQ: + constraint.change_operator(Operator.CONTAINS) + + elif constraint.operator == Operator.NEQ: + constraint.change_operator(Operator.NOTCONTAINS_EXCLUSIVE) + + group.constraints += [constraint] + + return group + + +@ungroupify +def _parse_virtualization(spec: Spec) -> BaseConstraint: + """ + Parse a virtualization-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_flag_constraints( + spec, + 'virtualization', + ('is-virtualized', 'is-supported', 'confidential'), + ) + group.constraints += _parse_text_constraints( + spec, + 'virtualization', + ('hypervisor',), + ) + + return group + + +@ungroupify +def _parse_compatible(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the compatible distro parameter. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + for distro in spec.get('distro', []): + constraint = TextConstraint.from_specification('compatible.distro', distro) + + constraint.change_operator(Operator.CONTAINS) + + group.constraints += [constraint] + + return group + + +@ungroupify +def _parse_cpu(spec: Spec) -> BaseConstraint: + """ + Parse a cpu-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_int_constraints( + spec, + 'cpu', + ( + 'processors', + 'sockets', + 'cores', + 'threads', + 'cores-per-socket', + 'threads-per-core', + 'model', + 'family', + 'vendor', + 'stepping', + ), + ) + + group.constraints += _parse_number_constraints( + spec, + 'cpu', + ('frequency',), + default_unit='MHz', + ) + + group.constraints += _parse_text_constraints( + spec, + 'cpu', + ( + 'family-name', + 'model-name', + 'vendor-name', + ), + ) + + if 'flag' in spec: + flag_group = And() + + for flag_spec in spec['flag']: + constraint = TextConstraint.from_specification('cpu.flag', flag_spec) + + if constraint.operator == Operator.EQ: + constraint.change_operator(Operator.CONTAINS) + + elif constraint.operator == Operator.NEQ: + constraint.change_operator(Operator.NOTCONTAINS) + + flag_group.constraints += [constraint] + + group.constraints += [flag_group] + + if 'hyper-threading' in spec: + group.constraints += [ + FlagConstraint.from_specification( + 'cpu.hyper_threading', + spec['hyper-threading'], + allowed_operators=[Operator.EQ, Operator.NEQ], + ) + ] + + return group + + +@ungroupify +def _parse_device(spec: Spec) -> BaseConstraint: + """ + Parse a device-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + return _parse_device_core(spec) + + +@ungroupify_indexed +def _parse_disk(spec: Spec, disk_index: int) -> BaseConstraint: + """ + Parse a disk-related constraints. + + :param spec: raw constraint block specification. + :param disk_index: index of this disk among its peers in specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_size_constraints( + spec, + f'disk[{disk_index}]', + ('size', 'physical-sector-size', 'logical-sector-size'), + ) + group.constraints += _parse_text_constraints( + spec, + f'disk[{disk_index}]', + ('model-name', 'driver'), + ) + + return group + + +@ungroupify +def _parse_disks(spec: Spec) -> BaseConstraint: + """ + Parse a storage-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + # The old-style constraint when `disk` was a mapping. Remove once v0.0.26 is gone. + if isinstance(spec, dict): + return _parse_disk(spec, 0) + + group = And() + + group.constraints += [ + _parse_disk(disk_spec, disk_index) for disk_index, disk_spec in enumerate(spec) + ] + + return group + + +@ungroupify +def _parse_gpu(spec: Spec) -> BaseConstraint: + """ + Parse a gpu-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + return _parse_device_core(spec, device_prefix='gpu') + + +@ungroupify_indexed +def _parse_network(spec: Spec, network_index: int) -> BaseConstraint: + """ + Parse a network-related constraints. + + :param spec: raw constraint block specification. + :param network_index: index of this network among its peers in specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = _parse_device_core(spec, f'network[{network_index}]') + group.constraints += _parse_text_constraints( + spec, + f'network[{network_index}]', + ('type',), + ) + + return group + + +@ungroupify +def _parse_networks(spec: Spec) -> BaseConstraint: + """ + Parse a network-related constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += [ + _parse_network(network_spec, network_index) + for network_index, network_spec in enumerate(spec) + ] + + return group + + +@ungroupify +def _parse_system(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``system`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = _parse_device_core( + spec, device_prefix='system', include_driver=False, include_device=False + ) + + group.constraints += _parse_int_constraints( + spec, + 'system', + ('model', 'numa-nodes'), + ) + group.constraints += _parse_text_constraints( + spec, + 'system', + ('model-name', 'type'), + ) + + return group + + +TPM_VERSION_ALLOWED_OPERATORS: tuple[Operator, ...] = ( + Operator.EQ, + Operator.NEQ, + Operator.LT, + Operator.LTE, + Operator.GT, + Operator.GTE, +) + + +@ungroupify +def _parse_tpm(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``tpm`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_text_constraints( + spec, + 'tpm', + ('version',), + allowed_operators=TPM_VERSION_ALLOWED_OPERATORS, + ) + + return group + + +def _parse_memory(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``memory`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + return SizeConstraint.from_specification( + 'memory', + str(spec['memory']), + allowed_operators=[ + Operator.EQ, + Operator.NEQ, + Operator.LT, + Operator.LTE, + Operator.GT, + Operator.GTE, + ], + default_unit='MiB', + ) + + +def _parse_hostname(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``hostname`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + return TextConstraint.from_specification( + 'hostname', + spec['hostname'], + allowed_operators=[Operator.EQ, Operator.NEQ, Operator.MATCH, Operator.NOTMATCH], + ) + + +@ungroupify +def _parse_zcrypt(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``zcrypt`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_text_constraints( + spec, + 'zcrypt', + ('adapter', 'mode'), + ) + + return group + + +@ungroupify +def _parse_iommu(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``iommu`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_flag_constraints( + spec, + 'iommu', + ('is-supported',), + ) + group.constraints += _parse_text_constraints( + spec, + 'iommu', + ('model-name',), + ) + + return group + + +@ungroupify +def _parse_location(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``location`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_text_constraints( + spec, + 'location', + ('lab-controller',), + ) + + return group + + +@ungroupify +def _parse_beaker(spec: Spec) -> BaseConstraint: + """ + Parse constraints related to the ``beaker`` HW requirement. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += _parse_text_constraints( + spec, + 'beaker', + ('pool',), + allowed_operators=(Operator.EQ, Operator.NEQ), + ) + + group.constraints += _parse_flag_constraints( + spec, + 'beaker', + ('panic-watchdog',), + ) + + return group + + +@ungroupify +def _parse_generic_spec(spec: Spec) -> BaseConstraint: + """ + Parse actual constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + if 'arch' in spec: + group.constraints += [TextConstraint.from_specification('arch', spec['arch'])] + + if 'beaker' in spec: + group.constraints += [_parse_beaker(spec['beaker'])] + + if 'boot' in spec: + group.constraints += [_parse_boot(spec['boot'])] + + if 'compatible' in spec: + group.constraints += [_parse_compatible(spec['compatible'])] + + if 'cpu' in spec: + group.constraints += [_parse_cpu(spec['cpu'])] + + if 'device' in spec: + group.constraints += [_parse_device(spec['device'])] + + if 'gpu' in spec: + group.constraints += [_parse_gpu(spec['gpu'])] + + if 'memory' in spec: + group.constraints += [_parse_memory(spec)] + + if 'disk' in spec: + group.constraints += [_parse_disks(spec['disk'])] + + if 'network' in spec: + group.constraints += [_parse_networks(spec['network'])] + + if 'hostname' in spec: + group.constraints += [_parse_hostname(spec)] + + if 'location' in spec: + group.constraints += [_parse_location(spec['location'])] + + if 'system' in spec: + group.constraints += [_parse_system(spec['system'])] + + if 'tpm' in spec: + group.constraints += [_parse_tpm(spec['tpm'])] + + if 'virtualization' in spec: + group.constraints += [_parse_virtualization(spec['virtualization'])] + + if 'zcrypt' in spec: + group.constraints += [_parse_zcrypt(spec['zcrypt'])] + + if 'iommu' in spec: + group.constraints += [_parse_iommu(spec['iommu'])] + + return group + + +@ungroupify +def _parse_and(spec: Spec) -> BaseConstraint: + """ + Parse an ``and`` clause holding one or more subblocks or constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = And() + + group.constraints += [_parse_block(member) for member in spec] + + return group + + +@ungroupify +def _parse_or(spec: Spec) -> BaseConstraint: + """ + Parse an ``or`` clause holding one or more subblocks or constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its subclasses. + """ + + group = Or() + + group.constraints += [_parse_block(member) for member in spec] + + return group + + +@ungroupify +def _parse_block(spec: Spec) -> BaseConstraint: + """ + Parse a generic block of HW constraints - may contain ``and`` and ``or`` + subblocks and actual constraints. + + :param spec: raw constraint block specification. + :returns: block representation as :py:class:`BaseConstraint` or one of its + subclasses. + """ + + if 'and' in spec: + return _parse_and(spec['and']) + + if 'or' in spec: + return _parse_or(spec['or']) + + return _parse_generic_spec(spec) + + +def parse_hw_requirements(spec: Spec) -> BaseConstraint: + """ + Convert raw specification of HW constraints to our internal representation. + + :param spec: raw constraints specification as stored in an environment. + :returns: root of HW constraints tree. + """ + + return _parse_block(spec) + + +@container +class Hardware(SpecBasedContainer[Spec, Spec]): + constraint: Optional[BaseConstraint] + spec: Spec + + @classmethod + def from_spec(cls: type['Hardware'], spec: Spec) -> 'Hardware': + if not spec: + return Hardware(constraint=None, spec=spec) + + return Hardware(constraint=parse_hw_requirements(spec), spec=spec) + + def to_spec(self) -> Spec: + return self.spec + + def to_minimal_spec(self) -> Spec: + return self.spec + + def and_(self, constraint: BaseConstraint) -> None: + if self.constraint: + group = And() + + group.constraints = [self.constraint, constraint] + + self.constraint = group + + else: + self.constraint = constraint + + self.spec = self.constraint.to_spec() + + def report_support( + self, + *, + names: Optional[list[str]] = None, + check: Optional[Callable[['Constraint'], bool]] = None, + logger: tmt.log.Logger, + ) -> None: + """ + Report all unsupported constraints. + + A helper method for plugins: plugin provides a callback which checks + whether a given constraint is or is not supported by the plugin, and + method calls the callback for each constraint stored in this container. + + Both ``names`` and ``check`` are optional, and both can be used and + combined. First, the ``names`` list is checked, if a constraint is not + found, ``check`` is called if it's defined. + + :param names: a list of constraint names. If a constraint name is on + this list, it is considered to be supported by the ``report_support`` + caller. Caller may list both full constraint name, e.g. ``cpu.cores``, + or just the subsystem name, ``cpu``, indicating all child constraints + are supported. + :param check: a callback to call for each constraint in this container. + Accepts a single parameter, a constraint to check, and if its return + value is true-ish, the constraint is considered to be supported + by the ``report_support`` caller. + """ + + if not self.constraint: + return + + names = names or [] + check = check or (lambda _: False) + + for variant in self.constraint.variants(): + for constraint in variant: + name, _, child_name = constraint.expand_name() + + if name in names or f'{name}.{child_name}' in names or check(constraint): + continue + + logger.warning(f"Hardware requirement '{constraint}' is not supported.") + + def format_variants(self) -> Iterator[str]: + """ + Format variants of constraints. + + :yields: for each variant, which is nothing but a list of constraints, + method yields a string variant's serial number and formatted + constraints. + """ + + if self.constraint is None: + return + + for i, constraints in enumerate(self.constraint.variants(), start=1): + for constraint in constraints: + yield f'variant #{i}: {constraint!s}' diff --git a/tmt/steps/provision/bootc.py b/tmt/steps/provision/bootc.py index 300aedb6f8..229351b77f 100644 --- a/tmt/steps/provision/bootc.py +++ b/tmt/steps/provision/bootc.py @@ -17,7 +17,7 @@ from tmt.utils.templates import render_template if TYPE_CHECKING: - from tmt.hardware import Size + from tmt.hardware.constraints import Size DEFAULT_TMP_PATH = "/var/tmp/tmt" # noqa: S108 diff --git a/tmt/steps/provision/mrack.py b/tmt/steps/provision/mrack.py index 32841ea755..9a988e707f 100644 --- a/tmt/steps/provision/mrack.py +++ b/tmt/steps/provision/mrack.py @@ -16,6 +16,7 @@ import tmt.config import tmt.guest import tmt.hardware +import tmt.hardware.constraints import tmt.log import tmt.steps import tmt.steps.provision @@ -899,13 +900,13 @@ def _transform_system_vendor_name( def constraint_to_beaker_filter( - constraint: tmt.hardware.BaseConstraint, logger: tmt.log.Logger + constraint: tmt.hardware.constraints.BaseConstraint, logger: tmt.log.Logger ) -> BeakerizedConstraint: """ Convert a hardware constraint into a Mrack-compatible filter """ - if isinstance(constraint, tmt.hardware.And): + if isinstance(constraint, tmt.hardware.constraints.And): return MrackHWAndGroup( children=[ constraint_to_beaker_filter(child_constraint, logger) @@ -913,7 +914,7 @@ def constraint_to_beaker_filter( ] ) - if isinstance(constraint, tmt.hardware.Or): + if isinstance(constraint, tmt.hardware.constraints.Or): return MrackHWOrGroup( children=[ constraint_to_beaker_filter(child_constraint, logger) @@ -1011,14 +1012,16 @@ def _translate_tmt_hw(self, hw: tmt.hardware.Hardware) -> dict[str, Any]: return {'hostRequires': transformed} - def _requires_panic_watchdog(self, constraint: tmt.hardware.BaseConstraint) -> bool: + def _requires_panic_watchdog( + self, constraint: tmt.hardware.constraints.BaseConstraint + ) -> bool: """ Check if any of the constraints are beaker panic-watchdog with the value True """ if isinstance(constraint, tmt.hardware.FlagConstraint): return constraint.name == 'beaker.panic_watchdog' and constraint.value - if isinstance(constraint, (tmt.hardware.And, tmt.hardware.Or)): + if isinstance(constraint, (tmt.hardware.constraints.And, tmt.hardware.constraints.Or)): return any( self._requires_panic_watchdog(child) for child in constraint.constraints ) diff --git a/tmt/steps/provision/testcloud.py b/tmt/steps/provision/testcloud.py index 4acc8ae033..d14051c631 100644 --- a/tmt/steps/provision/testcloud.py +++ b/tmt/steps/provision/testcloud.py @@ -37,7 +37,7 @@ if TYPE_CHECKING: import tmt.base.core - from tmt.hardware import Size + from tmt.hardware.constraints import Size libvirt: Optional[types.ModuleType] = None diff --git a/tmt/steps/report/reportportal.py b/tmt/steps/report/reportportal.py index 07a4874333..9e39fad680 100644 --- a/tmt/steps/report/reportportal.py +++ b/tmt/steps/report/reportportal.py @@ -28,7 +28,7 @@ if TYPE_CHECKING: from tmt._compat.typing import TypeAlias - from tmt.hardware import Size + from tmt.hardware.constraints import Size JSON: 'TypeAlias' = Any DEFAULT_LOG_SIZE_LIMIT: 'Size' = tmt.hardware.UNITS('1 MB') diff --git a/tmt/utils/__init__.py b/tmt/utils/__init__.py index c1f221669c..7211e21285 100644 --- a/tmt/utils/__init__.py +++ b/tmt/utils/__init__.py @@ -83,7 +83,7 @@ import tmt.utils.themes from tmt._compat.typing import ParamSpec, Self, TypeAlias from tmt.guest import GuestLog - from tmt.hardware import Size + from tmt.hardware.constraints import Size def sanitize_string(text: str) -> str: