Repository: elastic/rally

Issue:
Add new post_launch phase for plugin installers
Rally currently provides a `post_install` hook for plugin installers. For additional flexibility we should add a new phase `post_launch` which can also execute API requests after the cluster has been launched.

Discussion:

Patch:
diff --git a/esrally/mechanic/launcher.py b/esrally/mechanic/launcher.py
index 19e77cf8..a946e177 100644
--- a/esrally/mechanic/launcher.py
+++ b/esrally/mechanic/launcher.py
@@ -31,12 +31,35 @@ def wait_for_rest_layer(es, max_attempts=20):
 
 
 class ClusterLauncher:
-    def __init__(self, cfg, metrics_store, client_factory_class=client.EsClientFactory):
+    """
+    The cluster launcher performs cluster-wide tasks that need to be done in the startup / shutdown phase.
+
+    """
+    def __init__(self, cfg, metrics_store, on_post_launch=None, client_factory_class=client.EsClientFactory):
+        """
+
+        Creates a new ClusterLauncher.
+
+        :param cfg: The config object.
+        :param metrics_store: A metrics store that is configured to receive system metrics.
+        :param on_post_launch: An optional function that takes the Elasticsearch client as a parameter. It is invoked after the
+                               REST API is available.
+        :param client_factory_class: A factory class that can create an Elasticsearch client.
+        """
         self.cfg = cfg
         self.metrics_store = metrics_store
+        self.on_post_launch = on_post_launch
         self.client_factory = client_factory_class
 
     def start(self):
+        """
+        Performs final startup tasks.
+
+        Precondition: All cluster nodes have been started.
+        Postcondition: The cluster is ready to receive HTTP requests or a ``LaunchError`` is raised.
+
+        :return: A representation of the launched cluster.
+        """
         enabled_devices = self.cfg.opts("mechanic", "telemetry.devices")
         telemetry_params = self.cfg.opts("mechanic", "telemetry.params")
         hosts = self.cfg.opts("client", "hosts")
@@ -64,10 +87,16 @@ class ClusterLauncher:
             logger.error("REST API layer is not yet available. Forcefully terminating cluster.")
             self.stop(c)
             raise exceptions.LaunchError("Elasticsearch REST API layer is not available. Forcefully terminated cluster.")
-
+        if self.on_post_launch:
+            self.on_post_launch(es)
         return c
 
     def stop(self, c):
+        """
+        Performs cleanup tasks. This method should be called before nodes are shut down.
+
+        :param c: The cluster that is about to be stopped.
+        """
         c.telemetry.detach_from_cluster(c)
 
 
diff --git a/esrally/mechanic/mechanic.py b/esrally/mechanic/mechanic.py
index 08fdd111..aa0e6d4b 100644
--- a/esrally/mechanic/mechanic.py
+++ b/esrally/mechanic/mechanic.py
@@ -226,6 +226,7 @@ class MechanicActor(actor.RallyActor):
         self.race_control = None
         self.cluster_launcher = None
         self.cluster = None
+        self.plugins = None
 
     def receiveUnrecognizedMessage(self, msg, sender):
         logger.info("MechanicActor#receiveMessage unrecognized(msg = [%s] sender = [%s])" % (str(type(msg)), str(sender)))
@@ -256,6 +257,7 @@ class MechanicActor(actor.RallyActor):
         cls = metrics.metrics_store_class(self.cfg)
         self.metrics_store = cls(self.cfg)
         self.metrics_store.open(ctx=msg.open_metrics_context)
+        _, self.plugins = load_team(self.cfg, msg.external)
 
         # In our startup procedure we first create all mechanics. Only if this succeeds we'll continue.
         hosts = self.cfg.opts("client", "hosts")
@@ -345,7 +347,8 @@ class MechanicActor(actor.RallyActor):
         self.transition_when_all_children_responded(sender, msg, "cluster_stopping", "cluster_stopped", self.on_all_nodes_stopped)
 
     def on_all_nodes_started(self):
