From eb78a0a2ca4433fcc2cfd056b65d1778d05a3eca Mon Sep 17 00:00:00 2001 From: Cristian Le Date: Wed, 9 Oct 2024 19:18:29 +0200 Subject: [PATCH] Move container stuff Signed-off-by: Cristian Le Signed-off-by: Cristian Le --- docs/scripts/generate-plugins.py | 24 +- tests/unit/test_dataclasses.py | 7 +- tmt/base.py | 31 +- tmt/checks/__init__.py | 13 +- tmt/checks/dmesg.py | 6 +- tmt/checks/watchdog.py | 5 +- tmt/container/__init__.py | 799 +++++++++++++++++++++++++++++++ tmt/hardware.py | 15 +- tmt/result.py | 18 +- tmt/steps/__init__.py | 25 +- tmt/steps/discover/__init__.py | 6 +- tmt/steps/discover/fmf.py | 6 +- tmt/steps/discover/shell.py | 9 +- tmt/steps/execute/__init__.py | 4 +- tmt/steps/execute/internal.py | 5 +- tmt/steps/execute/upgrade.py | 8 +- tmt/steps/finish/__init__.py | 4 +- tmt/steps/finish/shell.py | 6 +- tmt/steps/prepare/__init__.py | 3 +- tmt/steps/prepare/ansible.py | 5 +- tmt/steps/prepare/distgit.py | 6 +- tmt/steps/prepare/feature.py | 6 +- tmt/steps/prepare/install.py | 6 +- tmt/steps/prepare/shell.py | 6 +- tmt/steps/provision/__init__.py | 14 +- tmt/steps/provision/artemis.py | 7 +- tmt/steps/provision/bootc.py | 5 +- tmt/steps/provision/connect.py | 8 +- tmt/steps/provision/local.py | 4 +- tmt/steps/provision/mrack.py | 6 +- tmt/steps/provision/podman.py | 8 +- tmt/steps/provision/testcloud.py | 7 +- tmt/steps/report/__init__.py | 4 +- tmt/steps/report/display.py | 6 +- tmt/steps/report/html.py | 6 +- tmt/steps/report/junit.py | 6 +- tmt/steps/report/polarion.py | 6 +- tmt/steps/report/reportportal.py | 5 +- tmt/utils/__init__.py | 788 +----------------------------- 39 files changed, 963 insertions(+), 940 deletions(-) create mode 100644 tmt/container/__init__.py diff --git a/docs/scripts/generate-plugins.py b/docs/scripts/generate-plugins.py index 9368f7311d..ecdb5fac9d 100755 --- a/docs/scripts/generate-plugins.py +++ b/docs/scripts/generate-plugins.py @@ -7,6 +7,7 @@ from typing import Any import tmt.checks +import tmt.container import tmt.log import tmt.plugins import tmt.steps @@ -17,7 +18,8 @@ import tmt.steps.provision import tmt.steps.report import tmt.utils -from tmt.utils import ContainerClass, Path +from tmt.container import ContainerClass +from tmt.utils import Path from tmt.utils.templates import render_template_file REVIEWED_PLUGINS: tuple[str, ...] = ( @@ -37,7 +39,7 @@ def _is_ignored( container: ContainerClass, field: dataclasses.Field[Any], - metadata: tmt.utils.FieldMetadata) -> bool: + metadata: tmt.container.FieldMetadata) -> bool: """ Check whether a given field is to be ignored in documentation """ if field.name in ('how', '_OPTIONLESS_FIELDS'): @@ -52,7 +54,7 @@ def _is_ignored( def _is_inherited( container: ContainerClass, field: dataclasses.Field[Any], - metadata: tmt.utils.FieldMetadata) -> bool: + metadata: tmt.container.FieldMetadata) -> bool: """ Check whether a given field is inherited from step data base class """ # TODO: for now, it's a list, but inspecting the actual tree of classes @@ -65,8 +67,8 @@ def container_ignored_fields(container: ContainerClass) -> list[str]: field_names: list[str] = [] - for field in tmt.utils.container_fields(container): - _, _, _, _, metadata = tmt.utils.container_field(container, field.name) + for field in tmt.container.container_fields(container): + _, _, _, _, metadata = tmt.container.container_field(container, field.name) if _is_ignored(container, field, metadata): field_names.append(field.name) @@ -79,8 +81,8 @@ def container_inherited_fields(container: ContainerClass) -> list[str]: field_names: list[str] = [] - for field in tmt.utils.container_fields(container): - _, _, _, _, metadata = tmt.utils.container_field(container, field.name) + for field in tmt.container.container_fields(container): + _, _, _, _, metadata = tmt.container.container_field(container, field.name) if _is_inherited(container, field, metadata): field_names.append(field.name) @@ -93,8 +95,8 @@ def container_intrinsic_fields(container: ContainerClass) -> list[str]: field_names: list[str] = [] - for field in tmt.utils.container_fields(container): - _, _, _, _, metadata = tmt.utils.container_field(container, field.name) + for field in tmt.container.container_fields(container): + _, _, _, _, metadata = tmt.container.container_field(container, field.name) if _is_ignored(container, field, metadata): continue @@ -192,8 +194,8 @@ def main() -> None: PLUGINS=plugin_generator, REVIEWED_PLUGINS=REVIEWED_PLUGINS, is_enum=is_enum, - container_fields=tmt.utils.container_fields, - container_field=tmt.utils.container_field, + container_fields=tmt.container.container_fields, + container_field=tmt.container.container_field, container_ignored_fields=container_ignored_fields, container_inherited_fields=container_inherited_fields, container_intrinsic_fields=container_intrinsic_fields)) diff --git a/tests/unit/test_dataclasses.py b/tests/unit/test_dataclasses.py index 694f754f00..035ee3d3bf 100644 --- a/tests/unit/test_dataclasses.py +++ b/tests/unit/test_dataclasses.py @@ -1,14 +1,13 @@ -import dataclasses from typing import Any, Optional import pytest import tmt.log import tmt.utils +from tmt.container import container, field from tmt.utils import ( SerializableContainer, dataclass_normalize_field, - field, ) @@ -28,7 +27,7 @@ def _normalize_foo(key_address: str, raw_value: Any, logger: tmt.log.Logger) -> raise tmt.utils.NormalizationError(key_address, raw_value, 'unset or an integer') \ from exc - @dataclasses.dataclass + @container class DummyContainer(SerializableContainer): foo: Optional[int] = field( default=1, @@ -57,7 +56,7 @@ class DummyContainer(SerializableContainer): def test_field_custom_serialize(): - @dataclasses.dataclass + @container class DummyContainer(SerializableContainer): foo: list[str] = field( default_factory=list, diff --git a/tmt/base.py b/tmt/base.py index 27fe185430..8a5dbb8112 100644 --- a/tmt/base.py +++ b/tmt/base.py @@ -56,6 +56,14 @@ import tmt.utils.git import tmt.utils.jira from tmt.checks import Check +from tmt.container import ( + SerializableContainer, + SpecBasedContainer, + container, + container_field, + container_fields, + field, + ) from tmt.lint import LinterOutcome, LinterReturn from tmt.result import Result, ResultInterpret from tmt.utils import ( @@ -64,14 +72,9 @@ EnvVarValue, FmfContext, Path, - SerializableContainer, ShellScript, - SpecBasedContainer, WorkdirArgumentType, - container_field, - container_fields, dict_to_yaml, - field, normalize_shell_script, verdict, ) @@ -137,7 +140,7 @@ class _RawFmfId(TypedDict, total=False): # An internal fmf id representation. -@dataclasses.dataclass +@container class FmfId( SpecBasedContainer[_RawFmfId, _RawFmfId], SerializableContainer, @@ -411,7 +414,7 @@ class _RawDependencyFmfId(_RawFmfId): type: Optional[str] -@dataclasses.dataclass +@container class DependencyFmfId( FmfId, # Repeat the SpecBasedContainer, with more fitting in/out spec type. @@ -500,7 +503,7 @@ class _RawDependencyFile(TypedDict): pattern: Optional[list[str]] -@dataclasses.dataclass +@container class DependencyFile( SpecBasedContainer[_RawDependencyFile, _RawDependencyFile], SerializableContainer, @@ -649,7 +652,7 @@ def _normalize_link(key_address: str, value: _RawLinks, logger: tmt.log.Logger) return Links(data=value) -@dataclasses.dataclass(repr=False) +@container(repr=False) class Core( tmt.utils.ValidateFmfMixin, tmt.utils.LoadFmfKeysMixin, @@ -1037,7 +1040,7 @@ def has_link(self, needle: 'LinkNeedle') -> bool: Node = Core -@dataclasses.dataclass(repr=False) +@container(repr=False) class Test( Core, tmt.export.Exportable['Test'], @@ -1658,7 +1661,7 @@ def expand_node_data(data: T, fmf_context: FmfContext) -> T: return data -@dataclasses.dataclass(repr=False) +@container(repr=False) class Plan( Core, tmt.export.Exportable['Plan'], @@ -2618,7 +2621,7 @@ def __str__(self) -> str: return self.value -@dataclasses.dataclass(repr=False) +@container(repr=False) class Story( Core, tmt.export.Exportable['Story'], @@ -3358,7 +3361,7 @@ def init( logger=logger) -@dataclasses.dataclass +@container class RunData(SerializableContainer): root: Optional[str] plans: Optional[list[str]] @@ -4111,7 +4114,7 @@ def matches(self, link: 'Link') -> bool: return False -@dataclasses.dataclass +@container class Link(SpecBasedContainer[Any, _RawLinkRelation]): """ An internal "link" as defined by tmt specification. diff --git a/tmt/checks/__init__.py b/tmt/checks/__init__.py index 0fcafb3e38..4f98430611 100644 --- a/tmt/checks/__init__.py +++ b/tmt/checks/__init__.py @@ -1,4 +1,3 @@ -import dataclasses import enum import functools from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypedDict, TypeVar, cast @@ -6,13 +5,15 @@ import tmt.log import tmt.steps.provision import tmt.utils -from tmt.plugins import PluginRegistry -from tmt.utils import ( - NormalizeKeysMixin, +from tmt.container import ( SerializableContainer, SpecBasedContainer, + container, field, + key_to_option, ) +from tmt.plugins import PluginRegistry +from tmt.utils import NormalizeKeysMixin if TYPE_CHECKING: import tmt.base @@ -100,7 +101,7 @@ def to_spec(self) -> str: return self.value -@dataclasses.dataclass +@container class Check( SpecBasedContainer[_RawCheck, _RawCheck], SerializableContainer, @@ -143,7 +144,7 @@ def from_spec( # type: ignore[override] def to_spec(self) -> _RawCheck: spec = cast(_RawCheck, { - tmt.utils.key_to_option(key): value + key_to_option(key): value for key, value in self.items() }) spec["result"] = self.result.to_spec() diff --git a/tmt/checks/dmesg.py b/tmt/checks/dmesg.py index c0bf460a26..289a167e82 100644 --- a/tmt/checks/dmesg.py +++ b/tmt/checks/dmesg.py @@ -1,4 +1,3 @@ -import dataclasses import datetime import re from re import Pattern @@ -9,9 +8,10 @@ import tmt.steps.provision import tmt.utils from tmt.checks import Check, CheckEvent, CheckPlugin, _RawCheck, provides_check +from tmt.container import container, field from tmt.result import CheckResult, ResultOutcome from tmt.steps.provision import GuestCapability -from tmt.utils import Path, field, format_timestamp, render_run_exception_streams +from tmt.utils import Path, format_timestamp, render_run_exception_streams if TYPE_CHECKING: import tmt.base @@ -29,7 +29,7 @@ ] -@dataclasses.dataclass +@container class DmesgCheck(Check): failure_pattern: list[Pattern[str]] = field( default_factory=lambda: DEFAULT_FAILURE_PATTERNS[:], diff --git a/tmt/checks/watchdog.py b/tmt/checks/watchdog.py index a125723994..54e06c666e 100644 --- a/tmt/checks/watchdog.py +++ b/tmt/checks/watchdog.py @@ -17,8 +17,9 @@ import tmt.steps.provision.testcloud import tmt.utils from tmt.checks import Check, CheckPlugin, provides_check +from tmt.container import container, field from tmt.result import CheckResult, ResultOutcome -from tmt.utils import Path, field, format_timestamp, render_run_exception_streams +from tmt.utils import Path, format_timestamp, render_run_exception_streams if TYPE_CHECKING: from tmt.steps.execute import TestInvocation @@ -101,7 +102,7 @@ class GuestContext: keep_running: bool = True -@dataclasses.dataclass +@container class WatchdogCheck(Check): interval: int = field( default=60, diff --git a/tmt/container/__init__.py b/tmt/container/__init__.py new file mode 100644 index 0000000000..fa939f7925 --- /dev/null +++ b/tmt/container/__init__.py @@ -0,0 +1,799 @@ +""" Tmt container decorators and helpers. """ + +import dataclasses +import functools +import inspect +import textwrap +from collections.abc import Iterator, Sequence +from typing import TYPE_CHECKING, Any, Callable, Generic, Optional, TypeVar, Union, cast, overload + +if TYPE_CHECKING: + from _typeshed import DataclassInstance + + import tmt.options + from tmt._compat.typing import TypeAlias + +import tmt.log + +# A stand-in variable for generic use. +T = TypeVar('T') + +container = dataclasses.dataclass # noqa: TID251 + +#: Type of field's normalization callback. +NormalizeCallback = Callable[[str, Any, tmt.log.Logger], T] + +#: Type of field's exporter callback. +FieldExporter = Callable[[T], Any] + +#: Type of field's CLI option specification. +FieldCLIOption = Union[str, Sequence[str]] + +#: Type of field's serialization callback. +SerializeCallback = Callable[[T], Any] + +#: Type of field's unserialization callback. +UnserializeCallback = Callable[[Any], T] + +#: Types for generic "data container" classes and instances. In tmt code, this +#: reduces to data classes and data class instances. Our :py:class:`DataContainer` +#: are perfectly compatible data classes, but some helper methods may be used +#: on raw data classes, not just on ``DataContainer`` instances. +ContainerClass: 'TypeAlias' = type['DataclassInstance'] +ContainerInstance: 'TypeAlias' = 'DataclassInstance' +Container = Union[ContainerClass, ContainerInstance] + + +def key_to_option(key: str) -> str: + """ Convert a key name to corresponding option name """ + + return key.replace('_', '-') + + +def option_to_key(option: str) -> str: + """ Convert an option name to corresponding key name """ + + return option.replace('-', '_') + + +@dataclasses.dataclass # noqa: TID251 +class FieldMetadata(Generic[T]): + """ + A dataclass metadata container used by our custom dataclass field management. + + Attached to fields defined with :py:func:`field` + """ + + internal: bool = False + + #: Help text documenting the field. + help: Optional[str] = None + + #: If field accepts a value, this string would represent it in documentation. + #: This stores the metavar provided when field was created - it may be unset. + #: py:attr:`metavar` provides the actual metavar to be used. + _metavar: Optional[str] = None + + #: The default value for the field. + default: Optional[T] = None + + #: A zero-argument callable that will be called when a default value is + #: needed for the field. + default_factory: Optional[Callable[[], T]] = None + + #: Marks the fields as a flag. + is_flag: bool = False + + #: Marks the field as accepting multiple values. When used on command line, + #: the option could be used multiple times, accumulating values. + multiple: bool = False + + #: If set, show the default value in command line help. + show_default: bool = False + + #: Either a list of allowed values the field can take, or a zero-argument + #: callable that would return such a list. + _choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None + + #: Environment variable providing value for the field. + envvar: Optional[str] = None + + #: Mark the option as deprecated. Instance of :py:class:`Deprecated` + #: describes the version in which the field was deprecated plus an optional + #: hint with the recommended alternative. Documentation and help texts would + #: contain this info. + deprecated: Optional['tmt.options.Deprecated'] = None + + #: One or more command-line option names. + cli_option: Optional[FieldCLIOption] = None + + #: A normalization callback to call when loading the value from key source + #: (performed by :py:class:`NormalizeKeysMixin`). + normalize_callback: Optional['NormalizeCallback[T]'] = None + + # Callbacks for custom serialize/unserialize operations (performed by + # :py:class:`SerializableContainer`). + serialize_callback: Optional['SerializeCallback[T]'] = None + unserialize_callback: Optional['SerializeCallback[T]'] = None + + #: An export callback to call when exporting the field (performed by + #: :py:class:`tmt.export.Exportable`). + export_callback: Optional['FieldExporter[T]'] = None + + #: CLI option parameters, for lazy option creation. + _option_args: Optional['FieldCLIOption'] = None + _option_kwargs: dict[str, Any] = dataclasses.field(default_factory=dict) # noqa: TID251 + + #: A :py:func:`click.option` decorator defining a corresponding CLI option. + _option: Optional['tmt.options.ClickOptionDecoratorType'] = None + + @functools.cached_property + def choices(self) -> Optional[Sequence[str]]: + """ A list of allowed values the field can take """ + + if isinstance(self._choices, (list, tuple)): + return list(self._choices) + + if callable(self._choices): + return self._choices() + + return None + + @functools.cached_property + def metavar(self) -> Optional[str]: + """ Placeholder for field's value in documentation and help """ + + if self._metavar: + return self._metavar + + if self.choices: + return '|'.join(self.choices) + + return None + + @property + def has_default(self) -> bool: + """ Whether the field has a default value """ + + return self.default_factory is not None \ + or self.default is not dataclasses.MISSING + + @property + def materialized_default(self) -> Optional[T]: + """ Returns the actual default value of the field """ + + if self.default_factory is not None: + return self.default_factory() + + if self.default is not dataclasses.MISSING: + return self.default + + return None + + @property + def option(self) -> Optional['tmt.options.ClickOptionDecoratorType']: + if self._option is None and self.cli_option: + from tmt.options import option + + self._option_args = (self.cli_option,) if isinstance(self.cli_option, str) \ + else self.cli_option + + self._option_kwargs.update({ + 'is_flag': self.is_flag, + 'multiple': self.multiple, + 'envvar': self.envvar, + 'metavar': self.metavar, + 'choices': self.choices, + 'show_default': self.show_default, + 'help': self.help, + 'deprecated': self.deprecated + }) + + if self.default is not dataclasses.MISSING and not self.is_flag: + self._option_kwargs['default'] = self.default + + self._option = option( + *self._option_args, + **self._option_kwargs + ) + + return self._option + + +def container_fields(container: Container) -> Iterator[dataclasses.Field[Any]]: + yield from dataclasses.fields(container) + + +def container_keys(container: Container) -> Iterator[str]: + """ Iterate over key names in a container """ + + for field in container_fields(container): + yield field.name + + +def container_values(container: ContainerInstance) -> Iterator[Any]: + """ Iterate over values in a container """ + + for field in container_fields(container): + yield container.__dict__[field.name] + + +def container_items(container: ContainerInstance) -> Iterator[tuple[str, Any]]: + """ Iterate over key/value pairs in a container """ + + for field in container_fields(container): + yield field.name, container.__dict__[field.name] + + +def container_field( + container: Container, + key: str) -> tuple[str, str, Any, dataclasses.Field[Any], 'FieldMetadata[Any]']: + """ + Return a dataclass/data container field info by the field's name. + + Surprisingly, :py:mod:`dataclasses` package does not have a helper for + this. One can iterate over fields, but there's no *public* API for + retrieving a field when one knows its name. + + :param cls: a dataclass/data container class whose fields to search. + :param key: field name to retrieve. + :raises GeneralError: when the field does not exist. + """ + import tmt.utils + + for field in container_fields(container): + if field.name != key: + continue + + metadata = field.metadata.get('tmt', FieldMetadata()) + return ( + field.name, + key_to_option(field.name), + container.__dict__[field.name] if not inspect.isclass(container) else None, + field, + metadata) + + if isinstance(container, DataContainer): + raise tmt.utils.GeneralError( + f"Could not find field '{key}' in class '{container.__class__.__name__}'.") + + raise tmt.utils.GeneralError(f"Could not find field '{key}' in class '{container}'.") + + +@container +class DataContainer: + """ A base class for objects that have keys and values """ + + def to_dict(self) -> dict[str, Any]: + """ + Convert to a mapping. + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + """ + + return dict(self.items()) + + def to_minimal_dict(self) -> dict[str, Any]: + """ + Convert to a mapping with unset keys omitted. + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + """ + + return { + key: value for key, value in self.items() if value is not None + } + + # This method should remain a class-method: 1. list of keys is known + # already, therefore it's not necessary to create an instance, and + # 2. some functionality makes use of this knowledge. + @classmethod + def keys(cls) -> Iterator[str]: + """ Iterate over key names """ + + yield from container_keys(cls) + + def values(self) -> Iterator[Any]: + """ Iterate over key values """ + + yield from container_values(self) + + def items(self) -> Iterator[tuple[str, Any]]: + """ Iterate over key/value pairs """ + + yield from container_items(self) + + @classmethod + def _default(cls, key: str, default: Any = None) -> Any: + """ + Return a default value for a given key. + + Keys may have a default value, or a default *factory* has been specified. + + :param key: key to look for. + :param default: when key has no default value, ``default`` is returned. + :returns: a default value defined for the key, or its ``default_factory``'s + return value of ``default_factory``, or ``default`` when key has no + default value. + """ + + for field in container_fields(cls): + if key != field.name: + continue + + if not isinstance(field.default_factory, dataclasses._MISSING_TYPE): + return field.default_factory() + + if not isinstance(field.default, dataclasses._MISSING_TYPE): + return field.default + + else: + return default + + @property + def is_bare(self) -> bool: + """ + Check whether all keys are either unset or have their default value. + + :returns: ``True`` if all keys either hold their default value + or are not set at all, ``False`` otherwise. + """ + + for field in container_fields(self): + value = getattr(self, field.name) + + if not isinstance(field.default_factory, dataclasses._MISSING_TYPE): + if value != field.default_factory(): + return False + + elif not isinstance(field.default, dataclasses._MISSING_TYPE): + if value != field.default: + return False + + else: + pass + + return True + + +#: A typevar bound to spec-based container base class. A stand-in for all classes +#: derived from :py:class:`SpecBasedContainer`. +SpecBasedContainerT = TypeVar( + 'SpecBasedContainerT', + # ignore[type-arg]: generic bounds are not supported by mypy. + bound='SpecBasedContainer') # type: ignore[type-arg] + +# It may look weird, having two different typevars for "spec", but it does make +# sense: tmt is fairly open to what it accepts, e.g. "a string or a list of +# strings". This is the input part of the flow. But then the input is normalized, +# and the output may be just a subset of types tmt is willing to accept. For +# example, if `tag` can be either a string or a list of strings, when processed +# by tmt and converted back to spec, a list of strings is the only output, even +# if the original was a single string. Therefore `SpecBasedContainer` accepts +# two types, one for each direction. Usually, the output one would be a subset +# of the input one. + +#: A typevar representing an *input* specification consumed by :py:class:`SpecBasedContainer`. +SpecInT = TypeVar('SpecInT') +#: A typevar representing an *output* specification produced by :py:class:`SpecBasedContainer`. +SpecOutT = TypeVar('SpecOutT') + + +@container +class SpecBasedContainer(Generic[SpecInT, SpecOutT], DataContainer): + @classmethod + def from_spec(cls: type[SpecBasedContainerT], spec: SpecInT) -> SpecBasedContainerT: + """ + Convert from a specification file or from a CLI option + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`to_spec` for its counterpart. + """ + + raise NotImplementedError + + def to_spec(self) -> SpecOutT: + """ + Convert to a form suitable for saving in a specification file + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`from_spec` for its counterpart. + """ + + return cast(SpecOutT, self.to_dict()) + + def to_minimal_spec(self) -> SpecOutT: + """ + Convert to specification, skip default values + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`from_spec` for its counterpart. + """ + + return cast(SpecOutT, self.to_minimal_dict()) + + +SerializableContainerDerivedType = TypeVar( + 'SerializableContainerDerivedType', + bound='SerializableContainer') + + +@container +class SerializableContainer(DataContainer): + """ A mixin class for saving and loading objects """ + + @classmethod + def default(cls, key: str, default: Any = None) -> Any: + return cls._default(key, default=default) + + # + # Moving data between containers and objects owning them + # + + def inject_to(self, obj: Any) -> None: + """ Inject keys from this container into attributes of a given object """ + + for name, value in self.items(): + setattr(obj, name, value) + + @classmethod + def extract_from(cls: type[SerializableContainerDerivedType], + obj: Any) -> SerializableContainerDerivedType: + """ Extract keys from given object, and save them in a container """ + + data = cls() + # SIM118: Use `{key} in {dict}` instead of `{key} in {dict}.keys()` + # "NormalizeKeysMixin" has no attribute "__iter__" (not iterable) + for key in cls.keys(): # noqa: SIM118 + value = getattr(obj, key) + if value is not None: + setattr(data, key, value) + + return data + + # + # Serialization - writing containers into YAML files, and restoring + # them later. + # + + def to_serialized(self) -> dict[str, Any]: + """ + Convert to a form suitable for saving in a file. + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`from_serialized` for its counterpart. + """ + + def _produce_serialized() -> Iterator[tuple[str, Any]]: + for key in container_keys(self): + _, option, value, _, metadata = container_field(self, key) + + if metadata.serialize_callback: + yield option, metadata.serialize_callback(value) + + else: + yield option, value + + serialized = dict(_produce_serialized()) + + # Add a special field tracking what class we just shattered to pieces. + serialized['__class__'] = { + 'module': self.__class__.__module__, + 'name': self.__class__.__name__ + } + + return serialized + + @classmethod + def from_serialized( + cls: type[SerializableContainerDerivedType], + serialized: dict[str, Any]) -> SerializableContainerDerivedType: + """ + Convert from a serialized form loaded from a file. + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`to_serialized` for its counterpart. + """ + + # Our special key may or may not be present, depending on who + # calls this method. In any case, it is not needed, because we + # already know what class to restore: this one. + serialized.pop('__class__', None) + + def _produce_unserialized() -> Iterator[tuple[str, Any]]: + for option, value in serialized.items(): + key = option_to_key(option) + + _, _, _, _, metadata = container_field(cls, key) + + if metadata.unserialize_callback: + yield key, metadata.unserialize_callback(value) + + else: + yield key, value + + # Set attribute by adding it to __dict__ directly. Messing with setattr() + # might cause reuse of mutable values by other instances. + # obj.__dict__[keyname] = unserialize_callback(value) + + return cls(**dict(_produce_unserialized())) + + # ignore[misc,type-var]: mypy is correct here, method does return a + # TypeVar, but there is no way to deduce the actual type, because + # the method is static. That's on purpose, method tries to find the + # class to unserialize, therefore it's simply unknown. Returning Any + # would make mypy happy, but we do know the return value will be + # derived from SerializableContainer. We can mention that, and + # silence mypy about the missing actual type. + @staticmethod + def unserialize( + serialized: dict[str, Any], + logger: tmt.log.Logger + ) -> SerializableContainerDerivedType: # type: ignore[misc,type-var] + """ + Convert from a serialized form loaded from a file. + + Similar to :py:meth:`from_serialized`, but this method knows + nothing about container's class, and will locate the correct + module and class by inspecting serialized data. Discovered + class' :py:meth:`from_serialized` is then used to create the + container. + + Used to transform data read from a YAML file into original + containers when their classes are not know to the code. + Restoring such containers requires inspection of serialized data + and dynamic imports of modules as needed. + + See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions + for more details. + + See :py:meth:`to_serialized` for its counterpart. + """ + + import tmt.utils + from tmt.plugins import import_member + + # Unpack class info, to get nicer variable names + if "__class__" not in serialized: + raise tmt.utils.GeneralError( + "Failed to load saved state, probably because of old data format.\n" + "Use 'tmt clean runs' to clean up old runs.") + + klass_info = serialized.pop('__class__') + klass = import_member( + module=klass_info['module'], + member=klass_info['name'], + logger=logger)[1] + + # Stay away from classes that are not derived from this one, to + # honor promise given by return value annotation. + assert issubclass(klass, SerializableContainer) + + # Apparently, the issubclass() check above is not good enough for mypy. + return cast(SerializableContainerDerivedType, klass.from_serialized(serialized)) + + +@overload +def field( + *, + default: bool, + # Options + option: Optional[FieldCLIOption] = None, + is_flag: bool = True, + choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, + multiple: bool = False, + metavar: Optional[str] = None, + envvar: Optional[str] = None, + deprecated: Optional['tmt.options.Deprecated'] = None, + help: Optional[str] = None, + show_default: bool = False, + internal: bool = False, + # Input data normalization - not needed, the field is a boolean + # flag. + # normalize: Optional[NormalizeCallback[T]] = None + # Custom serialization + # serialize: Optional[SerializeCallback[bool]] = None, + # unserialize: Optional[UnserializeCallback[bool]] = None + # Custom exporter + # exporter: Optional[FieldExporter[T]] = None + ) -> bool: + pass + + +@overload +def field( + *, + default: T, + # Options + option: Optional[FieldCLIOption] = None, + is_flag: bool = False, + choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, + multiple: bool = False, + metavar: Optional[str] = None, + envvar: Optional[str] = None, + deprecated: Optional['tmt.options.Deprecated'] = None, + help: Optional[str] = None, + show_default: bool = False, + internal: bool = False, + # Input data normalization + normalize: Optional[NormalizeCallback[T]] = None, + # Custom serialization + serialize: Optional[SerializeCallback[T]] = None, + unserialize: Optional[UnserializeCallback[T]] = None, + # Custom exporter + exporter: Optional[FieldExporter[T]] = None + ) -> T: + pass + + +@overload +def field( + *, + default_factory: Callable[[], T], + # Options + option: Optional[FieldCLIOption] = None, + is_flag: bool = False, + choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, + multiple: bool = False, + metavar: Optional[str] = None, + envvar: Optional[str] = None, + deprecated: Optional['tmt.options.Deprecated'] = None, + help: Optional[str] = None, + show_default: bool = False, + internal: bool = False, + # Input data normalization + normalize: Optional[NormalizeCallback[T]] = None, + # Custom serialization + serialize: Optional[SerializeCallback[T]] = None, + unserialize: Optional[UnserializeCallback[T]] = None, + # Custom exporter + exporter: Optional[FieldExporter[T]] = None + ) -> T: + pass + + +@overload +def field( + *, + # Options + option: Optional[FieldCLIOption] = None, + is_flag: bool = False, + choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, + multiple: bool = False, + metavar: Optional[str] = None, + envvar: Optional[str] = None, + deprecated: Optional['tmt.options.Deprecated'] = None, + help: Optional[str] = None, + show_default: bool = False, + internal: bool = False, + # Input data normalization + normalize: Optional[NormalizeCallback[T]] = None, + # Custom serialization + serialize: Optional[SerializeCallback[T]] = None, + unserialize: Optional[UnserializeCallback[T]] = None, + # Custom exporter + exporter: Optional[FieldExporter[T]] = None + ) -> T: + pass + + +def field( + *, + default: Any = dataclasses.MISSING, + default_factory: Any = None, + # Options + option: Optional[FieldCLIOption] = None, + is_flag: bool = False, + choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, + multiple: bool = False, + metavar: Optional[str] = None, + envvar: Optional[str] = None, + deprecated: Optional['tmt.options.Deprecated'] = None, + help: Optional[str] = None, + show_default: bool = False, + internal: bool = False, + # Input data normalization + normalize: Optional[NormalizeCallback[T]] = None, + # Custom serialization + serialize: Optional[SerializeCallback[T]] = None, + unserialize: Optional[UnserializeCallback[T]] = None, + # Custom exporter + exporter: Optional[FieldExporter[T]] = None + ) -> Any: + """ + Define a :py:class:`DataContainer` field. + + Effectively a fancy wrapper over :py:func:`dataclasses.field`, tailored for + tmt code needs and simplification of various common tasks. + + :param default: if provided, this will be the default value for this field. + Passed directly to :py:func:`dataclass.field`. + It is an error to specify both ``default`` and ``default_factory``. + :param default_factory: if provided, it must be a zero-argument callable + that will be called when a default value is needed for this field. + Passed directly to :py:func:`dataclass.field`. + It is an error to specify both ``default`` and ``default_factory``. + :param option: one or more command-line option names. + Passed directly to :py:func:`click.option`. + :param is_flag: marks this option as a flag. + Passed directly to :py:func:`click.option`. + :param choices: if provided, the command-line option would accept only + the listed input values. + Passed to :py:func:`click.option` as a :py:class:`click.Choice` instance. + :param multiple: accept multiple arguments of the same name. + Passed directly to :py:func:`click.option`. + :param metavar: how the input value is represented in the help page. + Passed directly to :py:func:`click.option`. + :param envvar: environment variable used for this option. + Passed directly to :py:func:`click.option`. + :param deprecated: mark the option as deprecated + Provide an instance of Deprecated() with version in which the + option was obsoleted and an optional hint with the recommended + alternative. A warning message will be added to the option help. + :param help: the help string for the command-line option. Multiline strings + can be used, :py:func:`textwrap.dedent` is applied before passing + ``help`` to :py:func:`click.option`. + :param show_default: show default value + Passed directly to :py:func:`click.option`. + :param internal: if set, the field is treated as internal-only, and will not + appear when showing objects via ``show()`` method, or in export created + by :py:meth:`Core._export`. + :param normalize: a callback for normalizing the input value. Consumed by + :py:class:`NormalizeKeysMixin`. + :param serialize: a callback for custom serialization of the field value. + Consumed by :py:class:`SerializableKeysMixin`. + :param unserialize: a callback for custom unserialization of the field value. + Consumed by :py:class:`SerializableKeysMixin`. + :param exporter: a callback for custom export of the field value. + Consumed by :py:class:`tmt.export.Exportable`. + """ + import tmt.utils + + if option: + if is_flag is False and isinstance(default, bool): + raise tmt.utils.GeneralError( + "Container field must be a flag to have boolean default value.") + + if is_flag is True and not isinstance(default, bool): + raise tmt.utils.GeneralError( + "Container field must have a boolean default value when it is a flag.") + + metadata: FieldMetadata[T] = FieldMetadata( + internal=internal, + help=textwrap.dedent(help).strip() if help else None, + _metavar=metavar, + default=default, + default_factory=default_factory, + show_default=show_default, + is_flag=is_flag, + multiple=multiple, + _choices=choices, + envvar=envvar, + deprecated=deprecated, + cli_option=option, + normalize_callback=normalize, + serialize_callback=serialize, + unserialize_callback=unserialize, + export_callback=exporter) + + # ignore[call-overload]: returning "wrong" type on purpose. field() must be annotated + # as if returning the value of type matching the field declaration, and the original + # field() is called with wider argument types than expected, because we use our own + # overloading to narrow types *our* custom field() accepts. + return dataclasses.field( # type: ignore[call-overload] # noqa: TID251 + default=default, + default_factory=default_factory or dataclasses.MISSING, + metadata={'tmt': metadata} + ) diff --git a/tmt/hardware.py b/tmt/hardware.py index d6cc484bfd..6293781e04 100644 --- a/tmt/hardware.py +++ b/tmt/hardware.py @@ -49,7 +49,8 @@ import tmt.log import tmt.utils -from tmt.utils import SpecBasedContainer, SpecificationError +from tmt.container import SpecBasedContainer, container +from tmt.utils import SpecificationError if TYPE_CHECKING: from pint import Quantity @@ -329,7 +330,7 @@ def __init__(self, constraint_name: str, raw_value: str, # Constraint classes # -@dataclasses.dataclass(repr=False) +@container(repr=False) class BaseConstraint(SpecBasedContainer[Spec, Spec]): """ Base class for all classes representing one or more constraints """ @@ -392,7 +393,7 @@ def variant(self) -> list['Constraint[Any]']: return variants[0] -@dataclasses.dataclass(repr=False) +@container(repr=False) class CompoundConstraint(BaseConstraint): """ Base class for all *compound* constraints """ @@ -463,7 +464,7 @@ def variants( raise NotImplementedError -@dataclasses.dataclass(repr=False) +@container(repr=False) class Constraint(BaseConstraint, Generic[ConstraintValueT]): """ A constraint imposing a particular limit to one of the guest properties """ @@ -791,7 +792,7 @@ def from_specification( ) -@dataclasses.dataclass(repr=False) +@container(repr=False) class And(CompoundConstraint): """ Represents constraints that are grouped in ``and`` fashion """ @@ -847,7 +848,7 @@ def variants( + simple_constraints -@dataclasses.dataclass(repr=False) +@container(repr=False) class Or(CompoundConstraint): """ Represents constraints that are grouped in ``or`` fashion """ @@ -1585,7 +1586,7 @@ def parse_hw_requirements(spec: Spec) -> BaseConstraint: return _parse_block(spec) -@dataclasses.dataclass +@container class Hardware(SpecBasedContainer[Spec, Spec]): constraint: Optional[BaseConstraint] spec: Spec diff --git a/tmt/result.py b/tmt/result.py index be495387b4..226775e800 100644 --- a/tmt/result.py +++ b/tmt/result.py @@ -1,4 +1,3 @@ -import dataclasses import enum import re from typing import TYPE_CHECKING, Any, Callable, Optional, cast @@ -11,7 +10,8 @@ import tmt.log import tmt.utils from tmt.checks import CheckEvent, CheckResultInterpret -from tmt.utils import GeneralError, Path, SerializableContainer, field +from tmt.container import SerializableContainer, container, field +from tmt.utils import GeneralError, Path if TYPE_CHECKING: import tmt.base @@ -125,7 +125,7 @@ def normalize( RawResult = Any -@dataclasses.dataclass +@container class ResultGuestData(SerializableContainer): """ Describes what tmt knows about a guest the result was produced on """ @@ -161,7 +161,7 @@ def _unserialize_fmf_id(serialized: 'tmt.base._RawFmfId') -> 'tmt.base.FmfId': return FmfId.from_spec(serialized) -@dataclasses.dataclass +@container class BaseResult(SerializableContainer): """ Describes what tmt knows about a result """ @@ -205,7 +205,7 @@ def show(self) -> str: return ' '.join(components) -@dataclasses.dataclass +@container class CheckResult(BaseResult): """ Describes what tmt knows about a single test check result """ @@ -220,7 +220,7 @@ def to_subcheck(self) -> 'SubCheckResult': return SubCheckResult.from_serialized(self.to_serialized()) -@dataclasses.dataclass +@container class SubCheckResult(CheckResult): """ Describes what tmt knows about a single subtest check result. @@ -231,7 +231,7 @@ class SubCheckResult(CheckResult): """ -@dataclasses.dataclass +@container class SubResult(BaseResult): """ Describes what tmt knows about a single test subresult """ @@ -243,12 +243,12 @@ class SubResult(BaseResult): ) -@dataclasses.dataclass +@container class PhaseResult(BaseResult): """ Describes what tmt knows about result of individual phases, e.g. prepare ansible """ -@dataclasses.dataclass +@container class Result(BaseResult): """ Describes what tmt knows about a single test result """ diff --git a/tmt/steps/__init__.py b/tmt/steps/__init__.py index 22eb5c5031..f0497ff897 100644 --- a/tmt/steps/__init__.py +++ b/tmt/steps/__init__.py @@ -35,6 +35,16 @@ import tmt.queue import tmt.utils import tmt.utils.rest +from tmt.container import ( + SerializableContainer, + SpecBasedContainer, + container, + container_field, + container_keys, + field, + key_to_option, + option_to_key, + ) from tmt.options import option, show_step_method_hints from tmt.utils import ( DEFAULT_NAME, @@ -43,13 +53,6 @@ GeneralError, Path, RunError, - SerializableContainer, - SpecBasedContainer, - container_field, - container_keys, - field, - key_to_option, - option_to_key, ) from tmt.utils.templates import render_template @@ -258,7 +261,7 @@ class _RawStepData(TypedDict, total=False): ResultT = TypeVar('ResultT', bound='BaseResult') -@dataclasses.dataclass +@container class StepData( SpecBasedContainer[_RawStepData, _RawStepData], tmt.utils.NormalizeKeysMixin, @@ -335,7 +338,7 @@ class RawWhereableStepData(TypedDict, total=False): where: Union[str, list[str]] -@dataclasses.dataclass +@container class WhereableStepData: """ Keys shared by step data that may be limited to a particular guest. @@ -1983,7 +1986,7 @@ def after_test( self._login(cwd, env) -@dataclasses.dataclass +@container class GuestTopology(SerializableContainer): """ Describes a guest in the topology of provisioned tmt guests """ @@ -1997,7 +2000,7 @@ def __init__(self, guest: 'Guest') -> None: self.hostname = guest.topology_address -@dataclasses.dataclass(init=False) +@container(init=False) class Topology(SerializableContainer): """ Describes the topology of provisioned tmt guests """ diff --git a/tmt/steps/discover/__init__.py b/tmt/steps/discover/__init__.py index ccc416fbc6..b8f163d59e 100644 --- a/tmt/steps/discover/__init__.py +++ b/tmt/steps/discover/__init__.py @@ -1,4 +1,3 @@ -import dataclasses from collections.abc import Iterator from typing import TYPE_CHECKING, Any, Optional, TypeVar, cast @@ -6,6 +5,7 @@ from fmf.utils import listed import tmt +from tmt.container import container, field, key_to_option if TYPE_CHECKING: import tmt.cli @@ -18,10 +18,10 @@ from tmt.options import option from tmt.plugins import PluginRegistry from tmt.steps import Action -from tmt.utils import GeneralError, Path, field, key_to_option +from tmt.utils import GeneralError, Path -@dataclasses.dataclass +@container class DiscoverStepData(tmt.steps.WhereableStepData, tmt.steps.StepData): dist_git_source: bool = field( default=False, diff --git a/tmt/steps/discover/fmf.py b/tmt/steps/discover/fmf.py index 2c4f5d6e12..7bc8ebc654 100644 --- a/tmt/steps/discover/fmf.py +++ b/tmt/steps/discover/fmf.py @@ -1,5 +1,4 @@ import contextlib -import dataclasses import glob import os import re @@ -19,8 +18,9 @@ import tmt.utils import tmt.utils.git from tmt.base import _RawAdjustRule +from tmt.container import container, field from tmt.steps.prepare.distgit import insert_to_prepare_step -from tmt.utils import Command, Environment, EnvVarValue, Path, field +from tmt.utils import Command, Environment, EnvVarValue, Path def normalize_ref( @@ -36,7 +36,7 @@ def normalize_ref( raise tmt.utils.NormalizationError(key_address, value, 'unset or a string') -@dataclasses.dataclass +@container class DiscoverFmfStepData(tmt.steps.discover.DiscoverStepData): # Basic options url: Optional[str] = field( diff --git a/tmt/steps/discover/shell.py b/tmt/steps/discover/shell.py index 986f0cd51c..e1820e616f 100644 --- a/tmt/steps/discover/shell.py +++ b/tmt/steps/discover/shell.py @@ -1,5 +1,4 @@ import copy -import dataclasses import shutil from typing import Any, Optional, TypeVar, cast @@ -13,22 +12,20 @@ import tmt.steps.discover import tmt.utils import tmt.utils.git +from tmt.container import SerializableContainer, SpecBasedContainer, container, field from tmt.steps.prepare.distgit import insert_to_prepare_step from tmt.utils import ( Command, Environment, EnvVarValue, Path, - SerializableContainer, ShellScript, - SpecBasedContainer, - field, ) T = TypeVar('T', bound='TestDescription') -@dataclasses.dataclass +@container class TestDescription( SpecBasedContainer[dict[str, Any], dict[str, Any]], tmt.utils.NormalizeKeysMixin, @@ -160,7 +157,7 @@ def to_spec(self) -> dict[str, Any]: return data -@dataclasses.dataclass +@container class DiscoverShellData(tmt.steps.discover.DiscoverStepData): tests: list[TestDescription] = field( default_factory=list, diff --git a/tmt/steps/execute/__init__.py b/tmt/steps/execute/__init__.py index c622d201af..a9bb1ea7a0 100644 --- a/tmt/steps/execute/__init__.py +++ b/tmt/steps/execute/__init__.py @@ -20,6 +20,7 @@ import tmt.steps import tmt.utils from tmt.checks import CheckEvent +from tmt.container import container, field from tmt.options import option from tmt.plugins import PluginRegistry from tmt.result import CheckResult, Result, ResultGuestData, ResultInterpret, ResultOutcome @@ -31,7 +32,6 @@ Path, ShellScript, Stopwatch, - field, format_duration, format_timestamp, ) @@ -244,7 +244,7 @@ def effective_scripts_dest_dir(default: Path = DEFAULT_SCRIPTS_DEST_DIR) -> Path ) -@dataclasses.dataclass +@container class ExecuteStepData(tmt.steps.WhereableStepData, tmt.steps.StepData): duration: str = field( # TODO: ugly circular dependency (see tmt.base.DEFAULT_TEST_DURATION_L2) diff --git a/tmt/steps/execute/internal.py b/tmt/steps/execute/internal.py index 92a5cd4aed..19aadb2523 100644 --- a/tmt/steps/execute/internal.py +++ b/tmt/steps/execute/internal.py @@ -1,4 +1,3 @@ -import dataclasses import os import subprocess import textwrap @@ -14,6 +13,7 @@ import tmt.steps import tmt.steps.execute import tmt.utils +from tmt.container import container, field from tmt.result import BaseResult, Result, ResultOutcome from tmt.steps import safe_filename from tmt.steps.execute import SCRIPTS, TEST_OUTPUT_FILENAME, TMT_REBOOT_SCRIPT, TestInvocation @@ -25,7 +25,6 @@ Path, ShellScript, Stopwatch, - field, format_duration, format_timestamp, ) @@ -187,7 +186,7 @@ def update(self, progress: str, test_name: str) -> None: # type: ignore[overrid self._update_message_area(message) -@dataclasses.dataclass +@container class ExecuteInternalData(tmt.steps.execute.ExecuteStepData): script: list[ShellScript] = field( default_factory=list, diff --git a/tmt/steps/execute/upgrade.py b/tmt/steps/execute/upgrade.py index 3e421cc655..d6fb56d0a1 100644 --- a/tmt/steps/execute/upgrade.py +++ b/tmt/steps/execute/upgrade.py @@ -1,4 +1,3 @@ -import dataclasses from typing import Any, Optional, Union, cast import fmf.utils @@ -11,13 +10,14 @@ import tmt.steps.execute import tmt.steps.provision import tmt.utils +from tmt.container import container, field, key_to_option from tmt.steps.discover import Discover, DiscoverPlugin, DiscoverStepData from tmt.steps.discover.fmf import DiscoverFmf, DiscoverFmfStepData, normalize_ref from tmt.steps.execute import ExecutePlugin from tmt.steps.execute.internal import ExecuteInternal, ExecuteInternalData from tmt.steps.prepare import PreparePlugin from tmt.steps.prepare.install import PrepareInstallData -from tmt.utils import Environment, EnvVarValue, Path, field +from tmt.utils import Environment, EnvVarValue, Path STATUS_VARIABLE = 'IN_PLACE_UPGRADE' BEFORE_UPGRADE_PREFIX = 'old' @@ -28,7 +28,7 @@ PROPAGATE_TO_DISCOVER_KEYS = ['url', 'ref', 'filter', 'test', 'exclude', 'upgrade_path'] -@dataclasses.dataclass +@container class ExecuteUpgradeData(ExecuteInternalData): url: Optional[str] = field( default=cast(Optional[str], None), @@ -291,7 +291,7 @@ def _prepare_remote_discover_data(self, plan: tmt.base.Plan) -> tmt.steps._RawSt 'how': 'fmf' } remote_raw_data.update(cast(tmt.steps._RawStepData, { - tmt.utils.key_to_option(key): value + key_to_option(key): value for key, value in data.items() if key in PROPAGATE_TO_DISCOVER_KEYS })) diff --git a/tmt/steps/finish/__init__.py b/tmt/steps/finish/__init__.py index 68996075c6..b3f6622320 100644 --- a/tmt/steps/finish/__init__.py +++ b/tmt/steps/finish/__init__.py @@ -1,5 +1,4 @@ import copy -import dataclasses from typing import TYPE_CHECKING, Any, Optional, TypeVar, Union, cast import click @@ -7,6 +6,7 @@ import tmt import tmt.steps +from tmt.container import container from tmt.options import option from tmt.plugins import PluginRegistry from tmt.result import PhaseResult @@ -25,7 +25,7 @@ import tmt.cli -@dataclasses.dataclass +@container class FinishStepData(tmt.steps.WhereableStepData, tmt.steps.StepData): pass diff --git a/tmt/steps/finish/shell.py b/tmt/steps/finish/shell.py index 2ac743c014..c2c6688f2c 100644 --- a/tmt/steps/finish/shell.py +++ b/tmt/steps/finish/shell.py @@ -1,4 +1,3 @@ -import dataclasses from typing import Any, Optional, cast import fmf @@ -7,15 +6,16 @@ import tmt.steps import tmt.steps.finish import tmt.utils +from tmt.container import container, field from tmt.result import PhaseResult from tmt.steps import safe_filename from tmt.steps.provision import Guest -from tmt.utils import ShellScript, field +from tmt.utils import ShellScript FINISH_WRAPPER_FILENAME = 'tmt-finish-wrapper.sh' -@dataclasses.dataclass +@container class FinishShellData(tmt.steps.finish.FinishStepData): script: list[ShellScript] = field( default_factory=list, diff --git a/tmt/steps/prepare/__init__.py b/tmt/steps/prepare/__init__.py index d995d02421..2100ca3225 100644 --- a/tmt/steps/prepare/__init__.py +++ b/tmt/steps/prepare/__init__.py @@ -12,6 +12,7 @@ import tmt.steps.discover import tmt.steps.provision import tmt.utils +from tmt.container import container from tmt.options import option from tmt.plugins import PluginRegistry from tmt.result import PhaseResult @@ -33,7 +34,7 @@ from tmt.base import Plan -@dataclasses.dataclass +@container class PrepareStepData(tmt.steps.WhereableStepData, tmt.steps.StepData): pass diff --git a/tmt/steps/prepare/ansible.py b/tmt/steps/prepare/ansible.py index 3b6d023c1a..d7c8c63c3f 100644 --- a/tmt/steps/prepare/ansible.py +++ b/tmt/steps/prepare/ansible.py @@ -1,4 +1,3 @@ -import dataclasses import tempfile from typing import Optional, Union @@ -12,6 +11,7 @@ import tmt.steps.prepare import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.result import PhaseResult from tmt.steps.provision import Guest from tmt.utils import ( @@ -19,7 +19,6 @@ ENVFILE_RETRY_SESSION_RETRIES, Path, PrepareError, - field, normalize_string_list, retry_session, ) @@ -30,7 +29,7 @@ class _RawAnsibleStepData(tmt.steps._RawStepData, total=False): playbooks: list[str] -@dataclasses.dataclass +@container class PrepareAnsibleData(tmt.steps.prepare.PrepareStepData): playbook: list[str] = field( default_factory=list, diff --git a/tmt/steps/prepare/distgit.py b/tmt/steps/prepare/distgit.py index a4a5e76f24..c0eb4beb4a 100644 --- a/tmt/steps/prepare/distgit.py +++ b/tmt/steps/prepare/distgit.py @@ -1,4 +1,3 @@ -import dataclasses import re from typing import TYPE_CHECKING, Any, Optional, cast @@ -7,12 +6,13 @@ import tmt.steps import tmt.steps.prepare import tmt.utils +from tmt.container import container, field from tmt.package_managers import Package from tmt.result import PhaseResult from tmt.steps.prepare import PreparePlugin from tmt.steps.prepare.install import PrepareInstallData from tmt.steps.provision import Guest -from tmt.utils import Command, Path, ShellScript, field, uniq +from tmt.utils import Command, Path, ShellScript, uniq if TYPE_CHECKING: import tmt.base @@ -102,7 +102,7 @@ def insert_to_prepare_step( ) -@dataclasses.dataclass +@container class DistGitData(tmt.steps.prepare.PrepareStepData): source_dir: Optional[Path] = field( default=None, diff --git a/tmt/steps/prepare/feature.py b/tmt/steps/prepare/feature.py index 7d0af56923..1b7e1129a1 100644 --- a/tmt/steps/prepare/feature.py +++ b/tmt/steps/prepare/feature.py @@ -1,4 +1,3 @@ -import dataclasses from typing import Optional, cast import tmt @@ -9,9 +8,10 @@ import tmt.steps.prepare import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.result import PhaseResult from tmt.steps.provision import Guest -from tmt.utils import Path, field +from tmt.utils import Path FEATURE_PLAYEBOOK_DIRECTORY = tmt.utils.resource_files('steps/prepare/feature') @@ -79,7 +79,7 @@ def disable(self) -> None: } -@dataclasses.dataclass +@container class PrepareFeatureData(tmt.steps.prepare.PrepareStepData): epel: Optional[str] = field( default=None, diff --git a/tmt/steps/prepare/install.py b/tmt/steps/prepare/install.py index 8b99bc9623..9ffd294007 100644 --- a/tmt/steps/prepare/install.py +++ b/tmt/steps/prepare/install.py @@ -1,4 +1,3 @@ -import dataclasses import re import shutil from collections.abc import Iterator @@ -15,6 +14,7 @@ import tmt.steps import tmt.steps.prepare import tmt.utils +from tmt.container import container, field from tmt.package_managers import ( FileSystemPath, Installable, @@ -25,7 +25,7 @@ ) from tmt.result import PhaseResult from tmt.steps.provision import Guest -from tmt.utils import Command, Path, ShellScript, field +from tmt.utils import Command, Path, ShellScript COPR_URL = 'https://copr.fedorainfracloud.org/coprs' @@ -526,7 +526,7 @@ def install_debuginfo(self) -> None: 'installing debuginfo packages.') -@dataclasses.dataclass +@container class PrepareInstallData(tmt.steps.prepare.PrepareStepData): package: list[tmt.base.DependencySimple] = field( default_factory=list, diff --git a/tmt/steps/prepare/shell.py b/tmt/steps/prepare/shell.py index 38415ca409..12e7f324b0 100644 --- a/tmt/steps/prepare/shell.py +++ b/tmt/steps/prepare/shell.py @@ -1,4 +1,3 @@ -import dataclasses from typing import Any, Optional, cast import fmf @@ -9,15 +8,16 @@ import tmt.steps import tmt.steps.prepare import tmt.utils +from tmt.container import container, field from tmt.result import PhaseResult from tmt.steps import safe_filename from tmt.steps.provision import Guest -from tmt.utils import ShellScript, field +from tmt.utils import ShellScript PREPARE_WRAPPER_FILENAME = 'tmt-prepare-wrapper.sh' -@dataclasses.dataclass +@container class PrepareShellData(tmt.steps.prepare.PrepareStepData): script: list[ShellScript] = field( default_factory=list, diff --git a/tmt/steps/provision/__init__.py b/tmt/steps/provision/__init__.py index 23c83998eb..d8076b9d08 100644 --- a/tmt/steps/provision/__init__.py +++ b/tmt/steps/provision/__init__.py @@ -40,6 +40,7 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import SerializableContainer, container, field, key_to_option from tmt.log import Logger from tmt.options import option from tmt.package_managers import FileSystemPath, Package, PackageManagerClass @@ -50,12 +51,9 @@ OnProcessStartCallback, Path, ProvisionError, - SerializableContainer, ShellScript, configure_constant, effective_workdir_root, - field, - key_to_option, ) if TYPE_CHECKING: @@ -179,7 +177,7 @@ class GuestCapability(enum.Enum): SYSLOG_ACTION_READ_CLEAR = 'syslog-action-read-clear' -@dataclasses.dataclass +@container class GuestFacts(SerializableContainer): """ Contains interesting facts about the guest. @@ -623,7 +621,7 @@ def _drop_placeholders(data: dict[str, Any]) -> dict[str, Any]: GuestDataT = TypeVar('GuestDataT', bound='GuestData') -@dataclasses.dataclass +@container class GuestData(SerializableContainer): """ Keys necessary to describe, create, save and restore a guest. @@ -756,7 +754,7 @@ def show( printable_value = str(value) logger.info( - tmt.utils.key_to_option(key).replace('-', ' '), + key_to_option(key).replace('-', ' '), printable_value, color='green') @@ -1362,7 +1360,7 @@ def essential_requires(cls) -> list['tmt.base.Dependency']: return [] -@dataclasses.dataclass +@container class GuestSshData(GuestData): """ Keys necessary to describe, create, save and restore a guest with SSH @@ -2084,7 +2082,7 @@ def remove(self) -> None: self.debug(f"Doing nothing to remove guest '{self.primary_address}'.") -@dataclasses.dataclass +@container class ProvisionStepData(tmt.steps.StepData): # guest role in the multihost scenario role: Optional[str] = None diff --git a/tmt/steps/provision/artemis.py b/tmt/steps/provision/artemis.py index 9f7e9078e9..cc7ee564b4 100644 --- a/tmt/steps/provision/artemis.py +++ b/tmt/steps/provision/artemis.py @@ -1,4 +1,3 @@ -import dataclasses import datetime import functools from typing import Any, Optional, TypedDict, cast @@ -12,11 +11,11 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.utils import ( ProvisionError, UpdatableMessage, dict_to_yaml, - field, normalize_string_dict, retry_session, ) @@ -101,7 +100,7 @@ def _normalize_log_type( key_address, raw_value, 'a string or a list of strings') -@dataclasses.dataclass +@container class ArtemisGuestData(tmt.steps.provision.GuestSshData): user: str = field( default=DEFAULT_USER, @@ -251,7 +250,7 @@ class ArtemisGuestData(tmt.steps.provision.GuestSshData): help='If set, the script provided or fetched will be executed.') -@dataclasses.dataclass +@container class ProvisionArtemisData(ArtemisGuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/provision/bootc.py b/tmt/steps/provision/bootc.py index d1131e8d79..6a7b2da988 100644 --- a/tmt/steps/provision/bootc.py +++ b/tmt/steps/provision/bootc.py @@ -1,4 +1,3 @@ -import dataclasses import os from typing import TYPE_CHECKING, Optional, cast @@ -10,8 +9,8 @@ import tmt.steps.provision import tmt.steps.provision.testcloud import tmt.utils +from tmt.container import container, field from tmt.steps.provision.testcloud import GuestTestcloud -from tmt.utils import field from tmt.utils.templates import render_template if TYPE_CHECKING: @@ -86,7 +85,7 @@ def remove(self) -> None: super().remove() -@dataclasses.dataclass +@container class BootcData(tmt.steps.provision.testcloud.ProvisionTestcloudData): container_file: Optional[str] = field( default=None, diff --git a/tmt/steps/provision/connect.py b/tmt/steps/provision/connect.py index 58608e286c..b987b933aa 100644 --- a/tmt/steps/provision/connect.py +++ b/tmt/steps/provision/connect.py @@ -1,16 +1,16 @@ -import dataclasses from typing import Any, Optional, Union import tmt import tmt.steps import tmt.steps.provision import tmt.utils -from tmt.utils import Command, ShellScript, field +from tmt.container import container, field +from tmt.utils import Command, ShellScript DEFAULT_USER = "root" -@dataclasses.dataclass +@container class ConnectGuestData(tmt.steps.provision.GuestSshData): # Connect plugin actually allows `guest` key to be controlled by an option. _OPTIONLESS_FIELDS = tuple( @@ -74,7 +74,7 @@ def from_plugin( return ConnectGuestData(**options) -@dataclasses.dataclass +@container class ProvisionConnectData(ConnectGuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/provision/local.py b/tmt/steps/provision/local.py index 78d2de1d3c..219dc555bf 100644 --- a/tmt/steps/provision/local.py +++ b/tmt/steps/provision/local.py @@ -1,4 +1,3 @@ -import dataclasses from typing import Any, Optional, Union import tmt @@ -7,10 +6,11 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import container from tmt.utils import Command, OnProcessStartCallback, Path, ShellScript -@dataclasses.dataclass +@container class ProvisionLocalData(tmt.steps.provision.GuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/provision/mrack.py b/tmt/steps/provision/mrack.py index 0439990957..ca94bb070b 100644 --- a/tmt/steps/provision/mrack.py +++ b/tmt/steps/provision/mrack.py @@ -18,13 +18,13 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.utils import ( Command, Path, ProvisionError, ShellScript, UpdatableMessage, - field, ) MRACK_VERSION: Optional[str] = None @@ -837,7 +837,7 @@ def update_wrapper(*args: Any, **kwargs: Any) -> Any: return update_wrapper -@dataclasses.dataclass +@container class BeakerGuestData(tmt.steps.provision.GuestSshData): # Override parent class with our defaults user: str = field( @@ -915,7 +915,7 @@ class BeakerGuestData(tmt.steps.provision.GuestSshData): """) -@dataclasses.dataclass +@container class ProvisionBeakerData(BeakerGuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/provision/podman.py b/tmt/steps/provision/podman.py index 36f6533be2..cf5f780668 100644 --- a/tmt/steps/provision/podman.py +++ b/tmt/steps/provision/podman.py @@ -1,4 +1,3 @@ -import dataclasses import os from shlex import quote from typing import Any, Optional, Union, cast @@ -9,8 +8,9 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.steps.provision import GuestCapability -from tmt.utils import Command, OnProcessStartCallback, Path, ShellScript, field, retry +from tmt.utils import Command, OnProcessStartCallback, Path, ShellScript, retry # Timeout in seconds of waiting for a connection CONNECTION_TIMEOUT = 60 @@ -24,7 +24,7 @@ DEFAULT_STOP_TIME = 1 -@dataclasses.dataclass +@container class PodmanGuestData(tmt.steps.provision.GuestData): image: str = field( default=DEFAULT_IMAGE, @@ -83,7 +83,7 @@ class PodmanGuestData(tmt.steps.provision.GuestData): normalize=tmt.utils.normalize_int) -@dataclasses.dataclass +@container class ProvisionPodmanData(PodmanGuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/provision/testcloud.py b/tmt/steps/provision/testcloud.py index 4120b6e6aa..1b54343dea 100644 --- a/tmt/steps/provision/testcloud.py +++ b/tmt/steps/provision/testcloud.py @@ -1,6 +1,5 @@ import collections -import dataclasses import datetime import functools import itertools @@ -23,6 +22,7 @@ import tmt.steps import tmt.steps.provision import tmt.utils +from tmt.container import container, field from tmt.utils import ( WORKDIR_ROOT, Command, @@ -30,7 +30,6 @@ ProvisionError, ShellScript, configure_constant, - field, retry_session, ) @@ -297,7 +296,7 @@ def _report_hw_requirement_support(constraint: tmt.hardware.Constraint[Any]) -> return False -@dataclasses.dataclass +@container class TestcloudGuestData(tmt.steps.provision.GuestSshData): # Override parent class with our defaults user: str = field( @@ -376,7 +375,7 @@ def show( logger.info('disk', f'{(self.disk or DEFAULT_DISK).to("GB")}', 'green') -@dataclasses.dataclass +@container class ProvisionTestcloudData(TestcloudGuestData, tmt.steps.provision.ProvisionStepData): pass diff --git a/tmt/steps/report/__init__.py b/tmt/steps/report/__init__.py index 813ce3bb0c..e321459fbe 100644 --- a/tmt/steps/report/__init__.py +++ b/tmt/steps/report/__init__.py @@ -1,4 +1,3 @@ -import dataclasses from typing import TYPE_CHECKING, Any, Optional, TypeVar, Union, cast import click @@ -6,6 +5,7 @@ import tmt import tmt.plugins import tmt.steps +from tmt.container import container from tmt.options import option from tmt.plugins import PluginRegistry from tmt.steps import Action @@ -14,7 +14,7 @@ import tmt.cli -@dataclasses.dataclass +@container class ReportStepData(tmt.steps.StepData): pass diff --git a/tmt/steps/report/display.py b/tmt/steps/report/display.py index af33095620..e872062950 100644 --- a/tmt/steps/report/display.py +++ b/tmt/steps/report/display.py @@ -1,4 +1,3 @@ -import dataclasses from collections.abc import Sequence from typing import Optional @@ -6,9 +5,10 @@ import tmt.log import tmt.steps import tmt.steps.report +from tmt.container import container, field from tmt.result import BaseResult, CheckResult, Result, SubCheckResult, SubResult from tmt.steps.execute import TEST_OUTPUT_FILENAME -from tmt.utils import Path, field +from tmt.utils import Path # How much test and test check info should be shifted to the right in the output. # We want tests to be shifted by one extra level, with their checks shifted by @@ -19,7 +19,7 @@ SUBRESULT_CHECK_SHIFT = SUBRESULT_SHIFT + 1 -@dataclasses.dataclass +@container class ReportDisplayData(tmt.steps.report.ReportStepData): display_guest: str = field( default='auto', diff --git a/tmt/steps/report/html.py b/tmt/steps/report/html.py index 6cb604c92a..4b32372097 100644 --- a/tmt/steps/report/html.py +++ b/tmt/steps/report/html.py @@ -1,4 +1,3 @@ -import dataclasses import webbrowser from typing import Optional @@ -11,12 +10,13 @@ import tmt.steps.report import tmt.utils import tmt.utils.templates -from tmt.utils import Path, field +from tmt.container import container, field +from tmt.utils import Path HTML_TEMPLATE_PATH = tmt.utils.resource_files('steps/report/html/template.html.j2') -@dataclasses.dataclass +@container class ReportHtmlData(tmt.steps.report.ReportStepData): open: bool = field( default=False, diff --git a/tmt/steps/report/junit.py b/tmt/steps/report/junit.py index 76bcb7430c..c78615d3b4 100644 --- a/tmt/steps/report/junit.py +++ b/tmt/steps/report/junit.py @@ -1,4 +1,3 @@ -import dataclasses import functools import re import sys @@ -15,9 +14,10 @@ import tmt.steps import tmt.steps.report import tmt.utils +from tmt.container import container, field from tmt.plugins import ModuleImporter from tmt.result import ResultOutcome -from tmt.utils import Path, field +from tmt.utils import Path from tmt.utils.templates import default_template_environment, render_template_file if TYPE_CHECKING: @@ -377,7 +377,7 @@ def _read_log_filter(log: Path) -> str: return str(xml_output.decode('utf-8')) -@dataclasses.dataclass +@container class ReportJUnitData(tmt.steps.report.ReportStepData): file: Optional[Path] = field( default=None, diff --git a/tmt/steps/report/polarion.py b/tmt/steps/report/polarion.py index b56297c08d..f428e79442 100644 --- a/tmt/steps/report/polarion.py +++ b/tmt/steps/report/polarion.py @@ -1,4 +1,3 @@ -import dataclasses import datetime import os from typing import Optional @@ -9,14 +8,15 @@ import tmt.steps import tmt.steps.report import tmt.utils -from tmt.utils import Path, field +from tmt.container import container, field +from tmt.utils import Path from .junit import ResultsContext, make_junit_xml DEFAULT_NAME = 'xunit.xml' -@dataclasses.dataclass +@container class ReportPolarionData(tmt.steps.report.ReportStepData): file: Optional[Path] = field( default=None, diff --git a/tmt/steps/report/reportportal.py b/tmt/steps/report/reportportal.py index cce7e9fda1..455458f59e 100644 --- a/tmt/steps/report/reportportal.py +++ b/tmt/steps/report/reportportal.py @@ -10,8 +10,9 @@ import tmt.log import tmt.steps.report import tmt.utils +from tmt.container import container, field from tmt.result import ResultOutcome -from tmt.utils import field, yaml_to_dict +from tmt.utils import yaml_to_dict if TYPE_CHECKING: from tmt._compat.typing import TypeAlias @@ -95,7 +96,7 @@ def _filter_log(log: str, settings: Optional[LogFilterSettings] = None) -> str: return log -@dataclasses.dataclass +@container class ReportReportPortalData(tmt.steps.report.ReportStepData): url: Optional[str] = field( diff --git a/tmt/utils/__init__.py b/tmt/utils/__init__.py index c2c3b36ea1..5d75df03bb 100644 --- a/tmt/utils/__init__.py +++ b/tmt/utils/__init__.py @@ -7,7 +7,6 @@ import enum import functools import importlib.resources -import inspect import io import json import os @@ -25,7 +24,7 @@ import unicodedata import urllib.parse from collections import Counter -from collections.abc import Iterable, Iterator, Sequence +from collections.abc import Iterable, Iterator from contextlib import suppress from math import ceil from re import Pattern @@ -68,11 +67,8 @@ from tmt.log import LoggableValue if TYPE_CHECKING: - from _typeshed import DataclassInstance - import tmt.base import tmt.cli - import tmt.options import tmt.steps from tmt._compat.typing import Self, TypeAlias from tmt.hardware import Size @@ -2904,571 +2900,6 @@ def json_to_list(data: Any) -> list[Any]: return loaded_data -#: A type representing compatible sources of keys and values. -KeySource = Union[dict[str, Any], fmf.Tree] - -#: Type of field's normalization callback. -NormalizeCallback = Callable[[str, Any, tmt.log.Logger], T] - -#: Type of field's exporter callback. -FieldExporter = Callable[[T], Any] - -#: Type of field's CLI option specification. -FieldCLIOption = Union[str, Sequence[str]] - -#: Type of field's serialization callback. -SerializeCallback = Callable[[T], Any] - -#: Type of field's unserialization callback. -UnserializeCallback = Callable[[Any], T] - -#: Types for generic "data container" classes and instances. In tmt code, this -#: reduces to data classes and data class instances. Our :py:class:`DataContainer` -#: are perfectly compatible data classes, but some helper methods may be used -#: on raw data classes, not just on ``DataContainer`` instances. -ContainerClass: 'TypeAlias' = type['DataclassInstance'] -ContainerInstance: 'TypeAlias' = 'DataclassInstance' -Container = Union[ContainerClass, ContainerInstance] - - -def key_to_option(key: str) -> str: - """ Convert a key name to corresponding option name """ - - return key.replace('_', '-') - - -def option_to_key(option: str) -> str: - """ Convert an option name to corresponding key name """ - - return option.replace('-', '_') - - -@dataclasses.dataclass # noqa: TID251 -class FieldMetadata(Generic[T]): - """ - A dataclass metadata container used by our custom dataclass field management. - - Attached to fields defined with :py:func:`field` - """ - - internal: bool = False - - #: Help text documenting the field. - help: Optional[str] = None - - #: If field accepts a value, this string would represent it in documentation. - #: This stores the metavar provided when field was created - it may be unset. - #: py:attr:`metavar` provides the actual metavar to be used. - _metavar: Optional[str] = None - - #: The default value for the field. - default: Optional[T] = None - - #: A zero-argument callable that will be called when a default value is - #: needed for the field. - default_factory: Optional[Callable[[], T]] = None - - #: Marks the fields as a flag. - is_flag: bool = False - - #: Marks the field as accepting multiple values. When used on command line, - #: the option could be used multiple times, accumulating values. - multiple: bool = False - - #: If set, show the default value in command line help. - show_default: bool = False - - #: Either a list of allowed values the field can take, or a zero-argument - #: callable that would return such a list. - _choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None - - #: Environment variable providing value for the field. - envvar: Optional[str] = None - - #: Mark the option as deprecated. Instance of :py:class:`Deprecated` - #: describes the version in which the field was deprecated plus an optional - #: hint with the recommended alternative. Documentation and help texts would - #: contain this info. - deprecated: Optional['tmt.options.Deprecated'] = None - - #: One or more command-line option names. - cli_option: Optional[FieldCLIOption] = None - - #: A normalization callback to call when loading the value from key source - #: (performed by :py:class:`NormalizeKeysMixin`). - normalize_callback: Optional['NormalizeCallback[T]'] = None - - # Callbacks for custom serialize/unserialize operations (performed by - # :py:class:`SerializableContainer`). - serialize_callback: Optional['SerializeCallback[T]'] = None - unserialize_callback: Optional['SerializeCallback[T]'] = None - - #: An export callback to call when exporting the field (performed by - #: :py:class:`tmt.export.Exportable`). - export_callback: Optional['FieldExporter[T]'] = None - - #: CLI option parameters, for lazy option creation. - _option_args: Optional['FieldCLIOption'] = None - _option_kwargs: dict[str, Any] = dataclasses.field(default_factory=dict) # noqa: TID251 - - #: A :py:func:`click.option` decorator defining a corresponding CLI option. - _option: Optional['tmt.options.ClickOptionDecoratorType'] = None - - @functools.cached_property - def choices(self) -> Optional[Sequence[str]]: - """ A list of allowed values the field can take """ - - if isinstance(self._choices, (list, tuple)): - return list(self._choices) - - if callable(self._choices): - return self._choices() - - return None - - @functools.cached_property - def metavar(self) -> Optional[str]: - """ Placeholder for field's value in documentation and help """ - - if self._metavar: - return self._metavar - - if self.choices: - return '|'.join(self.choices) - - return None - - @property - def has_default(self) -> bool: - """ Whether the field has a default value """ - - return self.default_factory is not None \ - or self.default is not dataclasses.MISSING - - @property - def materialized_default(self) -> Optional[T]: - """ Returns the actual default value of the field """ - - if self.default_factory is not None: - return self.default_factory() - - if self.default is not dataclasses.MISSING: - return self.default - - return None - - @property - def option(self) -> Optional['tmt.options.ClickOptionDecoratorType']: - if self._option is None and self.cli_option: - from tmt.options import option - - self._option_args = (self.cli_option,) if isinstance(self.cli_option, str) \ - else self.cli_option - - self._option_kwargs.update({ - 'is_flag': self.is_flag, - 'multiple': self.multiple, - 'envvar': self.envvar, - 'metavar': self.metavar, - 'choices': self.choices, - 'show_default': self.show_default, - 'help': self.help, - 'deprecated': self.deprecated - }) - - if self.default is not dataclasses.MISSING and not self.is_flag: - self._option_kwargs['default'] = self.default - - self._option = option( - *self._option_args, - **self._option_kwargs - ) - - return self._option - - -def container_fields(container: Container) -> Iterator[dataclasses.Field[Any]]: - yield from dataclasses.fields(container) - - -def container_keys(container: Container) -> Iterator[str]: - """ Iterate over key names in a container """ - - for field in container_fields(container): - yield field.name - - -def container_values(container: ContainerInstance) -> Iterator[Any]: - """ Iterate over values in a container """ - - for field in container_fields(container): - yield container.__dict__[field.name] - - -def container_items(container: ContainerInstance) -> Iterator[tuple[str, Any]]: - """ Iterate over key/value pairs in a container """ - - for field in container_fields(container): - yield field.name, container.__dict__[field.name] - - -def container_field( - container: Container, - key: str) -> tuple[str, str, Any, dataclasses.Field[Any], 'FieldMetadata[Any]']: - """ - Return a dataclass/data container field info by the field's name. - - Surprisingly, :py:mod:`dataclasses` package does not have a helper for - this. One can iterate over fields, but there's no *public* API for - retrieving a field when one knows its name. - - :param cls: a dataclass/data container class whose fields to search. - :param key: field name to retrieve. - :raises GeneralError: when the field does not exist. - """ - - for field in container_fields(container): - if field.name != key: - continue - - metadata = field.metadata.get('tmt', FieldMetadata()) - return ( - field.name, - key_to_option(field.name), - container.__dict__[field.name] if not inspect.isclass(container) else None, - field, - metadata) - - if isinstance(container, DataContainer): - raise GeneralError( - f"Could not find field '{key}' in class '{container.__class__.__name__}'.") - - raise GeneralError(f"Could not find field '{key}' in class '{container}'.") - - -@dataclasses.dataclass -class DataContainer: - """ A base class for objects that have keys and values """ - - def to_dict(self) -> dict[str, Any]: - """ - Convert to a mapping. - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - """ - - return dict(self.items()) - - def to_minimal_dict(self) -> dict[str, Any]: - """ - Convert to a mapping with unset keys omitted. - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - """ - - return { - key: value for key, value in self.items() if value is not None - } - - # This method should remain a class-method: 1. list of keys is known - # already, therefore it's not necessary to create an instance, and - # 2. some functionality makes use of this knowledge. - @classmethod - def keys(cls) -> Iterator[str]: - """ Iterate over key names """ - - yield from container_keys(cls) - - def values(self) -> Iterator[Any]: - """ Iterate over key values """ - - yield from container_values(self) - - def items(self) -> Iterator[tuple[str, Any]]: - """ Iterate over key/value pairs """ - - yield from container_items(self) - - @classmethod - def _default(cls, key: str, default: Any = None) -> Any: - """ - Return a default value for a given key. - - Keys may have a default value, or a default *factory* has been specified. - - :param key: key to look for. - :param default: when key has no default value, ``default`` is returned. - :returns: a default value defined for the key, or its ``default_factory``'s - return value of ``default_factory``, or ``default`` when key has no - default value. - """ - - for field in container_fields(cls): - if key != field.name: - continue - - if not isinstance(field.default_factory, dataclasses._MISSING_TYPE): - return field.default_factory() - - if not isinstance(field.default, dataclasses._MISSING_TYPE): - return field.default - - else: - return default - - @property - def is_bare(self) -> bool: - """ - Check whether all keys are either unset or have their default value. - - :returns: ``True`` if all keys either hold their default value - or are not set at all, ``False`` otherwise. - """ - - for field in container_fields(self): - value = getattr(self, field.name) - - if not isinstance(field.default_factory, dataclasses._MISSING_TYPE): - if value != field.default_factory(): - return False - - elif not isinstance(field.default, dataclasses._MISSING_TYPE): - if value != field.default: - return False - - else: - pass - - return True - - -#: A typevar bound to spec-based container base class. A stand-in for all classes -#: derived from :py:class:`SpecBasedContainer`. -SpecBasedContainerT = TypeVar( - 'SpecBasedContainerT', - # ignore[type-arg]: generic bounds are not supported by mypy. - bound='SpecBasedContainer') # type: ignore[type-arg] - -# It may look weird, having two different typevars for "spec", but it does make -# sense: tmt is fairly open to what it accepts, e.g. "a string or a list of -# strings". This is the input part of the flow. But then the input is normalized, -# and the output may be just a subset of types tmt is willing to accept. For -# example, if `tag` can be either a string or a list of strings, when processed -# by tmt and converted back to spec, a list of strings is the only output, even -# if the original was a single string. Therefore `SpecBasedContainer` accepts -# two types, one for each direction. Usually, the output one would be a subset -# of the input one. - -#: A typevar representing an *input* specification consumed by :py:class:`SpecBasedContainer`. -SpecInT = TypeVar('SpecInT') -#: A typevar representing an *output* specification produced by :py:class:`SpecBasedContainer`. -SpecOutT = TypeVar('SpecOutT') - - -class SpecBasedContainer(Generic[SpecInT, SpecOutT], DataContainer): - @classmethod - def from_spec(cls: type[SpecBasedContainerT], spec: SpecInT) -> SpecBasedContainerT: - """ - Convert from a specification file or from a CLI option - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`to_spec` for its counterpart. - """ - - raise NotImplementedError - - def to_spec(self) -> SpecOutT: - """ - Convert to a form suitable for saving in a specification file - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`from_spec` for its counterpart. - """ - - return cast(SpecOutT, self.to_dict()) - - def to_minimal_spec(self) -> SpecOutT: - """ - Convert to specification, skip default values - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`from_spec` for its counterpart. - """ - - return cast(SpecOutT, self.to_minimal_dict()) - - -SerializableContainerDerivedType = TypeVar( - 'SerializableContainerDerivedType', - bound='SerializableContainer') - - -@dataclasses.dataclass -class SerializableContainer(DataContainer): - """ A mixin class for saving and loading objects """ - - @classmethod - def default(cls, key: str, default: Any = None) -> Any: - return cls._default(key, default=default) - - # - # Moving data between containers and objects owning them - # - - def inject_to(self, obj: Any) -> None: - """ Inject keys from this container into attributes of a given object """ - - for name, value in self.items(): - setattr(obj, name, value) - - @classmethod - def extract_from(cls: type[SerializableContainerDerivedType], - obj: Any) -> SerializableContainerDerivedType: - """ Extract keys from given object, and save them in a container """ - - data = cls() - # SIM118: Use `{key} in {dict}` instead of `{key} in {dict}.keys()` - # "NormalizeKeysMixin" has no attribute "__iter__" (not iterable) - for key in cls.keys(): # noqa: SIM118 - value = getattr(obj, key) - if value is not None: - setattr(data, key, value) - - return data - - # - # Serialization - writing containers into YAML files, and restoring - # them later. - # - - def to_serialized(self) -> dict[str, Any]: - """ - Convert to a form suitable for saving in a file. - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`from_serialized` for its counterpart. - """ - - def _produce_serialized() -> Iterator[tuple[str, Any]]: - for key in container_keys(self): - _, option, value, _, metadata = container_field(self, key) - - if metadata.serialize_callback: - yield option, metadata.serialize_callback(value) - - else: - yield option, value - - serialized = dict(_produce_serialized()) - - # Add a special field tracking what class we just shattered to pieces. - serialized['__class__'] = { - 'module': self.__class__.__module__, - 'name': self.__class__.__name__ - } - - return serialized - - @classmethod - def from_serialized( - cls: type[SerializableContainerDerivedType], - serialized: dict[str, Any]) -> SerializableContainerDerivedType: - """ - Convert from a serialized form loaded from a file. - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`to_serialized` for its counterpart. - """ - - # Our special key may or may not be present, depending on who - # calls this method. In any case, it is not needed, because we - # already know what class to restore: this one. - serialized.pop('__class__', None) - - def _produce_unserialized() -> Iterator[tuple[str, Any]]: - for option, value in serialized.items(): - key = option_to_key(option) - - _, _, _, _, metadata = container_field(cls, key) - - if metadata.unserialize_callback: - yield key, metadata.unserialize_callback(value) - - else: - yield key, value - - # Set attribute by adding it to __dict__ directly. Messing with setattr() - # might cause reuse of mutable values by other instances. - # obj.__dict__[keyname] = unserialize_callback(value) - - return cls(**dict(_produce_unserialized())) - - # ignore[misc,type-var]: mypy is correct here, method does return a - # TypeVar, but there is no way to deduce the actual type, because - # the method is static. That's on purpose, method tries to find the - # class to unserialize, therefore it's simply unknown. Returning Any - # would make mypy happy, but we do know the return value will be - # derived from SerializableContainer. We can mention that, and - # silence mypy about the missing actual type. - @staticmethod - def unserialize( - serialized: dict[str, Any], - logger: tmt.log.Logger - ) -> SerializableContainerDerivedType: # type: ignore[misc,type-var] - """ - Convert from a serialized form loaded from a file. - - Similar to :py:meth:`from_serialized`, but this method knows - nothing about container's class, and will locate the correct - module and class by inspecting serialized data. Discovered - class' :py:meth:`from_serialized` is then used to create the - container. - - Used to transform data read from a YAML file into original - containers when their classes are not know to the code. - Restoring such containers requires inspection of serialized data - and dynamic imports of modules as needed. - - See https://tmt.readthedocs.io/en/stable/code/classes.html#class-conversions - for more details. - - See :py:meth:`to_serialized` for its counterpart. - """ - - from tmt.plugins import import_member - - # Unpack class info, to get nicer variable names - if "__class__" not in serialized: - raise GeneralError( - "Failed to load saved state, probably because of old data format.\n" - "Use 'tmt clean runs' to clean up old runs.") - - klass_info = serialized.pop('__class__') - klass = import_member( - module=klass_info['module'], - member=klass_info['name'], - logger=logger)[1] - - # Stay away from classes that are not derived from this one, to - # honor promise given by return value annotation. - assert issubclass(klass, SerializableContainer) - - # Apparently, the issubclass() check above is not good enough for mypy. - return cast(SerializableContainerDerivedType, klass.from_serialized(serialized)) - - def markdown_to_html(filename: Path) -> str: """ Convert markdown to html @@ -5334,6 +4765,8 @@ def dataclass_normalize_field( and the return value is assigned to container field instead of ``value``. """ + from tmt.container import container_field + # Find out whether there's a normalization callback, and use it. Otherwise, # the raw value is simply used. value = raw_value @@ -5865,6 +5298,8 @@ def _load_keys( logger: tmt.log.Logger) -> None: """ Extract values for class-level attributes, and verify they match declared types. """ + from tmt.container import key_to_option + log_shift, log_level = 2, 4 debug_intro = functools.partial( @@ -5971,219 +5406,6 @@ def __init__( super().__init__(node=node, logger=logger, **kwargs) -@overload -def field( - *, - default: bool, - # Options - option: Optional[FieldCLIOption] = None, - is_flag: bool = True, - choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, - multiple: bool = False, - metavar: Optional[str] = None, - envvar: Optional[str] = None, - deprecated: Optional['tmt.options.Deprecated'] = None, - help: Optional[str] = None, - show_default: bool = False, - internal: bool = False, - # Input data normalization - not needed, the field is a boolean - # flag. - # normalize: Optional[NormalizeCallback[T]] = None - # Custom serialization - # serialize: Optional[SerializeCallback[bool]] = None, - # unserialize: Optional[UnserializeCallback[bool]] = None - # Custom exporter - # exporter: Optional[FieldExporter[T]] = None - ) -> bool: - pass - - -@overload -def field( - *, - default: T, - # Options - option: Optional[FieldCLIOption] = None, - is_flag: bool = False, - choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, - multiple: bool = False, - metavar: Optional[str] = None, - envvar: Optional[str] = None, - deprecated: Optional['tmt.options.Deprecated'] = None, - help: Optional[str] = None, - show_default: bool = False, - internal: bool = False, - # Input data normalization - normalize: Optional[NormalizeCallback[T]] = None, - # Custom serialization - serialize: Optional[SerializeCallback[T]] = None, - unserialize: Optional[UnserializeCallback[T]] = None, - # Custom exporter - exporter: Optional[FieldExporter[T]] = None - ) -> T: - pass - - -@overload -def field( - *, - default_factory: Callable[[], T], - # Options - option: Optional[FieldCLIOption] = None, - is_flag: bool = False, - choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, - multiple: bool = False, - metavar: Optional[str] = None, - envvar: Optional[str] = None, - deprecated: Optional['tmt.options.Deprecated'] = None, - help: Optional[str] = None, - show_default: bool = False, - internal: bool = False, - # Input data normalization - normalize: Optional[NormalizeCallback[T]] = None, - # Custom serialization - serialize: Optional[SerializeCallback[T]] = None, - unserialize: Optional[UnserializeCallback[T]] = None, - # Custom exporter - exporter: Optional[FieldExporter[T]] = None - ) -> T: - pass - - -@overload -def field( - *, - # Options - option: Optional[FieldCLIOption] = None, - is_flag: bool = False, - choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, - multiple: bool = False, - metavar: Optional[str] = None, - envvar: Optional[str] = None, - deprecated: Optional['tmt.options.Deprecated'] = None, - help: Optional[str] = None, - show_default: bool = False, - internal: bool = False, - # Input data normalization - normalize: Optional[NormalizeCallback[T]] = None, - # Custom serialization - serialize: Optional[SerializeCallback[T]] = None, - unserialize: Optional[UnserializeCallback[T]] = None, - # Custom exporter - exporter: Optional[FieldExporter[T]] = None - ) -> T: - pass - - -def field( - *, - default: Any = dataclasses.MISSING, - default_factory: Any = None, - # Options - option: Optional[FieldCLIOption] = None, - is_flag: bool = False, - choices: Union[None, Sequence[str], Callable[[], Sequence[str]]] = None, - multiple: bool = False, - metavar: Optional[str] = None, - envvar: Optional[str] = None, - deprecated: Optional['tmt.options.Deprecated'] = None, - help: Optional[str] = None, - show_default: bool = False, - internal: bool = False, - # Input data normalization - normalize: Optional[NormalizeCallback[T]] = None, - # Custom serialization - serialize: Optional[SerializeCallback[T]] = None, - unserialize: Optional[UnserializeCallback[T]] = None, - # Custom exporter - exporter: Optional[FieldExporter[T]] = None - ) -> Any: - """ - Define a :py:class:`DataContainer` field. - - Effectively a fancy wrapper over :py:func:`dataclasses.field`, tailored for - tmt code needs and simplification of various common tasks. - - :param default: if provided, this will be the default value for this field. - Passed directly to :py:func:`dataclass.field`. - It is an error to specify both ``default`` and ``default_factory``. - :param default_factory: if provided, it must be a zero-argument callable - that will be called when a default value is needed for this field. - Passed directly to :py:func:`dataclass.field`. - It is an error to specify both ``default`` and ``default_factory``. - :param option: one or more command-line option names. - Passed directly to :py:func:`click.option`. - :param is_flag: marks this option as a flag. - Passed directly to :py:func:`click.option`. - :param choices: if provided, the command-line option would accept only - the listed input values. - Passed to :py:func:`click.option` as a :py:class:`click.Choice` instance. - :param multiple: accept multiple arguments of the same name. - Passed directly to :py:func:`click.option`. - :param metavar: how the input value is represented in the help page. - Passed directly to :py:func:`click.option`. - :param envvar: environment variable used for this option. - Passed directly to :py:func:`click.option`. - :param deprecated: mark the option as deprecated - Provide an instance of Deprecated() with version in which the - option was obsoleted and an optional hint with the recommended - alternative. A warning message will be added to the option help. - :param help: the help string for the command-line option. Multiline strings - can be used, :py:func:`textwrap.dedent` is applied before passing - ``help`` to :py:func:`click.option`. - :param show_default: show default value - Passed directly to :py:func:`click.option`. - :param internal: if set, the field is treated as internal-only, and will not - appear when showing objects via ``show()`` method, or in export created - by :py:meth:`Core._export`. - :param normalize: a callback for normalizing the input value. Consumed by - :py:class:`NormalizeKeysMixin`. - :param serialize: a callback for custom serialization of the field value. - Consumed by :py:class:`SerializableKeysMixin`. - :param unserialize: a callback for custom unserialization of the field value. - Consumed by :py:class:`SerializableKeysMixin`. - :param exporter: a callback for custom export of the field value. - Consumed by :py:class:`tmt.export.Exportable`. - """ - - if option: - if is_flag is False and isinstance(default, bool): - raise GeneralError( - "Container field must be a flag to have boolean default value.") - - if is_flag is True and not isinstance(default, bool): - raise GeneralError( - "Container field must have a boolean default value when it is a flag.") - - metadata: FieldMetadata[T] = FieldMetadata( - internal=internal, - help=textwrap.dedent(help).strip() if help else None, - _metavar=metavar, - default=default, - default_factory=default_factory, - show_default=show_default, - is_flag=is_flag, - multiple=multiple, - _choices=choices, - envvar=envvar, - deprecated=deprecated, - cli_option=option, - normalize_callback=normalize, - serialize_callback=serialize, - unserialize_callback=unserialize, - export_callback=exporter) - - # ignore[call-overload]: returning "wrong" type on purpose. field() must be annotated - # as if returning the value of type matching the field declaration, and the original - # field() is called with wider argument types than expected, because we use our own - # overloading to narrow types *our* custom field() accepts. - return dataclasses.field( # type: ignore[call-overload] # noqa: TID251 - default=default, - default_factory=default_factory or dataclasses.MISSING, - metadata={'tmt': metadata} - ) - - def locate_key_origin(node: fmf.Tree, key: str) -> Optional[fmf.Tree]: """ Find an fmf node where the given key is defined.