Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_plugins():
with pytest.raises(PluginException):
BasePlugin()
class Plugin(BasePlugin):
name = 'plugin'
defaults = {
'debug': True
}
pl1 = Plugin(test=42)
pl2 = Plugin(test=24)
assert pl1 is pl2
assert pl1.cfg.test == 42
app = muffin.Application(__name__)
app.install(pl1)
self.cfg.home = admin_home
app.register(self.cfg.prefix)(self.cfg.home)
if not self.cfg.i18n:
app.ps.jinja2.env.globals.update({
'_': lambda s: s,
'gettext': lambda s: s,
'ngettext': lambda s, p, n: (n != 1 and (p,) or (s,))[0],
})
return
if 'babel' not in app.ps or not isinstance(app.ps.babel, BPlugin):
raise PluginException(
'Plugin `%s` requires for plugin `%s` to be installed to the application.' % (
self.name, BPlugin))
# Connect admin locales
app.ps.babel.cfg.locales_dirs.append(op.join(PLUGIN_ROOT, 'locales'))
if not app.ps.babel.locale_selector_func:
app.ps.babel.locale_selector_func = app.ps.babel.select_locale_by_request
def setup(self, app):
""" Initialize the application. """
super().setup(app)
if 'manage' not in app.plugins:
raise PluginException('Peewee plugin requires Manage plugin initialized before.')
# Setup Database
self.database.initialize(connect(self.options['connection']))
self.threadpool = concurrent.futures.ThreadPoolExecutor(
max_workers=self.options['max_connections'])
if not self.options.migrations_enabled:
return
# Setup migration engine
self.router = Router(self)
self.register(MigrateHistory)
# Register migration commands
@self.app.ps.manage.command
def migrate(name:str=None):
def authorization(self, func):
""" Define a authorization process. """
if self.app is None:
raise PluginException('The plugin must be installed to application.')
self.authorize = muffin.to_coroutine(func)
return func
def setup(self, app):
"""Initialize the plugin.
Fill the plugin's options from application.
"""
self.app = app
for name, ptype in self.dependencies.items():
if name not in app.ps or not isinstance(app.ps[name], ptype):
raise PluginException(
'Plugin `%s` requires for plugin `%s` to be installed to the application.' % (
self.name, ptype))
for oname, dvalue in self.defaults.items():
aname = ('%s_%s' % (self.name, oname)).upper()
app.cfg.setdefault(aname, dvalue)
self.cfg.setdefault(oname, app.cfg[aname])
def __call__(cls, *args, **kwargs):
"""Check for the plugin is initialized already."""
if not cls.name:
raise PluginException('Plugin `%s` doesn\'t have a name.' % cls)
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]