-        self.cluster_launcher = launcher.ClusterLauncher(self.cfg, self.metrics_store)
+        plugin_handler = PostLaunchPluginHandler(self.plugins)
+        self.cluster_launcher = launcher.ClusterLauncher(self.cfg, self.metrics_store, on_post_launch=plugin_handler)
         # Workaround because we could raise a LaunchError here and thespian will attempt to retry a failed message.
         # In that case, we will get a followup RallyAssertionError because on the second attempt, Rally will check
         # the status which is now "nodes_started" but we expected the status to be "nodes_starting" previously.
@@ -392,6 +395,21 @@ class MechanicActor(actor.RallyActor):
         # do not self-terminate, let the parent actor handle this
 
 
+class PostLaunchPluginHandler:
+    def __init__(self, plugins, hook_handler_class=team.PluginBootstrapHookHandler):
+        self.handlers = []
+        if plugins:
+            for plugin in plugins:
+                handler = hook_handler_class(plugin)
+                if handler.can_load():
+                    handler.load()
+                    self.handlers.append(handler)
+
+    def __call__(self, client):
+        for handler in self.handlers:
+            handler.invoke(team.PluginBootstrapPhase.post_launch.name, client=client)
+
+
 @thespian.actors.requireCapability('coordinator')
 class Dispatcher(thespian.actors.ActorTypeDispatcher):
     def __init__(self):
@@ -590,11 +608,7 @@ class NodeMechanicActor(actor.RallyActor):
 # Internal API (only used by the actor and for tests)
 #####################################################
 
-def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=False, build=False, distribution=False, external=False,
-           docker=False):
-    races_root = paths.races_root(cfg)
-    challenge_root_path = paths.race_root(cfg)
-    node_ids = cfg.opts("provisioning", "node.ids", mandatory=False)
+def load_team(cfg, external):
     # externally provisioned clusters do not support cars / plugins
     if external:
         car = None
@@ -603,6 +617,15 @@ def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=Fals
         team_path = team.team_path(cfg)
         car = team.load_car(team_path, cfg.opts("mechanic", "car.names"), cfg.opts("mechanic", "car.params"))
         plugins = team.load_plugins(team_path, cfg.opts("mechanic", "car.plugins"), cfg.opts("mechanic", "plugin.params"))
+    return car, plugins
+
+
+def create(cfg, metrics_store, all_node_ips, cluster_settings=None, sources=False, build=False, distribution=False, external=False,
+           docker=False):
+    races_root = paths.races_root(cfg)
+    challenge_root_path = paths.race_root(cfg)
+    node_ids = cfg.opts("provisioning", "node.ids", mandatory=False)
+    car, plugins = load_team(cfg, external)
 
     if sources or distribution:
         s = supplier.create(cfg, sources, distribution, build, challenge_root_path, plugins)
diff --git a/esrally/mechanic/provisioner.py b/esrally/mechanic/provisioner.py
index 4a1c3b38..2942caae 100644
--- a/esrally/mechanic/provisioner.py
+++ b/esrally/mechanic/provisioner.py
@@ -2,12 +2,12 @@ import os
 import glob
 import shutil
 import logging
-from enum import Enum
 
 import jinja2
 
 from esrally import exceptions
-from esrally.utils import io, console, process, modules, versions
+from esrally.mechanic import team
+from esrally.utils import io, console, process, versions
 
 logger = logging.getLogger("rally.provisioner")
 
@@ -102,21 +102,6 @@ def cleanup(preserve, install_dir, data_paths):
                 logger.exception("Could not delete [%s]. Skipping..." % install_dir)
 
 
-class ProvisioningPhase(Enum):
-    post_install = 10
-
-    @classmethod
-    def valid(cls, name):
-        for n in ProvisioningPhase.names():
-            if n == name:
-                return True
-        return False
-
-    @classmethod
-    def names(cls):
-        return [p.name for p in list(ProvisioningPhase)]
-
-
 def _apply_config(source_root_path, target_root_path, config_vars):
     for root, dirs, files in os.walk(source_root_path):
         env = jinja2.Environment(loader=jinja2.FileSystemLoader(root))
@@ -173,7 +158,7 @@ class BareProvisioner:
 
         for installer in self.plugin_installers:
             # Never let install hooks modify our original provisioner variables and just provide a copy!
