Distreon-Systems
/
shepherd-agent
7
0
Fork 0
Code Issues 8 Pull Requests Releases Wiki Activity

Implement hook system

Browse Source
master
Tom Wilson 6 years ago
parent 3b499a8f53
commit 19d702e62a
9 changed files with 661 additions and 137 deletions
Show Stats Download Patch File Download Diff File

7
setup.py
Unescape View File

@ -13,6 +13,7 @@ setup(
'requests',
'apscheduler',
'paramiko',
'python-dateutil',
'click'
],
extras_require={
@ -23,11 +24,13 @@ setup(
'pytest-flake8',
'pytest-cov',
'pytest-sugar',
'tox'
'tox',
'responses'
],
'test': [
'pytest',
'pytest-flake8'
'pytest-flake8',
'responses'
]
},

543
shepherd/agent/plugin.py
Unescape View File

@ -5,36 +5,124 @@ import logging
import sys
import pkgutil
from collections import namedtuple
from collections.abc import Sequence
from functools import partial
from types import MappingProxyType
import pkg_resources
from configspec import ConfigSpecification
from configspec.specification import _ValueSpecification
import configspec
import preserve
from . import control
from . import tasks
from .. import base_plugins
# Note that while module attributes intended for external use are mixed here, all the external
# ones are pulled into the root package scope and are intended to be accessed that way.
log = logging.getLogger(__name__)
# Cache of loaded plugins so far
_loaded_plugins = {}
plugin_interfaces = {}
plugin_functions = {}
class NamespaceProxy():
"""
Read-only proxy of a mapping (like a dict) allowing item access via attributes. Mapping keys
that are not strings will be ignored, and attribute access to any names starting with "__"
will still be passed to the actual object attributes.
Being a proxy, attributes available and their values will change as the underlying backing
dict is changed.
class PluginLoadError(Exception):
pass
Intended for sitatuations where a dynamic mapping needs to be passed out to client code but
you'd like to heavily suggest that it not be modified.
Note that only the top-level mapping is read only - if the attribute values themselves are
mutable, they may still be modified via the NamespaceProxy.
"""
ClassMarker = namedtuple("ClassMarker", [])
def __init__(self, backing_dict):
"""
Create a new NamespaceProxy, with attribute access to the underlying backing dict passed
in.
"""
object.__setattr__(self, "_dict_proxy", MappingProxyType(backing_dict))
def __getattribute__(self, name):
if name.startswith("__"):
return object.__getattribute__(self, name)
return object.__getattribute__(self, "_dict_proxy")[name]
def __setattr__(self, *args):
raise TypeError("NamespaceProxy does not allow attributes to be modified")
def __delattr__(self, *args):
raise TypeError("NamespaceProxy does not allow attributes to be modified")
def __repr__(self):
keys = sorted(object.__getattribute__(self, "_dict_proxy"))
items = ("{}={!r}".format(k, object.__getattribute__(
self, "_dict_proxy")[k]) for k in keys)
return "{}({})".format(type(self).__name__, ", ".join(items))
def __eq__(self, other):
if isinstance(other, self.__class__):
return (object.__getattribute__(self, "_dict_proxy") ==
object.__getattribute__(other, "_dict_proxy"))
return False
class UnboundMethod():
"""
Simple wrapper to mark that this is a reference to a method hasn't been bound to an instance
yet, (or had a decorator like ''staticmethod'' or ''classmethod'' unwrapped). Sets its
signature to the result of binding to an anonymous object.
"""
def __init__(self, func):
self._func = func
self._bound_func = None
sigobj = self._func.__get__(object())
self.__signature__ = inspect.signature(sigobj)
self.__doc__ = inspect.getdoc(self._func)
self.__name__ = sigobj.__name__
def bind(self, obj):
"""
Bind the wrapped method to an object, and return the result. Once bound, calling
the UnboundMethod will actually call the bound result, and the ''func()'' property will
return it.
"""
self._bound_func = self._func.__get__(obj)
return self._bound_func
@property
def func(self):
if self._bound_func is None:
raise Exception("Cannot get func from UnboundMethod until it has been bound.")
return self._bound_func
def __call__(self, *args, **kwargs):
if self._bound_func is None:
raise Exception("Cannot call UnboundMethod until it has been bound.")
return self._bound_func(*args, **kwargs)
class PluginLoadError(Exception):
pass
def is_instance_check(classtype):
def instance_checker(obj):
def instance_checker(obj, classtype=classtype):
return isinstance(obj, classtype)
return instance_checker
ClassMarker = namedtuple("ClassMarker", [])
def plugin_class(cls):
"""
Class decorator, used to indicate that a class is to be used as the Plugin Class
@ -60,7 +148,10 @@ FunctionMarker = namedtuple("FunctionMarker", ["name"])
def plugin_function(func=None, *, name=None):
"""
Method decorator to register a method as a plugin interface function. Either used directly:
Method decorator to register a method as a plugin interface function.
If `name` is not supplied, the name of the decorated function is used.
Either used directly:
@plugin_function
def my_method(self):
@ -68,7 +159,8 @@ def plugin_function(func=None, *, name=None):
@plugin_function(name="someOtherName")
def my_badly_named_method(self):
Can only be used within the registered Plugin Class (either with @plugin_class or
Can either be used on functions in the root level of the plugin module, or
on methods within the registered Plugin Class (either with @plugin_class or
interface.register_plugin_class() )
"""
if func is None:
@ -78,64 +170,239 @@ def plugin_function(func=None, *, name=None):
return func
def plugin_hook(method):
return method
HookMarker = namedtuple("HookMarker", ["name", "signature"])
def plugin_hook(func=None, *, name=None):
"""
Method decorator to register a hook for the plugin. Will use the decorated function signature
for the hook, and replace the decorated function with the new hook on plugin init, so it can
be called directly. If `name` is not supplied, the name of the decorated function is used.
Like `plugin_function`, can be either used directly:
@plugin_hook
def my_method(self):
or with the optional keyword argument:
@plugin_hook(name="someOtherName")
def my_badly_named_method(self):
As the decorated function is only being used as a convenient way to declare the hook signature
and to call the hook later, the usual Python `self` method binding system isn't appropriate.
Methods in the plugin class _can_ be registered as hooks (as can module root level functions),
but the signature will be used directly (don't add the `self` argument). For technical clarity,
methods in the class should be decorated with `@staticmethod` below the `@plugin_hook`
decorator.
"""
if func is None:
return partial(plugin_hook, name=name)
if isinstance(func, staticmethod):
# Pull the underlying function out of a staticmethod. It's static, so bound object is
# irrelevant
func = func.__get__(object())
if not name:
name = func.__name__
func._shepherd_load_marker = HookMarker(name, inspect.signature(func))
return func
AttachmentMarker = namedtuple("AttachmentMarker", ["hook_identifier"])
def plugin_attachment(hookname):
def wrapped(method):
return method
return wrapped
def plugin_attachment(hook_identifier):
"""
Function decorator to register a function or method as an attachment to a plugin hook.
The `hook_identifier` is a string indicating what hook to attach to. It can either refer to a
hook in _another_ plugin with the form "my_plugin.my_hook", or to a local hook in the same
plugin with just the hook name: "my_hook".
Can either be used on functions in the root level of the plugin module, or
on methods within the registered Plugin Class (either with @plugin_class or
interface.register_plugin_class() )
"""
def attachment_decorator(func):
func._shepherd_load_marker = AttachmentMarker(hook_identifier)
return func
return attachment_decorator
@preserve.preservable(exclude_attrs=('function'))
class InterfaceCall():
def __init__(self, plugin_name, function_name, kwargs=None):
"""
Record an interface function call for future use.
"""
self.plugin_name = plugin_name
self.function_name = function_name
self.function = None
self.kwargs = kwargs
if self.kwargs is None:
self.kwargs = {}
def __restore_init__(self):
self.function = None
def resolve(self, interface_functions):
"""
Resolve the InterfaceFunction this call refers to. Requires a dict of plugin functions,
where the keys are plugin names and the values are NamedTuples containing the interface
functions for that plugin.
"""
if self.plugin_name not in interface_functions:
raise ValueError(F"Plugin '{self.plugin_name}' could not be found to resolve function")
if not hasattr(interface_functions[self.plugin_name], self.function_name):
raise ValueError(F"Interface function '{self.function_name}' could not be found in"
F" plugin '{self.plugin_name}'")
self.function = getattr(interface_functions[self.plugin_name], self.function_name)
def call(self):
"""
Make the interface function call this object refers to (using the stored kwargs).
Must make sure `resolve()` is called first to actually find the function.
"""
return self.function(**self.kwargs)
class InterfaceFunction():
def __init__(self, func, name=None, unbound=False):
def __init__(self, func, name=None, remote_command=False):
"""
Wrapper around a callable to define a plugin interface function. If unbound true, will use
a temp object to analyse the signature, and defer binding it as a method until _bind() is
called.
Wrapper around a callable to define a plugin interface function.
"""
self.func = func
self._unbound = unbound
sigfunc = func
if unbound:
sigfunc = func.__get__(object())
self.remote = remote_command
self.spec = None
if not callable(sigfunc):
if not callable(self.func):
raise TypeError("InterfaceFunction can only be created around a callable or method.")
sig = inspect.signature(self.func)
if self.remote:
self.spec = ConfigSpecification()
for param in sig.parameters:
if param.kind not in (inspect.Parameter.POSITIONAL_OR_KEYWORD,
inspect.Parameter.KEYWORD_ONLY):
raise ValueError("Interface functions must be callable with keyword arguments")
arg_spec = param.annotation
if arg_spec in ("str", str):
arg_spec = configspec.StringSpec()
if arg_spec in ("int", int):
arg_spec = configspec.IntSpec()
if not isinstance(arg_spec, _ValueSpecification):
raise ValueError("Function annotations for a Shepherd Interface function"
"must be a type of ConfigSpecification, or on the the valid"
"type shortcuts")
self.spec.add_spec(param.name, arg_spec)
if sig.return_annotation is not inspect.Signature.empty:
ret_spec = sig.return_annotation
if ret_spec in ("str", str):
ret_spec = configspec.StringSpec()
if ret_spec in ("int", int):
ret_spec = configspec.IntSpec()
if not isinstance(ret_spec, _ValueSpecification):
raise ValueError("Function annotations for a Shepherd Interface function"
"must be a type of ConfigSpecification, or on the the valid"
"type shortcuts")
self.spec.add_spec("return", arg_spec)
func_doc = inspect.getdoc(self.func)
if func_doc:
self.spec.helptext = func_doc
if name:
self.name = name
else:
self.name = sigfunc.__name__
self.name = self.func.__name__
params = inspect.signature(sigfunc)
log.debug(F"Loaded interface function {self.name} with parameters: {params}")
log.debug(F"Loaded interface function {self.name} with parameters: {sig.parameters}")
def _bind(self, obj):
def get_spec(self):
"""
Bind the wrapped method to an object
Get the function spec used for Shepherd Control to know how to call it as a command.
Will return None unless `remote_command` was marked True on creation.
Returns a ConfigSpecification. If a return value spec is present, it uses the reserved
spec name "return". Any docstring on the function is placed in the root ConfigSpecification
helptext.
"""
self.func = self.func.__get__(obj)
self._unbound = False
return self.spec
def __call__(self, *args, **kwargs):
if self._unbound:
raise Exception(
"Cannot call unbound InterfaceFunction (plugin has not yet initialised)")
return self.func(*args, **kwargs)
class HookAttachment():
"""
Simple record to store the details of a deferred hook attachment. Only a class to allow the
func attribute to be changed.
"""
def __init__(self, func, plugin_name, hook_name):
self.func = func
self.plugin_name = plugin_name
self.hook_name = hook_name
class PluginHook():
"""
A hook to call a set of attachements provided by plugins. Calling the hook directly
will call all attached functions, returning the results in a dict where the keys are
the name of the plugin the attachment came from (if there are no attachements, the
result will be an empty dict).
"""
def __init__(self, name, signature):
self.name = name
self.signature = signature
# dict of callables, plugin names as keys
self._attached_functions = {}
self.attachments = MappingProxyType(self._attached_functions)
def _attach(self, new_func, plugin_name):
if not callable(new_func):
raise TypeError("Hook attachment must be callable")
if plugin_name in self._attached_functions:
raise Exception(F"Hook already has attachment from plugin '{plugin_name}'")
new_sig = inspect.signature(new_func)
if str(new_sig) != str(self.signature):
raise Exception(F"Hook attachment signature '{new_sig}' must match the signature "
F"'{str(self.signature)}' for target hook {self.name}")
self._attached_functions[plugin_name] = new_func
def __call__(self, *args, **kwargs):
results = {}
for plugin_name, func in self._attached_functions.items():
results[plugin_name] = func(*args, **kwargs)
return results
class PluginInterface():
def __init__(self):
self._confspec = None
self._loaded = False
self._initialised = False
self._functions = {}
self._hooks = {}
self._attachments = []
self._tasks = []
self._plugin_class = None
self.config = None
self.plugins = None
self.hooks = None
self._plugin_name = "<not yet loaded>"
self._update_state = control.PluginUpdateState()
@ -163,23 +430,129 @@ class PluginInterface():
raise PluginLoadError("plugin_class must be a class")
self._plugin_class = cls
def register_function(self, func, name=None):
def register_function(self, func, name=None, remote_command=False):
"""
Register a function or method as an interface function for the plugin. If name is not
provided, the name of the callable will be used.
"""
self._load_guard()
if isinstance(func, InterfaceFunction):
ifunc = func
else:
ifunc = InterfaceFunction(func, name)
ifunc = InterfaceFunction(func, name, remote_command)
if ifunc.name in self._functions:
raise PluginLoadError(F"Interface function with name '{ifunc.name}' already exists")
self._functions[ifunc.name] = ifunc
def register_attachment(self, func, hook_identifier):
"""
Register a function or method as an attachment to a plugin hook.
The `hook_identifier` is a string indicating what hook to attach to. It can either refer
to a hook in _another_ plugin with the form "my_plugin.my_hook", or to a local hook in the
same plugin with just the hook name: "my_hook".
"""
self._load_guard()
if not callable(func):
raise TypeError("Hook attachment can only be created around a callable or method.")
extra, plugin_name, hook_name = ([None, None]+hook_identifier.split('.'))[-3:]
if extra is not None:
raise ValueError("Hook identifier can contain at most 2 parts around a '.' character")
self._attachments.append(HookAttachment(func, plugin_name, hook_name))
def register_hook(self, name, signature=None):
"""
Register a plugin hook for other plugins to attach to. Can only be registered during
plugin load. Hook will be accessible during init from `PluginInterface.hooks.<hook_name>`.
Optionally, the same hook object is also returned from `register_hook` to allow it to be
stored and called elsewhere.
In most cases, the decorator form (`@plugin_hook`) is more convenient to use, as it will
directly replace the decorated function or method (usually with just `pass` as the content)
with the hook - allowing it to be called as normal.
If the hook requires arguments, either use the decorator form or pass in the `signature`
argument. For basic keyword args, this can just be a sequence of string argument names,
for example:
`register_hook("my_hook", ["arg_a", "arg_b"])`
If a more complex signature is required, a `inspect.Signature` object can be passed in.
This is most easily expressed inline with a lambda (only to use standard Python function
argument definitions - the lambda doesn't get called).
For example:
`register_hook("my_hook", inspect.signature(lambda arg_a, arg_b_with_default=5: None))`
Args:
name: A string (and valid Python identifier) naming the hook. Must be unique within
the plugin.
signature: If None, registers the hook as requiring no arguments. Can either be a list
of argument names, or an `inspect.Signature` object.
"""
self._load_guard()
if not isinstance(name, str):
raise PluginLoadError(F"Hook name must be a string")
if name in self._hooks:
raise PluginLoadError(F"Hook with name '{name}' already exists")
if signature is None:
signature = inspect.Signature([])
elif isinstance(signature, Sequence):
params = []
for param_name in signature:
params.append(inspect.Parameter(
param_name, inspect.Parameter.POSITIONAL_OR_KEYWORD))
signature = inspect.Signature(params)
if not isinstance(signature, inspect.Signature):
raise PluginLoadError("Hook signature must either be a sequence of paremeter names or"
" and instance of inspect.Signature")
self._hooks[name] = PluginHook(name, signature)
return self._hooks[name]
def add_task(self, trigger, interface_function, kwargs=None):
"""
Add a task when creating a new session. Can only be called during init (object or hook).
Will be ignored if Shepherd is resuming an old session.
Args:
trigger: The trigger object (either a CronTrigger or the result of a ConfigTriggerSpec)
interface_function: The interface function on this plugin to call when triggered. Can
either be a callable that was registered as a plugin function or a string matching
the function name.
kwargs: Any keyword arguments to be passed to the function when the task is triggered.
Must be Preservable.
"""
if not self._loaded:
raise Exception("Cannot add plugin tasks until plugin is loaded")
if self._initialised:
raise Exception("Cannot add plugin tasks after plugin has been initialised")
if isinstance(interface_function, str):
if interface_function not in self._functions:
raise Exception("Plugin does not have interface function"
F" named {interface_function}")
task_call = InterfaceCall(self._plugin_name, interface_function, kwargs)
else:
# Find the callable in our interface functions
func_name = None
for name, ifunc in self._functions.items():
if ifunc.func == interface_function:
func_name = name
break
if func_name is None:
raise Exception(F"Function {interface_function} has not been registered"
" with the plugin")
task_call = InterfaceCall(self._plugin_name, func_name, kwargs)
self._tasks.append(tasks.Task(trigger, task_call))
def set_status(self, status):
"""
Set the plugin status, to be sent to Shepherd Control if configured.
@ -305,26 +678,47 @@ def load_plugin(plugin_name, plugin_dir=None):
else:
interface._confspec = ConfigSpecification()
# Scan module for load markers left by decorators
interface._update_state.set_confspec(interface.confspec)
# Scan module for load markers left by decorators and pass them over to register methods
for attr in module.__dict__.values():
for key, attr in module.__dict__.items():
if hasattr(attr, "_shepherd_load_marker"):
if isinstance(attr._shepherd_load_marker, FunctionMarker):
interface.register_function(attr, **attr._shepherd_load_marker._asdict())
elif isinstance(attr._shepherd_load_marker, AttachmentMarker):
interface.register_attachment(attr, **attr._shepherd_load_marker._asdict())
elif isinstance(attr._shepherd_load_marker, HookMarker):
# Hooks are a little different in that we replace the attr with the hook
newhook = interface.register_hook(**attr._shepherd_load_marker._asdict())
setattr(module, key, newhook)
elif isinstance(attr._shepherd_load_marker, ClassMarker):
interface.register_class(attr)
if interface._plugin_class is not None:
# Scan plugin class for marked methods
for attr in interface._plugin_class.__dict__.values():
for key, attr in interface._plugin_class.__dict__.items():
if hasattr(attr, "_shepherd_load_marker"):
if isinstance(attr._shepherd_load_marker, FunctionMarker):
# Instance doesn't exist yet, so need to save unbound methods for binding later
unbound_func = InterfaceFunction(
attr, unbound=True, **attr._shepherd_load_marker._asdict())
interface.register_function(unbound_func)
interface.register_function(UnboundMethod(attr),
**attr._shepherd_load_marker._asdict())
elif isinstance(attr._shepherd_load_marker, AttachmentMarker):
interface.register_attachment(UnboundMethod(attr),
**attr._shepherd_load_marker._asdict())
elif isinstance(attr._shepherd_load_marker, HookMarker):
# Hooks are a little different in that we replace the attr with the hook
newhook = interface.register_hook(**attr._shepherd_load_marker._asdict())
setattr(interface._plugin_class, key, newhook)
interface._update_state.set_confspec(interface.confspec)
# Assemble remote interface function specs
command_spec = {}
for function in interface._functions.values():
if function.remote:
command_spec[function.name] = function.get_spec()
interface._update_state.set_commandspec(command_spec)
interface._loaded = True
@ -332,33 +726,64 @@ def load_plugin(plugin_name, plugin_dir=None):
return interface
def init_plugins(plugin_configs, core_config):
def init_plugins(plugin_configs, core_config, core_interface_functions):
"""
Initialise plugins named as keys in plugin_configs. Plugins must already be loaded.
Returns dict of initialised plugin interfaces, and a dict of interface function namedtuples
(one for each plugin)
"""
# Pick out plugins to init
plugin_interfaces = {}
for plugin_name in plugin_configs.keys():
plugin_interfaces[plugin_name] = _loaded_plugins[plugin_name]
plugin_functions_tuples = {}
plugin_functions_view = MappingProxyType(plugin_functions_tuples)
interface_functions = {}
interface_functions_proxy = MappingProxyType(interface_functions)
# Run plugin init and init hooks
for plugin_name, interface in plugin_interfaces.items():
interface.plugins = plugin_functions_view
interface.plugins = interface_functions_proxy
interface.config = plugin_configs[plugin_name]
interface.hooks = NamespaceProxy(interface._hooks)
# If it has one, instantiate the plugin object and bind functions
# If it has one, instantiate the plugin object and bind methods to it.
if interface._plugin_class is not None:
interface._plugin_obj = interface._plugin_class()
for ifunc in interface._functions.values():
if ifunc._unbound:
ifunc._bind(interface._plugin_obj)
plugin_functions[plugin_name] = interface._functions
if isinstance(ifunc.func, UnboundMethod):
ifunc.func = ifunc.func.bind(interface._plugin_obj)
for attachment in interface._attachments:
if isinstance(attachment.func, UnboundMethod):
attachment.func = attachment.func.bind(interface._plugin_obj)
# Find hooks attachments are referring to and attach them
for attachment in interface._attachments:
hook_plugin_name = attachment.plugin_name
if hook_plugin_name is None:
hook_plugin_name = plugin_name
hook_name = attachment.hook_name
if hook_plugin_name not in plugin_interfaces:
raise ValueError(F"{plugin_name} attachment target plugin "
F"'{hook_plugin_name}' does not exist")
if hook_name not in plugin_interfaces[hook_plugin_name]._hooks:
raise ValueError(F"{plugin_name} attachment target hook "
F"'{hook_plugin_name}:{hook_name}' does not exist")
plugin_interfaces[hook_plugin_name]._hooks[hook_name]._attach(attachment.func,
plugin_name)
# TODO We've run the object __init__, but not the init hook, that needs to be done
interface._initialised = True
# Add core functions
interface_functions['shepherd'] = NamespaceProxy(core_interface_functions)
# Wait until all plugins have run their init before filling in and giving access
# to all the interface functions.
for plugin_name, interface in plugin_interfaces.items():
# Each plugin has a NamespaceProxy of it's interface functions for read-only attr access
interface_functions[plugin_name] = NamespaceProxy(interface._functions)
# Fill in the interface functions view that plugins now have access to
for name, functions in plugin_functions.items():
plugin_functions_tuples[name] = namedtuple(
F'{name}_interface_functions', functions.keys())(**functions)
return plugin_interfaces, interface_functions_proxy

84
tests/assets/classtestplugin.py
Unescape View File

@ -0,0 +1,84 @@
# pylint: disable=no-self-argument
from configspec import *
from shepherd import PluginInterface, plugin_class, plugin_function, plugin_hook, plugin_attachment
"""
Plugin to test the plugin class systems and the various decorator markers
"""
interface = PluginInterface()
confspec = ConfigSpecification()
confspec.add_spec("spec1", StringSpec(helptext="helping!"))
@plugin_function
def module_function(a):
return F"module func {a}"
@plugin_hook
def module_hook(a, b):
pass
@plugin_attachment("module_hook")
def module_attachment(a, b):
return F"module attachment {a} {b}"
@plugin_class
class ClassPlugin():
def __init__(self):
self.config = interface.config
self.interface = interface
self.plugins = interface.plugins
self.hooks = interface.hooks
# Interface functions
@plugin_function
def instance_method(self, a):
return F"instance method {a}"
@plugin_function
@classmethod
def class_method(cls, a):
return F"class method {a}"
@plugin_function
@staticmethod
def static_method(a):
return F"static method {a}"
# Hooks
@plugin_hook(name="instance_hook")
def instance_hook_name(a, b):
pass
@plugin_hook
@staticmethod
def static_hook(a, b):
pass
@plugin_hook
@staticmethod
def static_hook2(a, b):
pass
# Attachments (these are bound before attachment, so self and cls work as normal, and are
# not included in the signature)
@plugin_attachment("instance_hook")
def instance_attach(self, a, b):
return F"instance attachment {a} {b}"
@plugin_attachment("static_hook2")
@classmethod
def class_attach(cls, a, b):
return F"class attachment {a} {b}"
@plugin_attachment("classtestplugin.static_hook")
@staticmethod
def static_attach(a, b):
return F"static attachment {a} {b}"

0
tests/dirtytestplugin.py → tests/assets/dirtytestplugin.py
Unescape View File

42
tests/assets/simpletestplugin.py
Unescape View File

@ -0,0 +1,42 @@
from inspect import signature
from configspec import *
from shepherd import PluginInterface
"""
Plugin to test basic registration calls.
"""
interface = PluginInterface()
confspec = ConfigSpecification()
confspec.add_spec("spec1", StringSpec())
interface.register_confspec(confspec)
def my_interface_function():
return 42
interface.register_function(my_interface_function)
def basic_attachment():
return "basic attachment"
def attachment_with_args(arg_a, arg_b):
return F"attachment with args: {arg_a}, {arg_b}"
def attachment_with_fancy_args(arg_a, arg_b, arg_c=True):
return F"attachment with fancy args: {arg_a}, {arg_b}, {arg_c}"
interface.register_hook("basic_hook")
interface.register_hook("hook_with_args", ("arg_a", "arg_b"))
interface.register_hook("hook_with_fancy_args", signature(lambda arg_a, arg_b, arg_c=True: None))
interface.register_attachment(basic_attachment, "basic_hook")
interface.register_attachment(attachment_with_args, "simpletestplugin.hook_with_args")
interface.register_attachment(attachment_with_fancy_args, "hook_with_fancy_args")

47
tests/classtestplugin.py
Unescape View File

@ -1,47 +0,0 @@
from configspec import *
from shepherd import PluginInterface, plugin_class, plugin_function, plugin_hook, plugin_attachment
interface = PluginInterface()
confspec = ConfigSpecification()
confspec.add_spec("spec1", StringSpec(helptext="helping!"))
@plugin_function
def module_function(a: str):
return "module func return"
@plugin_class
class ClassPlugin():
def __init__(self):
self.config = interface.config
self.interface = interface
self.plugins = interface.plugins
# self.hooks = interface.hooks
@plugin_function
def instance_method(self, a: str):
return "instance method return"
@plugin_function
@classmethod
def class_method(cls, a: str):
return "class method return"
@plugin_function
@staticmethod
def static_method(a: str):
return "static method return"
# @plugin_hook
# def callback(self):
# pass
# @plugin_attachment("pluginname.hookname")
# def caller(self):
# pass
# interface.register_plugin(SystemPlugin)

0
tests/shepherd-test.toml
Unescape View File

16
tests/simpletestplugin.py
Unescape View File

@ -1,16 +0,0 @@
from configspec import *
from shepherd import PluginInterface
interface = PluginInterface()
confspec = ConfigSpecification()
confspec.add_spec("spec1", StringSpec())
interface.register_confspec(confspec)
def my_interface_function():
return 42
interface.register_function(my_interface_function)

59
tests/test_plugins.py
Unescape View File

@ -1,6 +1,9 @@
# pylint: disable=redefined-outer-name
import sys
from pathlib import Path
import pytest
from shepherd.agent import plugin
@ -14,7 +17,7 @@ def clear_plugin_state(plugin_name):
@pytest.fixture
def simple_plugin(request):
interface = plugin.load_plugin("simpletestplugin", request.fspath.dirname)
interface = plugin.load_plugin("simpletestplugin", Path(request.fspath.dirname)/'assets')
return interface
@ -32,30 +35,42 @@ def test_simple_interface_function_load(simple_plugin: plugin.PluginInterface):
@pytest.fixture
def simple_running_plugin(request):
clear_plugin_state("simpletestplugin")
interface = plugin.load_plugin("simpletestplugin", request.fspath.dirname)
interface = plugin.load_plugin("simpletestplugin", Path(request.fspath.dirname)/'assets')
template_config = interface.confspec.get_template()
plugin.init_plugins({"simpletestplugin": template_config}, {"ckey": "cval"})
plugin.init_plugins({"simpletestplugin": template_config}, {"ckey": "cval"}, {})
return interface
def test_simple_plugin_init(simple_running_plugin):
assert "simpletestplugin" in plugin.plugin_interfaces
assert simple_running_plugin._plugin_name == "simpletestplugin"
def test_simple_interface_functions(simple_running_plugin):
# Check module level function dict
assert plugin.plugin_functions["simpletestplugin"]["my_interface_function"]() == 42
assert simple_running_plugin._functions["my_interface_function"]() == 42
# Check functions handed back to plugin
assert simple_running_plugin.plugins["simpletestplugin"].my_interface_function() == 42
def test_simple_hook_attachments(simple_running_plugin):
assert "basic_hook" in simple_running_plugin._hooks
assert simple_running_plugin._hooks['basic_hook']() == {'simpletestplugin': "basic attachment"}
assert simple_running_plugin.hooks.hook_with_args(
3, 7) == {'simpletestplugin': "attachment with args: 3, 7"}
assert simple_running_plugin.hooks.hook_with_fancy_args(
2, 4) == {'simpletestplugin': "attachment with fancy args: 2, 4, True"}
with pytest.raises(TypeError, match="takes 2 positional arguments but 3 were"):
simple_running_plugin.hooks.hook_with_args(3, 7, 5)
def test_dirty_plugin_load(request):
"""
Corner cases in plugin load
"""
interface = plugin.load_plugin("dirtytestplugin", request.fspath.dirname)
interface = plugin.load_plugin("dirtytestplugin", Path(request.fspath.dirname)/'assets')
# Should prefer the confspec actually registered, even if declared after
assert "spec2" in interface.confspec.spec_dict
@ -64,19 +79,37 @@ def test_dirty_plugin_load(request):
@pytest.fixture
def running_class_plugin(request):
clear_plugin_state("classtestplugin")
interface = plugin.load_plugin("classtestplugin", request.fspath.dirname)
interface = plugin.load_plugin("classtestplugin", Path(request.fspath.dirname)/'assets')
template_config = interface.confspec.get_template()
plugin.init_plugins({"classtestplugin": template_config}, {"core_conf": "core_conf_vals"})
plugin.init_plugins({"classtestplugin": template_config}, {"core_conf": "core_conf_vals"}, {})
return interface
def test_class_plugin_init(running_class_plugin):
assert "classtestplugin" in plugin.plugin_interfaces
assert running_class_plugin._plugin_name == "classtestplugin"
def test_class_interface_functions(running_class_plugin):
ifuncs = running_class_plugin.plugins["classtestplugin"]
assert ifuncs.module_function("a") == "module func return"
assert ifuncs.instance_method("a") == "instance method return"
assert ifuncs.class_method("a") == "class method return"
assert ifuncs.static_method("a") == "static method return"
assert ifuncs.module_function(1) == "module func 1"
assert ifuncs.instance_method(2) == "instance method 2"
assert ifuncs.class_method(3) == "class method 3"
assert ifuncs.static_method(4) == "static method 4"
def test_class_hook_attachments(running_class_plugin):
assert running_class_plugin._hooks.keys() == {"module_hook", "instance_hook",
"static_hook", "static_hook2"}
# Internal hooks dict
assert running_class_plugin._hooks['module_hook'](1, 2) == {'classtestplugin':
"module attachment 1 2"}
# Interface hooks namespace
assert running_class_plugin.hooks.instance_hook(3, 4) == {'classtestplugin':
"instance attachment 3 4"}
# Replaced attr in plugin object
assert running_class_plugin._plugin_obj.static_hook(5, 6) == {'classtestplugin':
"static attachment 5 6"}
assert running_class_plugin.hooks.static_hook2(7, 8) == {'classtestplugin':
"class attachment 7 8"}
with pytest.raises(TypeError, match="takes 2 positional arguments but 3 were"):
running_class_plugin.hooks.static_hook(3, 7, 5)

Write Preview
Loading…
Cancel
Save