PMGR: Detect if plugin needs printer reconnect
This commit is contained in:
Gina Häußge
parent
188b59a319
commit
f6c3e5991d
7 changed files with 268 additions and 39 deletions
|
|
@ -54,6 +54,12 @@ under Settings > Plugin Manager, or by directly editing ``config.yaml``:
|
|||
# Time to live for the repository cache
|
||||
repository_ttl: 1440
|
||||
|
||||
# The URL of the plugin notices feed to use
|
||||
notices: http://plugins.octoprint.org/notices.json
|
||||
|
||||
# Time to live for the notices feed cache
|
||||
notices_ttl: 360
|
||||
|
||||
# Additional arguments to use with pip. Defaults to unset,
|
||||
# you normally shouldn't need to modify this
|
||||
pip_args: --some --additional --pip --arguments
|
||||
|
|
@ -76,6 +82,36 @@ under Settings > Plugin Manager, or by directly editing ``config.yaml``:
|
|||
- hidden
|
||||
- plugins
|
||||
|
||||
.. _sec-bundledplugins-pluginmanager-hooks:
|
||||
|
||||
Hooks
|
||||
-----
|
||||
|
||||
.. _sec-bundledplugins-pluginmanager-hooks-reconnect_hooks:
|
||||
|
||||
octoprint.plugin.pluginmanager.reconnect_hooks
|
||||
++++++++++++++++++++++++++++++++++++++++++++++
|
||||
|
||||
.. py:function:: reconnect_hooks_hook(*args, **kwargs)
|
||||
|
||||
Returns additional hooks defined by the plugin for which the plugin manager
|
||||
should display the "You should reconnect to your printer" message on plugin
|
||||
install/uninstall/enabling/disabling.
|
||||
|
||||
Handlers should return a Python list containing the affected hook names.
|
||||
|
||||
**Example**
|
||||
|
||||
.. code-block:: python
|
||||
|
||||
def reconnect_hooks_hook(*args, **kwargs):
|
||||
return ["octoprint.plugin.exampleplugin.some_custom_hook",
|
||||
"octoprint.plugin.exampleplugin.some_other_custom_hook"]
|
||||
|
||||
__plugin_hooks__ = {
|
||||
"octoprint.plugin.pluginmanager.reconnect_hooks": reconnect_hooks_hook
|
||||
}
|
||||
|
||||
.. _sec-bundledplugins-pluginmanager-sourcecode:
|
||||
|
||||
Source Code
|
||||
|
|
|
|||
|
|
@ -104,7 +104,8 @@ def plugin_manager(init=False, plugin_folders=None, plugin_types=None, plugin_en
|
|||
UiPlugin]
|
||||
|
||||
if plugin_restart_needing_hooks is None:
|
||||
plugin_restart_needing_hooks = ["octoprint.server.http"]
|
||||
plugin_restart_needing_hooks = ["octoprint.server.http.*",
|
||||
"octoprint.printer.factory"]
|
||||
|
||||
if plugin_obsolete_hooks is None:
|
||||
plugin_obsolete_hooks = ["octoprint.comm.protocol.gcode"]
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ import os
|
|||
import imp
|
||||
from collections import defaultdict, namedtuple, OrderedDict
|
||||
import logging
|
||||
import fnmatch
|
||||
|
||||
import pkg_resources
|
||||
import pkginfo
|
||||
|
|
@ -946,45 +947,107 @@ class PluginManager(object):
|
|||
return plugin.needs_restart or self.has_restart_needing_implementation(plugin) or self.has_restart_needing_hooks(plugin)
|
||||
|
||||
def has_restart_needing_implementation(self, plugin):
|
||||
if not plugin.implementation:
|
||||
return False
|
||||
|
||||
return isinstance(plugin.implementation, RestartNeedingPlugin)
|
||||
return self.has_any_of_mixins(plugin, RestartNeedingPlugin)
|
||||
|
||||
def has_restart_needing_hooks(self, plugin):
|
||||
if not plugin.hooks:
|
||||
return False
|
||||
|
||||
hooks = plugin.hooks.keys()
|
||||
for hook in hooks:
|
||||
if self.is_restart_needing_hook(hook):
|
||||
return True
|
||||
return False
|
||||
return self.has_any_of_hooks(plugin, self.plugin_restart_needing_hooks)
|
||||
|
||||
def has_obsolete_hooks(self, plugin):
|
||||
if not plugin.hooks:
|
||||
return False
|
||||
|
||||
hooks = plugin.hooks.keys()
|
||||
for hook in hooks:
|
||||
if self.is_obsolete_hook(hook):
|
||||
return True
|
||||
return False
|
||||
return self.has_any_of_hooks(plugin, self.plugin_obsolete_hooks)
|
||||
|
||||
def is_restart_needing_hook(self, hook):
|
||||
if self.plugin_restart_needing_hooks is None:
|
||||
return False
|
||||
|
||||
for h in self.plugin_restart_needing_hooks:
|
||||
if hook.startswith(h):
|
||||
return True
|
||||
|
||||
return False
|
||||
return self.hook_matches_hooks(hook, self.plugin_restart_needing_hooks)
|
||||
|
||||
def is_obsolete_hook(self, hook):
|
||||
if self.plugin_obsolete_hooks is None:
|
||||
return self.hook_matches_hooks(hook, self.plugin_obsolete_hooks)
|
||||
|
||||
@staticmethod
|
||||
def has_any_of_hooks(plugin, *hooks):
|
||||
"""
|
||||
Tests if the ``plugin`` contains any of the provided ``hooks``.
|
||||
|
||||
Uses :func:`octoprint.plugin.core.PluginManager.hook_matches_hooks`.
|
||||
|
||||
Args:
|
||||
plugin: plugin to test hooks for
|
||||
*hooks: hooks to test against
|
||||
|
||||
Returns:
|
||||
(bool): True if any of the plugin's hooks match the provided hooks,
|
||||
False otherwise.
|
||||
"""
|
||||
|
||||
if hooks and len(hooks) == 1 and isinstance(hooks[0], (list, tuple)):
|
||||
hooks = hooks[0]
|
||||
|
||||
hooks = filter(lambda hook: hook is not None, hooks)
|
||||
if not hooks:
|
||||
return False
|
||||
return hook in self.plugin_obsolete_hooks
|
||||
if not plugin or not plugin.hooks:
|
||||
return False
|
||||
|
||||
plugin_hooks = plugin.hooks.keys()
|
||||
|
||||
return any(map(lambda hook: PluginManager.hook_matches_hooks(hook, *hooks),
|
||||
plugin_hooks))
|
||||
|
||||
@staticmethod
|
||||
def hook_matches_hooks(hook, *hooks):
|
||||
"""
|
||||
Tests if ``hook`` matches any of the provided ``hooks`` to test for.
|
||||
|
||||
``hook`` is expected to be an exact hook name.
|
||||
|
||||
``hooks`` is expected to be a list containing one or more hook names or
|
||||
patterns. That can be either an exact hook name or an
|
||||
:func:`fnmatch.fnmatch` pattern.
|
||||
|
||||
Args:
|
||||
hook: the hook to test
|
||||
hooks: the hook name patterns to test against
|
||||
|
||||
Returns:
|
||||
(bool): True if the ``hook`` matches any of the ``hooks``, False otherwise.
|
||||
|
||||
"""
|
||||
|
||||
if hooks and len(hooks) == 1 and isinstance(hooks[0], (list, tuple)):
|
||||
hooks = hooks[0]
|
||||
|
||||
hooks = filter(lambda hook: hook is not None, hooks)
|
||||
if not hooks:
|
||||
return False
|
||||
if not hook:
|
||||
return False
|
||||
|
||||
return any(map(lambda h: fnmatch.fnmatch(hook, h),
|
||||
hooks))
|
||||
|
||||
@staticmethod
|
||||
def has_any_of_mixins(plugin, *mixins):
|
||||
"""
|
||||
Tests if the ``plugin`` has an implementation implementing any
|
||||
of the provided ``mixins``.
|
||||
|
||||
Args:
|
||||
plugin: plugin for which to check the implementation
|
||||
*mixins: mixins to test against
|
||||
|
||||
Returns:
|
||||
(bool): True if the plugin's implementation implements any of the
|
||||
provided mixins, False otherwise.
|
||||
"""
|
||||
|
||||
if mixins and len(mixins) == 1 and isinstance(mixins[0], (list, tuple)):
|
||||
mixins = mixins[0]
|
||||
|
||||
mixins = filter(lambda mixin: mixin is not None, mixins)
|
||||
if not mixins:
|
||||
return False
|
||||
if not plugin or not plugin.implementation:
|
||||
return False
|
||||
|
||||
return isinstance(plugin.implementation, tuple(mixins))
|
||||
|
||||
def initialize_implementations(self, additional_injects=None, additional_inject_factories=None, additional_pre_inits=None, additional_post_inits=None):
|
||||
for name, plugin in self.enabled_plugins.items():
|
||||
|
|
|
|||
|
|
@ -743,7 +743,7 @@ class UiPlugin(OctoPrintPlugin, SortablePlugin):
|
|||
OctoPrint's version, current ``UI_API_KEY``, tracked file paths and ``LastModified`` value).
|
||||
|
||||
Returns:
|
||||
basestring: An alternatively calculated ETag value. Ignored if ``None`` is returned (default).
|
||||
str: An alternatively calculated ETag value. Ignored if ``None`` is returned (default).
|
||||
"""
|
||||
return None
|
||||
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ __author__ = "Gina Häußge <osd@foosel.net>"
|
|||
__license__ = 'GNU Affero General Public License http://www.gnu.org/licenses/agpl.html'
|
||||
__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"
|
||||
|
||||
from past.builtins import basestring
|
||||
|
||||
import octoprint.plugin
|
||||
import octoprint.plugin.core
|
||||
|
|
@ -44,7 +45,9 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
macos=["darwin"],
|
||||
freebsd=lambda x: x.startswith("freebsd"))
|
||||
|
||||
pip_inapplicable_arguments = dict(uninstall=["--user"])
|
||||
PIP_INAPPLICABLE_ARGUMENTS = dict(uninstall=["--user"])
|
||||
|
||||
RECONNECT_HOOKS = ["octoprint.comm.protocol.*",]
|
||||
|
||||
def __init__(self):
|
||||
self._pending_enable = set()
|
||||
|
|
@ -415,6 +418,7 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
source_type=source_type,
|
||||
needs_restart=True,
|
||||
needs_refresh=True,
|
||||
needs_reconnect=True,
|
||||
was_reinstalled=False,
|
||||
plugin="unknown")
|
||||
self._send_result_notification("install", result)
|
||||
|
|
@ -426,6 +430,7 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
or reinstall is not None
|
||||
needs_refresh = new_plugin.implementation \
|
||||
and isinstance(new_plugin.implementation, octoprint.plugin.ReloadNeedingPlugin)
|
||||
needs_reconnect = self._plugin_manager.has_any_of_hooks(new_plugin, self._reconnect_hooks) and self._printer.is_operational()
|
||||
|
||||
is_reinstall = self._plugin_manager.is_plugin_marked(new_plugin.key, "uninstalled")
|
||||
self._plugin_manager.mark_plugin(new_plugin.key,
|
||||
|
|
@ -440,6 +445,7 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
source_type=source_type,
|
||||
needs_restart=needs_restart,
|
||||
needs_refresh=needs_refresh,
|
||||
needs_reconnect=needs_reconnect,
|
||||
was_reinstalled=new_plugin.key in all_plugins_before or reinstall is not None,
|
||||
plugin=self._to_external_plugin(new_plugin))
|
||||
self._send_result_notification("install", result)
|
||||
|
|
@ -496,6 +502,7 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
|
||||
needs_restart = self._plugin_manager.is_restart_needing_plugin(plugin)
|
||||
needs_refresh = plugin.implementation and isinstance(plugin.implementation, octoprint.plugin.ReloadNeedingPlugin)
|
||||
needs_reconnect = self._plugin_manager.has_any_of_hooks(plugin, self._reconnect_hooks) and self._printer.is_operational()
|
||||
|
||||
was_pending_install = self._plugin_manager.is_plugin_marked(plugin.key, "installed")
|
||||
self._plugin_manager.mark_plugin(plugin.key,
|
||||
|
|
@ -521,7 +528,11 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
|
||||
self._plugin_manager.reload_plugins()
|
||||
|
||||
result = dict(result=True, needs_restart=needs_restart, needs_refresh=needs_refresh, plugin=self._to_external_plugin(plugin))
|
||||
result = dict(result=True,
|
||||
needs_restart=needs_restart,
|
||||
needs_refresh=needs_refresh,
|
||||
needs_reconnect=needs_reconnect,
|
||||
plugin=self._to_external_plugin(plugin))
|
||||
self._send_result_notification("uninstall", result)
|
||||
return jsonify(result)
|
||||
|
||||
|
|
@ -531,11 +542,13 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
|
||||
needs_restart = self._plugin_manager.is_restart_needing_plugin(plugin)
|
||||
needs_refresh = plugin.implementation and isinstance(plugin.implementation, octoprint.plugin.ReloadNeedingPlugin)
|
||||
needs_reconnect = self._plugin_manager.has_any_of_hooks(plugin, self._reconnect_hooks) and self._printer.is_operational()
|
||||
|
||||
pending = ((command == "disable" and plugin.key in self._pending_enable) or (command == "enable" and plugin.key in self._pending_disable))
|
||||
safe_mode_victim = getattr(plugin, "safe_mode_victim", False)
|
||||
needs_restart_api = (needs_restart or safe_mode_victim) and not pending
|
||||
needs_refresh_api = needs_refresh and not pending
|
||||
needs_reconnect_api = needs_reconnect and not pending
|
||||
|
||||
try:
|
||||
if command == "disable":
|
||||
|
|
@ -549,11 +562,13 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
result = dict(result=True,
|
||||
needs_restart=True,
|
||||
needs_refresh=True,
|
||||
needs_reconnect=True,
|
||||
plugin=self._to_external_plugin(plugin))
|
||||
else:
|
||||
result = dict(result=True,
|
||||
needs_restart=needs_restart_api,
|
||||
needs_refresh=needs_refresh_api,
|
||||
needs_reconnect=needs_reconnect_api,
|
||||
plugin=self._to_external_plugin(plugin))
|
||||
|
||||
self._send_result_notification(command, result)
|
||||
|
|
@ -605,7 +620,7 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
|
||||
if additional_args is not None:
|
||||
|
||||
inapplicable_arguments = self.__class__.pip_inapplicable_arguments.get(args[0], list())
|
||||
inapplicable_arguments = self.__class__.PIP_INAPPLICABLE_ARGUMENTS.get(args[0], list())
|
||||
for inapplicable_argument in inapplicable_arguments:
|
||||
additional_args = re.sub("(^|\s)" + re.escape(inapplicable_argument) + "\\b", "", additional_args)
|
||||
|
||||
|
|
@ -881,6 +896,22 @@ class PluginManagerPlugin(octoprint.plugin.SimpleApiPlugin,
|
|||
octoprint_version = pkg_resources.parse_version(octoprint_version.base_version)
|
||||
return octoprint_version
|
||||
|
||||
@property
|
||||
def _reconnect_hooks(self):
|
||||
reconnect_hooks = self.__class__.RECONNECT_HOOKS
|
||||
|
||||
reconnect_hook_provider_hooks = self._plugin_manager.get_hooks("octoprint.plugin.pluginmanager.reconnect_hooks")
|
||||
for name, hook in reconnect_hook_provider_hooks.items():
|
||||
try:
|
||||
result = hook()
|
||||
if isinstance(result, (list, tuple)):
|
||||
reconnect_hooks.extend(filter(lambda x: isinstance(x, basestring), result))
|
||||
except:
|
||||
self._logger.exception("Error while retrieving additional hooks for which a "
|
||||
"reconnect is required from plugin {name}".format(**locals()))
|
||||
|
||||
return reconnect_hooks
|
||||
|
||||
def _get_plugins(self):
|
||||
plugins = self._plugin_manager.plugins
|
||||
|
||||
|
|
|
|||
|
|
@ -663,7 +663,7 @@ $(function() {
|
|||
return self.isCompatible(data) ? (self.installed(data) ? gettext("Reinstall") : gettext("Install")) : (data.disabled ? gettext("Disabled") : gettext("Incompatible"));
|
||||
};
|
||||
|
||||
self._displayNotification = function(response, titleSuccess, textSuccess, textRestart, textReload, titleError, textError) {
|
||||
self._displayNotification = function(response, titleSuccess, textSuccess, textRestart, textReload, textReconnect, titleError, textError) {
|
||||
var notification;
|
||||
|
||||
var beforeClose = function(notification) {
|
||||
|
|
@ -747,6 +747,15 @@ $(function() {
|
|||
},
|
||||
hide: false
|
||||
})
|
||||
} else if (response.needs_reconnect) {
|
||||
notification = new PNotify({
|
||||
title: titleSuccess,
|
||||
text: textReconnect,
|
||||
callbacks: {
|
||||
before_close: beforeClose
|
||||
},
|
||||
hide: false
|
||||
})
|
||||
} else {
|
||||
notification = new PNotify({
|
||||
title: titleSuccess,
|
||||
|
|
@ -1060,7 +1069,7 @@ $(function() {
|
|||
});
|
||||
self._scrollWorkingOutputToEnd();
|
||||
} else if (messageType == "result") {
|
||||
var titleSuccess, textSuccess, textRestart, textReload, titleError, textError;
|
||||
var titleSuccess, textSuccess, textRestart, textReload, textReconnect, titleError, textError;
|
||||
var action = data.action;
|
||||
|
||||
var name = "Unknown";
|
||||
|
|
@ -1080,16 +1089,19 @@ $(function() {
|
|||
textSuccess = gettext("A plugin was installed successfully, however it was impossible to detect which one. Please Restart OctoPrint to make sure everything will be registered properly");
|
||||
textRestart = textSuccess;
|
||||
textReload = textSuccess;
|
||||
textReconnect = textSuccess;
|
||||
} else if (data.was_reinstalled) {
|
||||
titleSuccess = _.sprintf(gettext("Plugin \"%(name)s\" reinstalled"), {name: name});
|
||||
textSuccess = gettext("The plugin was reinstalled successfully");
|
||||
textRestart = gettext("The plugin was reinstalled successfully, however a restart of OctoPrint is needed for that to take effect.");
|
||||
textReload = gettext("The plugin was reinstalled successfully, however a reload of the page is needed for that to take effect.");
|
||||
textReconnect = gettext("The plugin was reinstalled successfully, however a reconnect to the printer is needed for that to take effect.");
|
||||
} else {
|
||||
titleSuccess = _.sprintf(gettext("Plugin \"%(name)s\" installed"), {name: name});
|
||||
textSuccess = gettext("The plugin was installed successfully");
|
||||
textRestart = gettext("The plugin was installed successfully, however a restart of OctoPrint is needed for that to take effect.");
|
||||
textReload = gettext("The plugin was installed successfully, however a reload of the page is needed for that to take effect.");
|
||||
textReconnect = gettext("The plugin was installed successfully, however a reconnect to the printer is needed for that to take effect.");
|
||||
}
|
||||
|
||||
titleError = gettext("Something went wrong");
|
||||
|
|
@ -1141,6 +1153,7 @@ $(function() {
|
|||
textSuccess = gettext("The plugin was uninstalled successfully");
|
||||
textRestart = gettext("The plugin was uninstalled successfully, however a restart of OctoPrint is needed for that to take effect.");
|
||||
textReload = gettext("The plugin was uninstalled successfully, however a reload of the page is needed for that to take effect.");
|
||||
textReconnect = gettext("The plugin was uninstalled successfully, however a reconnect to the printer is needed for that to take effect.");
|
||||
|
||||
titleError = gettext("Something went wrong");
|
||||
if (data.hasOwnProperty("reason")) {
|
||||
|
|
@ -1158,6 +1171,7 @@ $(function() {
|
|||
textSuccess = gettext("The plugin was enabled successfully.");
|
||||
textRestart = gettext("The plugin was enabled successfully, however a restart of OctoPrint is needed for that to take effect.");
|
||||
textReload = gettext("The plugin was enabled successfully, however a reload of the page is needed for that to take effect.");
|
||||
textReconnect = gettext("The plugin was enabled successfully, however a reconnect to the printer is needed for that to take effect.");
|
||||
|
||||
titleError = gettext("Something went wrong");
|
||||
if (data.hasOwnProperty("reason")) {
|
||||
|
|
@ -1175,6 +1189,7 @@ $(function() {
|
|||
textSuccess = gettext("The plugin was disabled successfully.");
|
||||
textRestart = gettext("The plugin was disabled successfully, however a restart of OctoPrint is needed for that to take effect.");
|
||||
textReload = gettext("The plugin was disabled successfully, however a reload of the page is needed for that to take effect.");
|
||||
textReconnect = gettext("The plugin was disabled successfully, however a reconnect to the printer is needed for that to take effect.");
|
||||
|
||||
titleError = gettext("Something went wrong");
|
||||
if (data.hasOwnProperty("reason")) {
|
||||
|
|
@ -1187,7 +1202,7 @@ $(function() {
|
|||
return;
|
||||
}
|
||||
|
||||
self._displayNotification(data, titleSuccess, textSuccess, textRestart, textReload, titleError, textError);
|
||||
self._displayNotification(data, titleSuccess, textSuccess, textRestart, textReload, textReconnect, titleError, textError);
|
||||
self.requestData();
|
||||
}
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,10 +1,11 @@
|
|||
import unittest
|
||||
import mock
|
||||
import ddt
|
||||
|
||||
import octoprint.plugin
|
||||
import octoprint.plugin.core
|
||||
|
||||
|
||||
@ddt.ddt
|
||||
class PluginTestCase(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
|
|
@ -191,3 +192,85 @@ class PluginTestCase(unittest.TestCase):
|
|||
plugin = self.plugin_manager.enabled_plugins["deprecated_plugin"]
|
||||
self.assertTrue(hasattr(plugin.instance, plugin.__class__.attr_implementation))
|
||||
self.assertFalse(hasattr(plugin.instance, plugin.__class__.attr_implementations))
|
||||
|
||||
@ddt.data(
|
||||
(["octoprint.some_hook"], ["octoprint.some_hook", "octoprint.another_hook"], True),
|
||||
(["octoprint.*"], ["octoprint.some_hook", "octoprint.another_hook"], True),
|
||||
(["octoprint.some_hook"], ["octoprint.another_hook"], False),
|
||||
(["octoprint.some_hook"], [], False),
|
||||
([], ["octoprint.some_hook"], False)
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_has_any_of_hooks(self, hooks_to_test_for, plugin_hooks, expected):
|
||||
plugin = mock.MagicMock()
|
||||
plugin.hooks = dict((hook, hook) for hook in plugin_hooks)
|
||||
|
||||
actual = octoprint.plugin.core.PluginManager.has_any_of_hooks(plugin, hooks_to_test_for)
|
||||
self.assertEqual(actual, expected)
|
||||
|
||||
def test_has_any_of_hooks_varargs(self):
|
||||
plugin = mock.MagicMock()
|
||||
plugin.hooks = dict((hook, hook) for hook in ["octoprint.some_hook", "octoprint.another_hook"])
|
||||
|
||||
result = octoprint.plugin.core.PluginManager.has_any_of_hooks(plugin, "octoprint.some_hook", "octoprint.some_other_hook")
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_has_any_of_hooks_nohooks(self):
|
||||
plugin = mock.MagicMock()
|
||||
|
||||
result = octoprint.plugin.core.PluginManager.has_any_of_hooks(plugin, "octoprint.some_hook", "octoprint.some_other_hook")
|
||||
self.assertFalse(result)
|
||||
|
||||
@ddt.data(
|
||||
("octoprint.some_hook", ["octoprint.another_hook", "octoprint.some_hook"], True),
|
||||
("octoprint.some_hook", ["octoprint.*"], True),
|
||||
("octoprint.some_hook", ["octoprint.some_hook*"], True),
|
||||
("octoprint.some_hook", ["octoprint.*_hook"], True),
|
||||
("octoprint.some_hook", ["octoprint.another_hook.*"], False),
|
||||
("", ["octoprint.some_hook"], False),
|
||||
(None, ["octoprint.some_hook"], False),
|
||||
("octoprint.some_hook", [], False),
|
||||
("octoprint.some_hook", None, False),
|
||||
("octoprint.some_hook", [None], False)
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_hook_matches_hooks(self, hook, hooks, expected):
|
||||
actual = octoprint.plugin.core.PluginManager.hook_matches_hooks(hook, hooks)
|
||||
self.assertEqual(actual, expected)
|
||||
|
||||
def test_hook_matches_hooks_varargs(self):
|
||||
result = octoprint.plugin.core.PluginManager.hook_matches_hooks("octoprint.some_hook",
|
||||
"octoprint.another_hook", "octoprint.some_hook")
|
||||
self.assertTrue(result)
|
||||
|
||||
@ddt.data(
|
||||
([octoprint.plugin.RestartNeedingPlugin], [octoprint.plugin.Plugin, octoprint.plugin.RestartNeedingPlugin], True),
|
||||
([octoprint.plugin.RestartNeedingPlugin], [octoprint.plugin.Plugin], False),
|
||||
([], [octoprint.plugin.Plugin], False),
|
||||
([octoprint.plugin.RestartNeedingPlugin], [], False)
|
||||
)
|
||||
@ddt.unpack
|
||||
def test_has_any_of_mixins(self, mixins_to_test_for, plugin_mixins, expected):
|
||||
plugin = mock.MagicMock()
|
||||
plugin.implementation = mock.MagicMock()
|
||||
|
||||
for mixin in plugin_mixins:
|
||||
plugin.implementation.mock_add_spec(mixin)
|
||||
|
||||
actual = octoprint.plugin.core.PluginManager.has_any_of_mixins(plugin, mixins_to_test_for)
|
||||
self.assertEqual(actual, expected)
|
||||
|
||||
def test_has_any_of_mixins_varargs(self):
|
||||
plugin = mock.MagicMock()
|
||||
plugin.implementation = mock.MagicMock()
|
||||
plugin.implementation.mock_add_spec(octoprint.plugin.Plugin)
|
||||
plugin.implementation.mock_add_spec(octoprint.plugin.RestartNeedingPlugin)
|
||||
|
||||
result = octoprint.plugin.core.PluginManager.has_any_of_mixins(plugin, octoprint.plugin.RestartNeedingPlugin)
|
||||
self.assertTrue(result)
|
||||
|
||||
def test_has_any_of_mixins_noimplementation(self):
|
||||
plugin = mock.MagicMock()
|
||||
|
||||
result = octoprint.plugin.core.PluginManager.has_any_of_mixins(plugin, octoprint.plugin.RestartNeedingPlugin)
|
||||
self.assertFalse(result)
|
||||
|
|
|
|||
Loading…
Reference in a new issue