-            installer.invoke_install_hook(ProvisioningPhase.post_install, provisioner_vars.copy())
+            installer.invoke_install_hook(team.PluginBootstrapPhase.post_install, provisioner_vars.copy())
 
         return NodeConfiguration(self.es_installer.car, self.es_installer.node_ip, self.es_installer.node_name,
                                  self.es_installer.node_root_dir, self.es_installer.es_home_path, self.es_installer.node_log_dir,
@@ -297,52 +282,8 @@ class ElasticsearchInstaller:
             return [os.path.join(self.es_home_path, "data")]
 
 
-class InstallHookHandler:
-    def __init__(self, plugin, loader_class=modules.ComponentLoader):
-        self.plugin = plugin
-        # Don't allow the loader to recurse. The subdirectories may contain Elasticsearch specific files which we do not want to add to
-        # Rally's Python load path. We may need to define a more advanced strategy in the future.
-        self.loader = loader_class(root_path=self.plugin.root_path, component_entry_point="plugin", recurse=False)
-        self.hooks = {}
-
-    def can_load(self):
-        return self.loader.can_load()
-
-    def load(self):
-        root_module = self.loader.load()
-        try:
-            # every module needs to have a register() method
-            root_module.register(self)
-        except exceptions.RallyError:
-            # just pass our own exceptions transparently.
-            raise
-        except BaseException:
-            msg = "Could not load install hooks in [%s]" % self.loader.root_path
-            logger.exception(msg)
-            raise exceptions.SystemSetupError(msg)
-
-    def register(self, phase, hook):
-        logger.info("Registering install hook [%s] for phase [%s] in plugin [%s]" % (hook.__name__, phase, self.plugin.name))
-        if not ProvisioningPhase.valid(phase):
-            raise exceptions.SystemSetupError("Provisioning phase [%s] is unknown. Valid phases are: %s." %
-                                              (phase, ProvisioningPhase.names()))
-        if phase not in self.hooks:
-            self.hooks[phase] = []
-        self.hooks[phase].append(hook)
-
-    def invoke(self, phase, variables):
-        if phase in self.hooks:
-            logger.info("Invoking phase [%s] for plugin [%s] in config [%s]" % (phase, self.plugin.name, self.plugin.config))
-            for hook in self.hooks[phase]:
-                logger.info("Invoking hook [%s]." % hook.__name__)
-                # hooks should only take keyword arguments to be forwards compatible with Rally!
-                hook(config_names=self.plugin.config, variables=variables)
-        else:
-            logger.debug("Plugin [%s] in config [%s] has no hook registered for phase [%s]." % (self.plugin.name, self.plugin.config, phase))
-
-
 class PluginInstaller:
-    def __init__(self, plugin, hook_handler_class=InstallHookHandler):
+    def __init__(self, plugin, hook_handler_class=team.PluginBootstrapHookHandler):
         self.plugin = plugin
         self.hook_handler = hook_handler_class(self.plugin)
         if self.hook_handler.can_load():
@@ -371,7 +312,7 @@ class PluginInstaller:
                                         (self.plugin_name, str(return_code)))
 
     def invoke_install_hook(self, phase, variables):
-        self.hook_handler.invoke(phase.name, variables)
+        self.hook_handler.invoke(phase.name, variables=variables)
 
     @property
     def variables(self):
diff --git a/esrally/mechanic/team.py b/esrally/mechanic/team.py
index e37e5c3a..37c4a518 100644
--- a/esrally/mechanic/team.py
+++ b/esrally/mechanic/team.py
@@ -1,11 +1,12 @@
 import os
 import logging
 import configparser
+from enum import Enum
 
 import tabulate
 
 from esrally import exceptions, PROGRAM_NAME
-from esrally.utils import console, repo, io
+from esrally.utils import console, repo, io, modules
 
 logger = logging.getLogger("rally.team")
 
@@ -341,3 +342,62 @@ class PluginDescriptor:
 
     def __eq__(self, other):
         return isinstance(other, type(self)) and (self.name, self.config, self.core_plugin) == (other.name, other.config, other.core_plugin)
