Define "plugin registry" class

The class is fairly trivial, but draws a clear distinction between
an arbitrary dictionary and a "plugin registry" object. New types
of plugins are heading our way, therefore making boundaries more
visible, to clear up responsibilities.

tmt/cli.py

@@ -692,7 +692,7 @@ def tests_import(
         tmt.convert.adjust_runtest(path / 'runtest.sh')
 
 
-_test_export_formats = list(tmt.Test.get_export_plugin_registry().keys())
+_test_export_formats = list(tmt.Test.get_export_plugin_registry().iter_plugin_ids())
 _test_export_default = 'yaml'
 
 
@@ -959,7 +959,7 @@ def plans_create(
     tmt.Plan.create(name, template, context.obj.tree.root, force)
 
 
-_plan_export_formats = list(tmt.Plan.get_export_plugin_registry().keys())
+_plan_export_formats = list(tmt.Plan.get_export_plugin_registry().iter_plugin_ids())
 _plan_export_default = 'yaml'
 
 
@@ -1208,7 +1208,7 @@ def headfoot(text: str) -> None:
     echo()
 
 
-_story_export_formats = list(tmt.Story.get_export_plugin_registry().keys())
+_story_export_formats = list(tmt.Story.get_export_plugin_registry().iter_plugin_ids())
 _story_export_default = 'yaml'
 
 

tmt/export/__init__.py

@@ -26,7 +26,9 @@
 from click import echo, style
 
 import tmt
+import tmt.log
 import tmt.utils
+from tmt.plugins import PluginRegistry
 from tmt.utils import Path
 
 if TYPE_CHECKING:
@@ -69,20 +71,21 @@ class Exportable(Generic[ExportableT], tmt.utils._CommonBase):
 
     # Declare export plugin registry as a class variable, but do not initialize it. If initialized
     # here, the mapping would be shared by all classes, which is not a desirable attribute.
-    # Instead, mapping will be created by provides_*_export decorators.
-    _export_plugin_registry: ClassVar[Dict[str, ExportClass]]
+    # Instead, mapping will be created by get_export_plugin_registry() method when called for the
+    # first time.
+    _export_plugin_registry: ClassVar[PluginRegistry[ExportClass]]
 
     # Keep this method around, to correctly support Python's method resolution order.
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
 
     # Cannot use @property as this must remain classmethod
     @classmethod
-    def get_export_plugin_registry(cls) -> Dict[str, ExportClass]:
+    def get_export_plugin_registry(cls) -> PluginRegistry[ExportClass]:
         """ Return - or initialize - export plugin registry """
 
         if not hasattr(cls, '_export_plugin_registry'):
-            cls._export_plugin_registry = {}
+            cls._export_plugin_registry = PluginRegistry()
 
         return cls._export_plugin_registry
 
@@ -95,7 +98,10 @@ def provides_export(cls, format: str) -> Callable[[ExportClass], ExportClass]:
         """
 
         def _provides_export(export_cls: ExportClass) -> ExportClass:
-            cls.get_export_plugin_registry()[format] = export_cls
+            cls.get_export_plugin_registry().register_plugin(
+                plugin_id=format,
+                plugin=export_cls,
+                logger=tmt.log.Logger.get_bootstrap_logger())
 
             return export_cls
 
@@ -116,7 +122,7 @@ def _get_exporter(cls, format: str) -> Exporter:
             format.
         """
 
-        exporter_class = cls.get_export_plugin_registry().get(format, None)
+        exporter_class = cls.get_export_plugin_registry().get_plugin(format)
 
         if exporter_class is None:
             raise tmt.utils.GeneralError(

tmt/plugins/__init__.py

@@ -6,14 +6,16 @@
 import os
 import pkgutil
 import sys
-from typing import Any, Generator, List, Optional, Tuple
+from typing import (Any, Dict, Generator, Generic, List, Optional, Tuple,
+                    TypeVar)
 
 if sys.version_info < (3, 9):
     from importlib_metadata import entry_points
 else:
     from importlib.metadata import entry_points
 
 import tmt
+import tmt.utils
 from tmt.log import Logger
 from tmt.steps import STEPS
 from tmt.utils import Path
@@ -219,3 +221,60 @@ def explore_export_package(logger: Logger) -> None:
     """ Import all plugins bundled into tmt.export package """
 
     _explore_package('tmt.export', _TMT_ROOT / 'export', logger.descend())
+
+
+RegisterableT = TypeVar('RegisterableT')
+
+
+class PluginRegistry(Generic[RegisterableT]):
+    """
+    A container for plugins of shared purpose.
+
+    A fancy wrapper for a dictionary at its core, but allows for nicer
+    annotations and more visible semantics.
+    """
+
+    _plugins: Dict[str, RegisterableT]
+
+    def __init__(self) -> None:
+        self._plugins = {}
+
+    def register_plugin(
+            self,
+            *,
+            plugin_id: str,
+            plugin: RegisterableT,
+            raise_on_conflict: bool = True,
+            logger: Logger) -> None:
+        if plugin_id in self._plugins and raise_on_conflict:
+            # TODO: would be raising an exception better? Probably, but since
+            # plugin discovery happens in import time, it's very hard to manage
+            # it. For now, report a warning, but do not raise an exception yet.
+            logger.warn(
+                f"Registering plugin '{plugin.__module__}' collides"
+                f" with an already registered id '{plugin_id}'"
+                f" of plugin '{self._plugins[plugin_id]}'.")
+
+            # raise tmt.utils.GeneralError(
+            #     f"Registering plugin '{plugin.__module__}' collides"
+            #     f" with an already registered id '{plugin_id}'"
+            #     f" of plugin '{self._plugins[plugin_id]}'.")
+
+        self._plugins[plugin_id] = plugin
+
+        logger.debug(f'registered plugin "{plugin}" with id "{plugin_id}"')
+
+    def get_plugin(self, plugin_id: str) -> Optional[RegisterableT]:
+        """
+        Find a plugin by its id.
+
+        :returns: plugin or ``None`` if no such id has been registered.
+        """
+
+        return self._plugins.get(plugin_id, None)
+
+    def iter_plugin_ids(self) -> Generator[str, None, None]:
+        yield from self._plugins.keys()
+
+    def iter_plugins(self) -> Generator[RegisterableT, None, None]:
+        yield from self._plugins.values()

tmt/steps/__init__.py

@@ -31,6 +31,7 @@
 
     import tmt.base
     import tmt.cli
+    import tmt.plugins
     import tmt.steps.discover
     import tmt.steps.execute
     from tmt.base import Plan
@@ -653,7 +654,11 @@ def _method(cls: PluginClass) -> PluginClass:
         plugin_method = Method(name, class_=cls, doc=doc, order=order)
 
         # FIXME: make sure cls.__bases__[0] is really BasePlugin class
-        cast('BasePlugin', cls.__bases__[0])._supported_methods.append(plugin_method)
+        cast('BasePlugin', cls.__bases__[0])._supported_methods \
+            .register_plugin(
+                plugin_id=name,
+                plugin=plugin_method,
+                logger=tmt.log.Logger.get_bootstrap_logger())
 
         return cls
 
@@ -671,8 +676,14 @@ class BasePlugin(Phase):
     # except for provision (virtual) and report (display)
     how: str = 'shell'
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    #
+    # The field is declared here, in a base class of all plugin classes, and
+    # each step-specific base plugin class assignes it a value as a class-level
+    # attribute. This guarantees steps would not share a registry instance while
+    # the declaration below make the name and type visible across all
+    # subclasses.
+    _supported_methods: 'tmt.plugins.PluginRegistry[Method]'
 
     _data_class: Type[StepData] = StepData
     data: StepData
@@ -791,7 +802,7 @@ def command(cls) -> click.Command:
     @classmethod
     def methods(cls) -> List[Method]:
         """ Return all supported methods ordered by priority """
-        return sorted(cls._supported_methods, key=lambda method: method.order)
+        return sorted(cls._supported_methods.iter_plugins(), key=lambda method: method.order)
 
     @classmethod
     def delegate(

tmt/steps/discover/__init__.py

@@ -15,6 +15,7 @@
 import tmt.base
 import tmt.steps
 import tmt.utils
+from tmt.plugins import PluginRegistry
 from tmt.steps import Action
 from tmt.utils import Command, GeneralError, Path, flatten
 
@@ -42,8 +43,8 @@ class DiscoverPlugin(tmt.steps.GuestlessPlugin):
 
     _data_class = DiscoverStepData
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[tmt.steps.Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[tmt.steps.Method] = PluginRegistry()
 
     @classmethod
     def base_command(

tmt/steps/execute/__init__.py

@@ -13,6 +13,7 @@
 import tmt.base
 import tmt.steps
 import tmt.utils
+from tmt.plugins import PluginRegistry
 from tmt.queue import TaskOutcome
 from tmt.result import Result, ResultGuestData, ResultOutcome
 from tmt.steps import Action, PhaseQueue, QueuedPhase, Step, StepData
@@ -120,8 +121,8 @@ class ExecutePlugin(tmt.steps.Plugin):
 
     _data_class = ExecuteStepData
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[tmt.steps.Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[tmt.steps.Method] = PluginRegistry()
 
     # Internal executor is the default implementation
     how = 'tmt'

tmt/steps/finish/__init__.py

@@ -7,6 +7,7 @@
 
 import tmt
 import tmt.steps
+from tmt.plugins import PluginRegistry
 from tmt.steps import (Action, Method, PhaseQueue, PullTask, QueuedPhase,
                        TaskOutcome, sync_with_guests)
 from tmt.steps.provision import Guest
@@ -25,8 +26,8 @@ class FinishPlugin(tmt.steps.Plugin):
 
     _data_class = FinishStepData
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[Method] = PluginRegistry()
 
     @classmethod
     def base_command(

tmt/steps/prepare/__init__.py

@@ -13,6 +13,7 @@
 import tmt.steps
 import tmt.steps.provision
 import tmt.utils
+from tmt.plugins import PluginRegistry
 from tmt.queue import TaskOutcome
 from tmt.steps import (Action, PhaseQueue, PullTask, PushTask, QueuedPhase,
                        sync_with_guests)
@@ -44,8 +45,8 @@ class PreparePlugin(tmt.steps.Plugin):
 
     _data_class = PrepareStepData
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[tmt.steps.Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[tmt.steps.Method] = PluginRegistry()
 
     @classmethod
     def base_command(

tmt/steps/provision/__init__.py

@@ -22,6 +22,7 @@
 import tmt.plugins
 import tmt.steps
 import tmt.utils
+from tmt.plugins import PluginRegistry
 from tmt.steps import Action
 from tmt.utils import BaseLoggerFnType, Command, Path, ShellScript, field
 
@@ -1334,8 +1335,8 @@ class ProvisionPlugin(tmt.steps.GuestlessPlugin):
     # Default implementation for provision is a virtual machine
     how = 'virtual'
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[tmt.steps.Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[tmt.steps.Method] = PluginRegistry()
 
     # TODO: Generics would provide a better type, https://github.com/teemtee/tmt/issues/1437
     _guest: Optional[Guest] = None

tmt/steps/report/__init__.py

@@ -1,10 +1,12 @@
 import dataclasses
-from typing import TYPE_CHECKING, Any, List, Optional, Type, Union, cast
+from typing import TYPE_CHECKING, Any, Optional, Type, Union, cast
 
 import click
 
 import tmt
+import tmt.plugins
 import tmt.steps
+from tmt.plugins import PluginRegistry
 from tmt.steps import Action
 
 if TYPE_CHECKING:
@@ -24,8 +26,8 @@ class ReportPlugin(tmt.steps.GuestlessPlugin):
     # Default implementation for report is display
     how = 'display'
 
-    # List of all supported methods aggregated from all plugins of the same step.
-    _supported_methods: List[tmt.steps.Method] = []
+    # Methods ("how: ..." implementations) registered for the same step.
+    _supported_methods: PluginRegistry[tmt.steps.Method] = PluginRegistry()
 
     @classmethod
     def base_command(