from django.db import models
from django.forms import MultipleChoiceField, ChoiceField, CheckboxSelectMultiple
from pkg_resources import EntryPoint, iter_entry_points
from pkgutil import iter_modules
import os
import squad
class PluginNotFound(Exception):
pass
class PluginLoader(object):
__plugins__ = None
@classmethod
def load_all(cls):
if cls.__plugins__ is not None:
return cls.__plugins__
entry_points = []
# builtin plugins
builtin_plugins_path = os.path.join(squad.__path__[0], 'plugins')
for _, m, _ in iter_modules([builtin_plugins_path]):
e = EntryPoint(m, 'squad.plugins.' + m, attrs=('Plugin',))
entry_points.append(e)
# external plugins
plugins = iter_entry_points('squad_plugins')
entry_points += list(plugins)
cls.__plugins__ = {e.name: e.resolve() for e in entry_points}
return cls.__plugins__
def get_plugin_instance(name):
try:
plugin_class = PluginLoader.load_all()[name]
except KeyError:
raise PluginNotFound(name)
return plugin_class()
def get_all_plugins():
plugins = PluginLoader.load_all()
return plugins.keys()
def get_plugins_by_feature(features):
"""
Returns a list of plugin names where the plugins implement at least one of
the *features*. *features* must a list of Plugin methods, e.g.
[Plugin.postprocess_testrun, Plugin.postprocess_testjob]
"""
if not features:
return get_all_plugins()
plugins = PluginLoader.load_all().items()
names = set([f.__name__ for f in features])
return [e for e, plugin in plugins if names & set(plugin.__dict__.keys())]
[docs]
def apply_plugins(plugin_names):
"""
This function should be used by code in the SQUAD core to trigger
functionality from plugins.
The ``plugin_names`` argument is list of plugins names to be used. Most
probably, you will want to pass the list of plugins enabled for a given
project, e.g. ``project.enabled_plugins``.
Example::
from squad.core.plugins import apply_plugins
# ...
for plugin in apply_plugins(project.enabled_plugins):
plugin.method(...)
"""
if plugin_names is None:
return
for p in plugin_names:
try:
plugin = get_plugin_instance(p)
yield plugin
except PluginNotFound:
pass
[docs]
class Plugin(object):
"""
This can be used to pass extra arguments to plugins
"""
extra_args = {}
"""
This class must be used as a superclass for all SQUAD plugins. All the
methods declared here have empty implementations (i.e. they do nothing),
and should be overriden in your plugin to provide extra functionality to
the SQUAD core.
"""
[docs]
def postprocess_testrun(self, testrun):
"""
This method is called after a test run has been received by SQUAD, and
the test run data (tests, metrics, metadata, logs, etc) have been saved
to the database.
You can use this method to parse logs, do any special handling of
metadata, test results, etc.
The ``testrun`` arguments is an instance of
``squad.core.models.TestRun``.
"""
pass
[docs]
def postprocess_testjob(self, testjob):
"""
This method is called after a test job has been fetched by SQUAD, and
the test run data (tests, metrics, metadata, logs, etc) have been saved
to the database.
You can use this method to do any processing that is specific to a
given CI backend (e.g. LAVA).
The ``testjob`` arguments is an instance of
``squad.ci.models.TestJob``.
"""
pass
[docs]
def notify_patch_build_created(self, build):
"""
This method is called when a patch build is created. It should notify
the corresponding patch source that the checks are in progress.
The ``build`` argument is an instance of ``squad.core.Build``.
"""
pass
[docs]
def notify_patch_build_finished(self, build):
"""
This method is called when a patch build is finished. It should notify
the patch source about the status of the tests (success, failure, etc).
The ``build`` argument is an instance of ``squad.core.Build``.
"""
pass
[docs]
def get_url(self, object_id):
"""
This method might return service specific URL with given object_id
"""
pass
[docs]
def has_subtasks(self):
"""
This method tells whether or not the plugin will use subtasks to
complete work, meaning that the main function will return but more
results are still working in parallel.
"""
return False
class PluginField(models.CharField):
def __init__(self, **args):
defaults = {'max_length': 256}
defaults.update(args)
self.features = defaults.pop('features', None)
return super(PluginField, self).__init__(**defaults)
def deconstruct(self):
name, path, args, kwargs = super(PluginField, self).deconstruct()
del kwargs["max_length"]
return name, path, args, kwargs
def formfield(self, **kwargs):
plugins = ((v, v) for v in get_plugins_by_feature(self.features))
return ChoiceField(choices=plugins)
class PluginListField(models.TextField):
def __init__(self, **args):
self.features = args.pop('features', None)
return super(PluginListField, self).__init__(**args)
def from_db_value(self, value, *args):
if value is None:
return None
return [item.strip() for item in value.split(',')]
def to_python(self, value):
if isinstance(value, list):
return value
if value is None:
return None
return [item.strip() for item in value.split(',')]
def get_prep_value(self, value):
if value is None:
return value
return ', '.join(value)
def formfield(self, **kwargs):
plugins = ((v, v) for v in get_plugins_by_feature(self.features))
required = not self.null
return MultipleChoiceField(
required=required,
choices=plugins,
widget=CheckboxSelectMultiple,
)