Skip to content
Snippets Groups Projects
Commit cfdf5d8d authored by Stefano Borini's avatar Stefano Borini
Browse files

Merge branch 'master' into remove-old-plugins

parents 8629eb23 fdc66ae1
No related branches found
No related tags found
1 merge request!47Aggregate old plugins under the "dummy" core plugin
from stevedore import extension
import functools
import logging
from stevedore.extension import ExtensionManager
from stevedore.exception import NoMatches
from envisage.api import Application
......@@ -34,18 +37,35 @@ class BDSSApplication(Application):
else:
plugins.append(CoreMCODriver())
mgr = extension.ExtensionManager(
mgr = ExtensionManager(
namespace='force.bdss.extensions',
invoke_on_load=True
invoke_on_load=True,
on_load_failure_callback=functools.partial(_load_failure_callback,
plugins)
)
def import_extensions(ext):
print("Found extension {}".format(ext.name))
plugins.append(ext.obj)
try:
mgr.map(import_extensions)
mgr.map(functools.partial(_import_extensions, plugins))
except NoMatches:
print("No extensions found")
logging.info("No extensions found")
super(BDSSApplication, self).__init__(plugins=plugins)
def _import_extensions(plugins, ext):
"""Service routine extracted for testing.
Imports the extension in the plugins argument.
"""
logging.info("Found extension {}".format(ext.obj))
plugins.append(ext.obj)
def _load_failure_callback(plugins, manager, entry_point, exception):
"""Service routine extracted for testing.
Reports failure to load a module through stevedore, using the
on_load_failure_callback option.
"""
logging.error(
"Unable to load plugin {}. Exception: {}. Message: {}".format(
entry_point, exception.__class__.__name__, exception)
)
import unittest
import testfixtures
from force_bdss.bdss_application import (
BDSSApplication,
_load_failure_callback,
_import_extensions
)
try:
import mock
except ImportError:
from unittest import mock
class TestBDSSApplication(unittest.TestCase):
def test_initialization(self):
app = BDSSApplication(False, "foo/bar")
self.assertFalse(app.evaluate)
self.assertEqual(app.workflow_filepath, "foo/bar")
def test_extension_load_failure(self):
plugins = []
with testfixtures.LogCapture() as log:
_load_failure_callback(plugins,
mock.Mock(),
"foo",
Exception("hello"))
log.check(
('root', 'ERROR', "Unable to load plugin foo. Exception: "
"Exception. Message: hello")
)
self.assertEqual(plugins, [])
def test_import_extension(self):
plugins = []
plugin = mock.Mock()
ext = mock.Mock()
ext.obj = plugin
_import_extensions(plugins, ext)
self.assertEqual(plugins[0], plugin)
flake8
coverage
mock
testfixtures
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment