Plugin frameworks, QA: add unit tests for PluginManager, HookPluginManager, and PageServingPluginManager

This commit is contained in:
Carl Eberhard
2013-09-18 18:24:00 -04:00
parent f131a96c46
commit d4a7c72bb5
10 changed files with 600 additions and 15 deletions
+47 -10
View File
@@ -280,28 +280,65 @@ class HookPluginManager( PluginManager ):
Search all plugins for a function named ``hook_fn_prefix`` + ``hook_name``
and run it passing in args and kwargs.
Return values from each hook are returned in a dictionary keyed with the
plugin names.
:type hook_name: string
:param hook_name: name (suffix) of the hook to run
:rtype: 2-tuple containing (list, dict)
:returns: (possibly modified) args, kwargs
:rtype: dictionary
:returns: where keys are plugin.names and
values return values from the hooks
"""
#TODO: is hook prefix necessary?
#TODO: could be made more efficient if cached by hook_name in the manager on load_plugin
# (low maint. overhead since no dynamic loading/unloading of plugins)
hook_fn_name = ''.join([ self.hook_fn_prefix, hook_name ])
returned = {}
for plugin_name, plugin in self.plugins.items():
hook_fn = getattr( plugin.module, hook_fn_name, None )
if hook_fn and hasattr( hook_fn, '__call__' ):
try:
#log.debug( 'calling %s from %s(%s)', hook_fn.func_name, plugin.name, plugin.module )
hook_fn( *args, **kwargs )
fn_returned = hook_fn( *args, **kwargs )
returned[ plugin.name ] = fn_returned
except Exception, exc:
# fail gracefully and continue with other plugins
log.exception( 'Hook function "%s" failed for plugin "%s"', hook_name, plugin.name )
# may have been altered by hook fns, return in order to act like filter
return args, kwargs
# not sure of utility of this - seems better to be fire-and-forget pub-sub
return returned
def filter_hook( self, hook_name, hook_arg, *args, **kwargs ):
"""
Search all plugins for a function named ``hook_fn_prefix`` + ``hook_name``
and run the first with ``hook_arg`` and every function after with the
return value of the previous.
..note:
This makes plugin load order very important.
:type hook_name: string
:param hook_name: name (suffix) of the hook to run
:type hook_arg: any
:param hook_arg: the arg to be passed between hook functions
:rtype: any
:returns: the modified hook_arg
"""
hook_fn_name = ''.join([ self.hook_fn_prefix, hook_name ])
for plugin_name, plugin in self.plugins.items():
hook_fn = getattr( plugin.module, hook_fn_name, None )
if hook_fn and hasattr( hook_fn, '__call__' ):
try:
hook_arg = hook_fn( hook_arg, *args, **kwargs )
except Exception, exc:
# fail gracefully and continue with other plugins
log.exception( 'Filter hook function "%s" failed for plugin "%s"', hook_name, plugin.name )
# may have been altered by hook fns, return
return hook_arg
# ============================================================================= exceptions
@@ -370,12 +407,12 @@ class PageServingPluginManager( PluginManager ):
:rtype: bool
:returns: True if the path contains a plugin
"""
if not os.path.isdir( plugin_path ):
if not super( PageServingPluginManager, self ).is_plugin( plugin_path ):
return False
#TODO: this is not reliable and forces the inclusion of empty dirs in some situations
if self.serves_templates and not 'templates' in os.listdir( plugin_path ):
return False
if self.serves_static and not 'static' in os.listdir( plugin_path ):
# reject only if we don't have either
listdir = os.listdir( plugin_path )
if( ( 'templates' not in listdir )
and ( 'static' not in listdir ) ):
return False
return True
@@ -7,15 +7,16 @@ Unit tests for base DataProviders.
#TODO: fix off by ones in FilteredDataProvider counters
import imp
import unittest
import StringIO
import tempfilecache
import utility
utility = imp.load_source( 'utility', '../../util/utility.py' )
log = utility.set_up_filelogger( __name__ + '.log' )
utility.add_galaxy_lib_to_path( 'test/unit/datatypes/dataproviders' )
from galaxy.datatypes.dataproviders import base, exceptions
from galaxy import eggs
@@ -7,17 +7,17 @@ Unit tests for base DataProviders.
#TODO: fix off by ones in FilteredDataProvider counters
import imp
import unittest
import StringIO
import tempfilecache
import utility
import test_base_dataproviders
utility = imp.load_source( 'utility', '../../util/utility.py' )
log = utility.set_up_filelogger( __name__ + '.log' )
utility.add_galaxy_lib_to_path( 'test/unit/datatypes/dataproviders' )
from galaxy import eggs
from galaxy.datatypes.dataproviders import line
Binary file not shown.
+61
View File
@@ -0,0 +1,61 @@
"""
"""
import tempfile
import os
import shutil
class MockDir( object ):
def __init__( self, structure_dict, where=None ):
self.structure_dict = structure_dict
self.create_root( structure_dict, where )
def create_root( self, structure_dict, where=None ):
self.root_path = tempfile.mkdtemp( dir=where )
#print 'created root:', self.root_path
self.create_structure( self.root_path, structure_dict )
def create_structure( self, current_path, structure_dict ):
for k, v in structure_dict.items():
# if value is string, create a file in the current path and write v as file contents
if isinstance( v, str ):
self.create_file( os.path.join( current_path, k ), v )
# if it's a dict, create a dir here named k and recurse into it
if isinstance( v, dict ):
subdir_path = os.path.join( current_path, k )
#print 'subdir:', subdir_path
os.mkdir( subdir_path )
self.create_structure( subdir_path, v )
def create_file( self, path, contents ):
#print 'file:', path
with open( path, 'w' ) as newfile:
newfile.write( contents )
def remove( self ):
#print 'removing:', self.root_path
shutil.rmtree( self.root_path )
class MockAppConfig( object ):
def __init__( self, root ):
self.root = root
class MockApp( object ):
def __init__( self, root ):
self.config = MockAppConfig( root )
class MockTrans( object ):
def fill_template( self, filename, template_lookup=None, **kwargs ):
template = template_lookup.get_template( filename )
template.output_encoding = 'utf-8'
return template.render( **kwargs )
if __name__ == '__main__':
td = MockDir({
'file1' : 'Hello\nthere,\t...you',
'dir1' : {
'file2' : 'Blerbler',
}
})
td.remove()
@@ -0,0 +1,254 @@
"""
"""
import os
import imp
import unittest
import types
utility = imp.load_source( 'utility', '../../util/utility.py' )
log = utility.set_up_filelogger( __name__ + '.log' )
utility.add_galaxy_lib_to_path( 'test/unit/web/base' )
from galaxy.web.base.pluginframework import HookPluginManager
import mock
# ----------------------------------------------------------------------------- globals
loading_point = HookPluginManager.loading_point_filename
contents1 = """
import os
def bler( x, y=3 ):
return ( x, y )
"""
contents2 = """
raise Exception( 'Bler' )
"""
contents3 = """
import contents1
def blah( w ):
return tuple( [ w ] + list( contents1.bler( 2 ) ) )
"""
contents4 = """
from galaxy import util
def blah( s ):
return util.listify( s )
"""
contents5 = """
def hook_blah( s ):
return s.title()
def hook_filter_test( s ):
s += ' one'
return s
"""
contents6 = """
def hook_blah( s ):
return s.upper()
def hook_filter_test( s ):
s += ' two'
return s
"""
contents7 = """
def hook_blah( s ):
raise Exception( 'bler' )
def hook_filter_test( s ):
raise Exception( 'bler' )
"""
# -----------------------------------------------------------------------------
class HookPluginManager_TestCase( unittest.TestCase ):
def test_loading_point( self ):
"""should attempt load on dirs containing loading_point file"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
loading_point : contents1
},
'not_a_plugin' : 'blerbler'
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1' ] )
self.assertFalse( plugin_mgr.plugins.get( 'not_a_plugin', False ) )
plugin = plugin_mgr.plugins[ 'plugin1' ]
self.assertEqual( plugin.name, 'plugin1' )
self.assertEqual( plugin.path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertIsInstance( plugin.module, types.ModuleType )
self.assertEqual( plugin.module.bler( 2 ), ( 2, 3 ) )
mock_app_dir.remove()
def test_bad_loading_points( self ):
"""should NOT attempt load on dirs NOT containing loading_point file"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {},
'plugin2' : {
'plogin.py' : 'wot'
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertEqual( plugin_mgr.plugins.keys(), [] )
mock_app_dir.remove()
def test_bad_import( self ):
"""should error gracefully (skip) on bad import"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
loading_point : contents2
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertEqual( plugin_mgr.plugins.keys(), [] )
mock_app_dir.remove()
def test_import_w_rel_import( self ):
"""should allow loading_point to rel. import other modules"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
'contents1.py': contents1,
loading_point : contents3
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins', skip_bad_plugins=False )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1' ] )
plugin = plugin_mgr.plugins[ 'plugin1' ]
self.assertEqual( plugin.name, 'plugin1' )
self.assertEqual( plugin.path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertIsInstance( plugin.module, types.ModuleType )
self.assertEqual( plugin.module.blah( 1 ), ( 1, 2, 3 ) )
mock_app_dir.remove()
def test_import_w_galaxy_import( self ):
"""should allow loading_point to rel. import GALAXY modules"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
loading_point : contents4
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins', skip_bad_plugins=False )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1' ] )
plugin = plugin_mgr.plugins[ 'plugin1' ]
self.assertEqual( plugin.name, 'plugin1' )
self.assertEqual( plugin.path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertIsInstance( plugin.module, types.ModuleType )
self.assertEqual( plugin.module.blah( 'one,two' ), [ 'one', 'two' ] )
mock_app_dir.remove()
def test_run_hooks( self ):
"""should run hooks of loaded plugins"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
loading_point : contents5
},
'plugin2' : {
loading_point : contents6
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins', skip_bad_plugins=False )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2' ] )
return_val_dict = plugin_mgr.run_hook( 'blah', 'one two check' )
self.assertEqual( return_val_dict, { 'plugin1' : 'One Two Check', 'plugin2' : 'ONE TWO CHECK' } )
result = plugin_mgr.filter_hook( 'filter_test', 'check' )
self.assertEqual( result, 'check one two' )
mock_app_dir.remove()
def test_hook_errs( self ):
"""should fail gracefully if hook fails (and continue with other plugins)"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
loading_point : contents5
},
'plugin2' : {
loading_point : contents6
},
'plugin3' : {
loading_point : contents7
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = HookPluginManager( mock_app, directories_setting='plugins', skip_bad_plugins=False )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2', 'plugin3' ] )
return_val_dict = plugin_mgr.run_hook( 'blah', 'one two check' )
self.assertEqual( return_val_dict, { 'plugin1' : 'One Two Check', 'plugin2' : 'ONE TWO CHECK' } )
result = plugin_mgr.filter_hook( 'filter_test', 'check' )
self.assertEqual( result, 'check one two' )
mock_app_dir.remove()
if __name__ == '__main__':
unittest.main()
@@ -0,0 +1,129 @@
"""
"""
import os
import imp
import unittest
utility = imp.load_source( 'utility', '../../util/utility.py' )
log = utility.set_up_filelogger( __name__ + '.log' )
utility.add_galaxy_lib_to_path( 'test/unit/web/base' )
from galaxy.web.base.pluginframework import PageServingPluginManager
import mock
# ----------------------------------------------------------------------------- globals
contents1 = """${what} ${you} ${say}"""
# -----------------------------------------------------------------------------
class PageServingPluginManager_TestCase( unittest.TestCase ):
def test_plugin_load( self ):
"""should attempt load if criteria met"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
'templates' : {},
'static' : {}
},
'plugin2' : {
'static' : {}
},
'plugin3' : {
'templates' : {}
},
'not_a_plugin1' : 'blerbler',
'not_a_plugin2' : {},
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = PageServingPluginManager( mock_app, 'test', directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertEqual( plugin_mgr.base_url, 'test' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2', 'plugin3' ] )
self.assertFalse( plugin_mgr.plugins.get( 'not_a_plugin', False ) )
plugin1 = plugin_mgr.plugins[ 'plugin1' ]
self.assertEqual( plugin1.name, 'plugin1' )
self.assertEqual( plugin1.path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertEqual( plugin1.base_url, '/'.join([ plugin_mgr.base_url, plugin1.name ]) )
self.assertTrue( plugin1.serves_static )
self.assertEqual( plugin1.static_path, os.path.join( plugin1.path, 'static' ) )
self.assertEqual( plugin1.static_url, '/'.join([ plugin1.base_url, 'static' ]) )
self.assertTrue( plugin1.serves_templates )
self.assertEqual( plugin1.template_path, os.path.join( plugin1.path, 'templates' ) )
self.assertEqual( plugin1.template_lookup.__class__.__name__, 'TemplateLookup' )
plugin2 = plugin_mgr.plugins[ 'plugin2' ]
self.assertEqual( plugin2.name, 'plugin2' )
self.assertEqual( plugin2.path, os.path.join( expected_plugins_path, 'plugin2' ) )
self.assertEqual( plugin2.base_url, '/'.join([ plugin_mgr.base_url, plugin2.name ]) )
self.assertTrue( plugin2.serves_static )
self.assertEqual( plugin2.static_path, os.path.join( plugin2.path, 'static' ) )
self.assertEqual( plugin2.static_url, '/'.join([ plugin2.base_url, 'static' ]) )
self.assertFalse( plugin2.serves_templates )
plugin3 = plugin_mgr.plugins[ 'plugin3' ]
self.assertEqual( plugin3.name, 'plugin3' )
self.assertEqual( plugin3.path, os.path.join( expected_plugins_path, 'plugin3' ) )
self.assertEqual( plugin3.base_url, '/'.join([ plugin_mgr.base_url, plugin3.name ]) )
self.assertFalse( plugin3.serves_static )
self.assertTrue( plugin3.serves_templates )
self.assertEqual( plugin1.template_path, os.path.join( plugin1.path, 'templates' ) )
self.assertEqual( plugin1.template_lookup.__class__.__name__, 'TemplateLookup' )
mock_app_dir.remove()
def test_plugin_static_map( self ):
""""""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
'templates' : {},
'static' : {}
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = PageServingPluginManager( mock_app, 'test', directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1' ] )
plugin = plugin_mgr.plugins[ 'plugin1' ]
self.assertEqual( plugin_mgr.get_static_urls_and_paths(), [( plugin.static_url, plugin.static_path )] )
mock_app_dir.remove()
def test_plugin_templates( self ):
""""""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {
'templates' : {
'test.mako' : contents1
},
}
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = PageServingPluginManager( mock_app, 'test', directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1' ] )
plugin = plugin_mgr.plugins[ 'plugin1' ]
rendered = plugin_mgr.fill_template( mock.MockTrans(), plugin, 'test.mako',
what='Hey', you='Ho', say='HeyHey HoHo' )
self.assertEqual( rendered, 'Hey Ho HeyHey HoHo' )
mock_app_dir.remove()
if __name__ == '__main__':
unittest.main()
+103
View File
@@ -0,0 +1,103 @@
"""
Unit tests for ``galaxy.web.base.pluginframework.PluginManager``
"""
import os
import imp
import unittest
utility = imp.load_source( 'utility', '../../util/utility.py' )
log = utility.set_up_filelogger( __name__ + '.log' )
utility.add_galaxy_lib_to_path( 'test/unit/web/base' )
from galaxy.web.base.pluginframework import PluginManager
import mock
class PluginManager_TestCase( unittest.TestCase ):
def test_rel_path_search( self ):
"""should be able to search given rel. path"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {},
'plugin2' : {},
'file1' : 'blerbler'
}
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = PluginManager( mock_app, directories_setting='plugins' )
app_path = mock_app_dir.root_path
expected_plugins_path = os.path.join( app_path, 'plugins' )
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2' ] )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].name, 'plugin1' )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].name, 'plugin2' )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].path, os.path.join( expected_plugins_path, 'plugin2' ) )
self.assertFalse( plugin_mgr.plugins.get( 'file1', False ) )
mock_app_dir.remove()
def test_abs_path_search( self ):
"""should be able to search given abs. path"""
mock_app_dir = mock.MockDir({})
mock_plugin_dir = mock.MockDir({
'plugin1' : {},
'plugin2' : {},
'file1' : 'blerbler'
})
mock_app = mock.MockApp( mock_app_dir.root_path )
plugin_mgr = PluginManager( mock_app, directories_setting=mock_plugin_dir.root_path )
app_path = mock_app_dir.root_path
expected_plugins_path = mock_plugin_dir.root_path
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2' ] )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].name, 'plugin1' )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].path, os.path.join( expected_plugins_path, 'plugin1' ) )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].name, 'plugin2' )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].path, os.path.join( expected_plugins_path, 'plugin2' ) )
self.assertFalse( plugin_mgr.plugins.get( 'file1', False ) )
def test_multiple_dirs( self ):
"""should search in multiple directories"""
mock_app_dir = mock.MockDir({
'plugins' : {
'plugin1' : {},
'plugin2' : {},
'file1' : 'blerbler'
}
})
mock_abs_plugin_dir = mock.MockDir({
'plugin3' : {},
'plugin4' : {},
'file2' : 'blerbler'
})
mock_app = mock.MockApp( mock_app_dir.root_path )
directories_setting=','.join([ 'plugins', mock_abs_plugin_dir.root_path ])
plugin_mgr = PluginManager( mock_app, directories_setting=directories_setting )
app_path = mock_app_dir.root_path
expected_plugins_rel_path = os.path.join( app_path, 'plugins' )
expected_plugins_abs_path = mock_abs_plugin_dir.root_path
self.assertItemsEqual( plugin_mgr.directories, [ expected_plugins_rel_path, expected_plugins_abs_path ] )
self.assertItemsEqual( plugin_mgr.plugins.keys(), [ 'plugin1', 'plugin2', 'plugin3', 'plugin4' ] )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].name, 'plugin1' )
self.assertEqual( plugin_mgr.plugins[ 'plugin1' ].path, os.path.join( expected_plugins_rel_path, 'plugin1' ) )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].name, 'plugin2' )
self.assertEqual( plugin_mgr.plugins[ 'plugin2' ].path, os.path.join( expected_plugins_rel_path, 'plugin2' ) )
self.assertEqual( plugin_mgr.plugins[ 'plugin3' ].name, 'plugin3' )
self.assertEqual( plugin_mgr.plugins[ 'plugin3' ].path, os.path.join( expected_plugins_abs_path, 'plugin3' ) )
self.assertEqual( plugin_mgr.plugins[ 'plugin4' ].name, 'plugin4' )
self.assertEqual( plugin_mgr.plugins[ 'plugin4' ].path, os.path.join( expected_plugins_abs_path, 'plugin4' ) )
self.assertFalse( plugin_mgr.plugins.get( 'file1', False ) )
self.assertFalse( plugin_mgr.plugins.get( 'file2', False ) )
if __name__ == '__main__':
unittest.main()