+
+
+class PluginBootstrapPhase(Enum):
+    post_install = 10
+    post_launch = 20
+
+    @classmethod
+    def valid(cls, name):
+        for n in PluginBootstrapPhase.names():
+            if n == name:
+                return True
+        return False
+
+    @classmethod
+    def names(cls):
+        return [p.name for p in list(PluginBootstrapPhase)]
+
+
+class PluginBootstrapHookHandler:
+    def __init__(self, plugin, loader_class=modules.ComponentLoader):
+        self.plugin = plugin
+        # Don't allow the loader to recurse. The subdirectories may contain Elasticsearch specific files which we do not want to add to
+        # Rally's Python load path. We may need to define a more advanced strategy in the future.
+        self.loader = loader_class(root_path=self.plugin.root_path, component_entry_point="plugin", recurse=False)
+        self.hooks = {}
+
+    def can_load(self):
+        return self.loader.can_load()
+
+    def load(self):
+        root_module = self.loader.load()
+        try:
+            # every module needs to have a register() method
+            root_module.register(self)
+        except exceptions.RallyError:
+            # just pass our own exceptions transparently.
+            raise
+        except BaseException:
+            msg = "Could not load plugin bootstrap hooks in [{}]".format(self.loader.root_path)
+            logger.exception(msg)
+            raise exceptions.SystemSetupError(msg)
+
+    def register(self, phase, hook):
+        logger.info("Registering plugin bootstrap hook [%s] for phase [%s] in plugin [%s]", hook.__name__, phase, self.plugin.name)
+        if not PluginBootstrapPhase.valid(phase):
+            raise exceptions.SystemSetupError("Phase [{}] is unknown. Valid phases are: {}.".format(phase, PluginBootstrapPhase.names()))
+        if phase not in self.hooks:
+            self.hooks[phase] = []
+        self.hooks[phase].append(hook)
+
+    def invoke(self, phase, **kwargs):
+        if phase in self.hooks:
+            logger.info("Invoking phase [%s] for plugin [%s] in config [%s]", phase, self.plugin.name, self.plugin.config)
+            for hook in self.hooks[phase]:
+                logger.info("Invoking hook [%s].", hook.__name__)
+                # hooks should only take keyword arguments to be forwards compatible with Rally!
+                hook(config_names=self.plugin.config, **kwargs)
+        else:
+            logger.debug("Plugin [%s] in config [%s] has no hook registered for phase [%s].", self.plugin.name, self.plugin.config, phase)
\ No newline at end of file


Test patch:
diff --git a/tests/mechanic/launcher_test.py b/tests/mechanic/launcher_test.py
index 7e9d89bf..8110aae9 100644
--- a/tests/mechanic/launcher_test.py
+++ b/tests/mechanic/launcher_test.py
@@ -1,6 +1,6 @@
-from unittest import TestCase
+from unittest import TestCase, mock
 
-from esrally import config
+from esrally import config, exceptions
 from esrally.mechanic import launcher
 
 
@@ -11,14 +11,15 @@ class MockMetricsStore:
 
 class MockClientFactory:
     def __init__(self, hosts, client_options):
-        pass
+        self.client_options = client_options
 
     def create(self):
-        return MockClient()
+        return MockClient(self.client_options)
 
 
 class MockClient:
-    def __init__(self):
+    def __init__(self, client_options):
+        self.client_options = client_options
         self.cluster = SubClient({
             "cluster_name": "rally-benchmark-cluster",
             "nodes": {
@@ -54,8 +55,14 @@ class MockClient:
         }
 
     def info(self):
+        if self.client_options.get("raise-error-on-info", False):
+            import elasticsearch
+            raise elasticsearch.ConnectionError("Unittest error")
         return self._info
 
+    def search(self, *args, **kwargs):
+        return {}
+
 
 class SubClient:
     def __init__(self, info):
@@ -73,7 +80,7 @@ class ExternalLauncherTests(TestCase):
         cfg = config.Config()
         cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
         cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
-        cfg.add(config.Scope.application, "client", "options", [])
+        cfg.add(config.Scope.application, "client", "options", {})
 
         m = launcher.ExternalLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
         m.start()
@@ -85,10 +92,58 @@ class ExternalLauncherTests(TestCase):
         cfg = config.Config()
         cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
         cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
-        cfg.add(config.Scope.application, "client", "options", [])
+        cfg.add(config.Scope.application, "client", "options", {})
         cfg.add(config.Scope.application, "mechanic", "distribution.version", "2.3.3")
 
         m = launcher.ExternalLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
         m.start()
         # did not change user defined value
         self.assertEqual(cfg.opts("mechanic", "distribution.version"), "2.3.3")
+
+
+class ClusterLauncherTests(TestCase):
+    def test_launches_cluster_with_post_launch_handler(self):
+        on_post_launch = mock.Mock()
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        cfg.add(config.Scope.application, "client", "options", {})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(),
+                                                    on_post_launch=on_post_launch, client_factory_class=MockClientFactory)
+        cluster = cluster_launcher.start()
+
+        self.assertEqual(["10.0.0.10:9200", "10.0.0.11:9200"], cluster.hosts)
+        self.assertIsNotNone(cluster.telemetry)
+        on_post_launch.assert_called_once()
+
+    def test_launches_cluster_without_post_launch_handler(self):
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        cfg.add(config.Scope.application, "client", "options", {})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(), client_factory_class=MockClientFactory)
+        cluster = cluster_launcher.start()
+
+        self.assertEqual(["10.0.0.10:9200", "10.0.0.11:9200"], cluster.hosts)
+        self.assertIsNotNone(cluster.telemetry)
+
+    @mock.patch("time.sleep")
+    def test_error_on_cluster_launch(self, sleep):
+        on_post_launch = mock.Mock()
+        cfg = config.Config()
+        cfg.add(config.Scope.application, "client", "hosts", ["10.0.0.10:9200", "10.0.0.11:9200"])
+        # Simulate that the client will raise an error upon startup
+        cfg.add(config.Scope.application, "client", "options", {"raise-error-on-info": True})
+        cfg.add(config.Scope.application, "mechanic", "telemetry.devices", [])
+        cfg.add(config.Scope.application, "mechanic", "telemetry.params", {})
+
+        cluster_launcher = launcher.ClusterLauncher(cfg, MockMetricsStore(),
+                                                    on_post_launch=on_post_launch, client_factory_class=MockClientFactory)
+        with self.assertRaisesRegex(exceptions.LaunchError,
+                                    "Elasticsearch REST API layer is not available. Forcefully terminated cluster."):
+            cluster_launcher.start()
+        on_post_launch.assert_not_called()
\ No newline at end of file
diff --git a/tests/mechanic/provisioner_test.py b/tests/mechanic/provisioner_test.py
index 7c29095a..b06ead1f 100644
--- a/tests/mechanic/provisioner_test.py
+++ b/tests/mechanic/provisioner_test.py
@@ -439,60 +439,11 @@ class PluginInstallerTests(TestCase):
         installer = provisioner.PluginInstaller(plugin, hook_handler_class=PluginInstallerTests.NoopHookHandler)
 
         self.assertEqual(0, len(installer.hook_handler.hook_calls))
-        installer.invoke_install_hook(provisioner.ProvisioningPhase.post_install, {"foo": "bar"})
+        installer.invoke_install_hook(team.PluginBootstrapPhase.post_install, {"foo": "bar"})
         self.assertEqual(1, len(installer.hook_handler.hook_calls))
         self.assertEqual({"foo": "bar"}, installer.hook_handler.hook_calls["post_install"])
 
 
-class InstallHookHandlerTests(TestCase):
-    class UnitTestComponentLoader:
-        def __init__(self, root_path, component_entry_point, recurse):
-            self.root_path = root_path
-            self.component_entry_point = component_entry_point
-            self.recurse = recurse
-            self.registration_function = None
-
-        def load(self):
-            return self.registration_function
-
-    class UnitTestHook:
-        def __init__(self, phase="post_install"):
-            self.phase = phase
-            self.call_counter = 0
-
-        def post_install_hook(self, config_names, variables, **kwargs):
-            self.call_counter += variables["increment"]
-
-        def register(self, handler):
-            # we can register multiple hooks here
-            handler.register(self.phase, self.post_install_hook)
-            handler.register(self.phase, self.post_install_hook)
-
-    def test_loads_module(self):
-        plugin = team.PluginDescriptor("unittest-plugin")
-        hook = InstallHookHandlerTests.UnitTestHook()
-        handler = provisioner.InstallHookHandler(plugin, loader_class=InstallHookHandlerTests.UnitTestComponentLoader)
-
-        handler.loader.registration_function = hook
-        handler.load()
-
-        handler.invoke("post_install", {"increment": 4})
-
-        # we registered our hook twice. Check that it has been called twice.
-        self.assertEqual(hook.call_counter, 2 * 4)
-
-    def test_cannot_register_for_unknown_phase(self):
-        plugin = team.PluginDescriptor("unittest-plugin")
-        hook = InstallHookHandlerTests.UnitTestHook(phase="this_is_an_unknown_install_phase")
-        handler = provisioner.InstallHookHandler(plugin, loader_class=InstallHookHandlerTests.UnitTestComponentLoader)
-
-        handler.loader.registration_function = hook
-        with self.assertRaises(exceptions.SystemSetupError) as ctx:
-            handler.load()
-        self.assertEqual("Provisioning phase [this_is_an_unknown_install_phase] is unknown. Valid phases are: ['post_install'].",
-                         ctx.exception.args[0])
-
-
 class DockerProvisionerTests(TestCase):
     @mock.patch("esrally.utils.sysstats.total_memory")
     @mock.patch("uuid.uuid4")
diff --git a/tests/mechanic/team_test.py b/tests/mechanic/team_test.py
index 37f80635..b8a94ecb 100644
--- a/tests/mechanic/team_test.py
+++ b/tests/mechanic/team_test.py
@@ -118,3 +118,52 @@ class PluginLoaderTests(TestCase):
             "var": "0",
             "hello": "true"
         }, plugin.variables)
+
+
+class PluginBootstrapHookHandlerTests(TestCase):
+    class UnitTestComponentLoader:
+        def __init__(self, root_path, component_entry_point, recurse):
+            self.root_path = root_path
+            self.component_entry_point = component_entry_point
+            self.recurse = recurse
+            self.registration_function = None
+
+        def load(self):
+            return self.registration_function
+
+    class UnitTestHook:
+        def __init__(self, phase="post_install"):
+            self.phase = phase
+            self.call_counter = 0
+
+        def post_install_hook(self, config_names, variables, **kwargs):
+            self.call_counter += variables["increment"]
+
+        def register(self, handler):
+            # we can register multiple hooks here
+            handler.register(self.phase, self.post_install_hook)
+            handler.register(self.phase, self.post_install_hook)
+
+    def test_loads_module(self):
+        plugin = team.PluginDescriptor("unittest-plugin")
+        hook = PluginBootstrapHookHandlerTests.UnitTestHook()
+        handler = team.PluginBootstrapHookHandler(plugin, loader_class=PluginBootstrapHookHandlerTests.UnitTestComponentLoader)
+
+        handler.loader.registration_function = hook
+        handler.load()
+
+        handler.invoke("post_install", variables={"increment": 4})
+
+        # we registered our hook twice. Check that it has been called twice.
+        self.assertEqual(hook.call_counter, 2 * 4)
+
+    def test_cannot_register_for_unknown_phase(self):
+        plugin = team.PluginDescriptor("unittest-plugin")
+        hook = PluginBootstrapHookHandlerTests.UnitTestHook(phase="this_is_an_unknown_install_phase")
+        handler = team.PluginBootstrapHookHandler(plugin, loader_class=PluginBootstrapHookHandlerTests.UnitTestComponentLoader)
+
+        handler.loader.registration_function = hook
+        with self.assertRaises(exceptions.SystemSetupError) as ctx:
+            handler.load()
+        self.assertEqual("Phase [this_is_an_unknown_install_phase] is unknown. Valid phases are: ['post_install', 'post_launch'].",
+                         ctx.exception.args[0])