aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/python/pluggy
diff options
context:
space:
mode:
authordeshevoy <deshevoy@yandex-team.ru>2022-02-10 16:46:57 +0300
committerDaniil Cherednik <dcherednik@yandex-team.ru>2022-02-10 16:46:57 +0300
commit28148f76dbfcc644d96427d41c92f36cbf2fdc6e (patch)
treeb83306b6e37edeea782e9eed673d89286c4fef35 /contrib/python/pluggy
parente988f30484abe5fdeedcc7a5d3c226c01a21800c (diff)
downloadydb-28148f76dbfcc644d96427d41c92f36cbf2fdc6e.tar.gz
Restoring authorship annotation for <deshevoy@yandex-team.ru>. Commit 2 of 2.
Diffstat (limited to 'contrib/python/pluggy')
-rw-r--r--contrib/python/pluggy/py2/pluggy/__init__.py36
-rw-r--r--contrib/python/pluggy/py2/pluggy/_tracing.py96
-rw-r--r--contrib/python/pluggy/py2/pluggy/_version.py6
-rw-r--r--contrib/python/pluggy/py2/pluggy/callers.py416
-rw-r--r--contrib/python/pluggy/py2/pluggy/hooks.py644
-rw-r--r--contrib/python/pluggy/py2/pluggy/manager.py616
-rw-r--r--contrib/python/pluggy/py2/ya.make32
-rw-r--r--contrib/python/pluggy/ya.make10
8 files changed, 928 insertions, 928 deletions
diff --git a/contrib/python/pluggy/py2/pluggy/__init__.py b/contrib/python/pluggy/py2/pluggy/__init__.py
index da33974c59..fb4f991a61 100644
--- a/contrib/python/pluggy/py2/pluggy/__init__.py
+++ b/contrib/python/pluggy/py2/pluggy/__init__.py
@@ -1,18 +1,18 @@
-try:
- from ._version import version as __version__
-except ImportError:
- # broken installation, we don't even try
- # unknown only works because we do poor mans version compare
- __version__ = "unknown"
-
-__all__ = [
- "PluginManager",
- "PluginValidationError",
- "HookCallError",
- "HookspecMarker",
- "HookimplMarker",
-]
-
-from .manager import PluginManager, PluginValidationError
-from .callers import HookCallError
-from .hooks import HookspecMarker, HookimplMarker
+try:
+ from ._version import version as __version__
+except ImportError:
+ # broken installation, we don't even try
+ # unknown only works because we do poor mans version compare
+ __version__ = "unknown"
+
+__all__ = [
+ "PluginManager",
+ "PluginValidationError",
+ "HookCallError",
+ "HookspecMarker",
+ "HookimplMarker",
+]
+
+from .manager import PluginManager, PluginValidationError
+from .callers import HookCallError
+from .hooks import HookspecMarker, HookimplMarker
diff --git a/contrib/python/pluggy/py2/pluggy/_tracing.py b/contrib/python/pluggy/py2/pluggy/_tracing.py
index 2a80152a92..8b9715f026 100644
--- a/contrib/python/pluggy/py2/pluggy/_tracing.py
+++ b/contrib/python/pluggy/py2/pluggy/_tracing.py
@@ -1,62 +1,62 @@
-"""
-Tracing utils
-"""
-
-
-class TagTracer(object):
- def __init__(self):
+"""
+Tracing utils
+"""
+
+
+class TagTracer(object):
+ def __init__(self):
self._tags2proc = {}
self._writer = None
- self.indent = 0
-
- def get(self, name):
- return TagTracerSub(self, (name,))
-
+ self.indent = 0
+
+ def get(self, name):
+ return TagTracerSub(self, (name,))
+
def _format_message(self, tags, args):
- if isinstance(args[-1], dict):
- extra = args[-1]
- args = args[:-1]
- else:
- extra = {}
-
- content = " ".join(map(str, args))
- indent = " " * self.indent
-
- lines = ["%s%s [%s]\n" % (indent, content, ":".join(tags))]
-
- for name, value in extra.items():
- lines.append("%s %s: %s\n" % (indent, name, value))
-
+ if isinstance(args[-1], dict):
+ extra = args[-1]
+ args = args[:-1]
+ else:
+ extra = {}
+
+ content = " ".join(map(str, args))
+ indent = " " * self.indent
+
+ lines = ["%s%s [%s]\n" % (indent, content, ":".join(tags))]
+
+ for name, value in extra.items():
+ lines.append("%s %s: %s\n" % (indent, name, value))
+
return "".join(lines)
def _processmessage(self, tags, args):
if self._writer is not None and args:
self._writer(self._format_message(tags, args))
- try:
+ try:
processor = self._tags2proc[tags]
- except KeyError:
- pass
+ except KeyError:
+ pass
else:
processor(tags, args)
-
- def setwriter(self, writer):
+
+ def setwriter(self, writer):
self._writer = writer
-
- def setprocessor(self, tags, processor):
- if isinstance(tags, str):
- tags = tuple(tags.split(":"))
- else:
- assert isinstance(tags, tuple)
+
+ def setprocessor(self, tags, processor):
+ if isinstance(tags, str):
+ tags = tuple(tags.split(":"))
+ else:
+ assert isinstance(tags, tuple)
self._tags2proc[tags] = processor
-
-
-class TagTracerSub(object):
- def __init__(self, root, tags):
- self.root = root
- self.tags = tags
-
- def __call__(self, *args):
+
+
+class TagTracerSub(object):
+ def __init__(self, root, tags):
+ self.root = root
+ self.tags = tags
+
+ def __call__(self, *args):
self.root._processmessage(self.tags, args)
-
- def get(self, name):
- return self.__class__(self.root, self.tags + (name,))
+
+ def get(self, name):
+ return self.__class__(self.root, self.tags + (name,))
diff --git a/contrib/python/pluggy/py2/pluggy/_version.py b/contrib/python/pluggy/py2/pluggy/_version.py
index 41ad9f5ba5..2ba90cb83e 100644
--- a/contrib/python/pluggy/py2/pluggy/_version.py
+++ b/contrib/python/pluggy/py2/pluggy/_version.py
@@ -1,4 +1,4 @@
-# coding: utf-8
-# file generated by setuptools_scm
-# don't change, don't track in version control
+# coding: utf-8
+# file generated by setuptools_scm
+# don't change, don't track in version control
version = '0.13.1'
diff --git a/contrib/python/pluggy/py2/pluggy/callers.py b/contrib/python/pluggy/py2/pluggy/callers.py
index ff0b5e8fd2..e7ea464b0d 100644
--- a/contrib/python/pluggy/py2/pluggy/callers.py
+++ b/contrib/python/pluggy/py2/pluggy/callers.py
@@ -1,208 +1,208 @@
-"""
-Call loop machinery
-"""
-import sys
-import warnings
-
-_py3 = sys.version_info > (3, 0)
-
-
-if not _py3:
- exec(
- """
-def _reraise(cls, val, tb):
- raise cls, val, tb
-"""
- )
-
-
-def _raise_wrapfail(wrap_controller, msg):
- co = wrap_controller.gi_code
- raise RuntimeError(
- "wrap_controller at %r %s:%d %s"
- % (co.co_name, co.co_filename, co.co_firstlineno, msg)
- )
-
-
-class HookCallError(Exception):
- """ Hook was called wrongly. """
-
-
-class _Result(object):
- def __init__(self, result, excinfo):
- self._result = result
- self._excinfo = excinfo
-
- @property
- def excinfo(self):
- return self._excinfo
-
- @property
- def result(self):
- """Get the result(s) for this hook call (DEPRECATED in favor of ``get_result()``)."""
- msg = "Use get_result() which forces correct exception handling"
- warnings.warn(DeprecationWarning(msg), stacklevel=2)
- return self._result
-
- @classmethod
- def from_call(cls, func):
- __tracebackhide__ = True
- result = excinfo = None
- try:
- result = func()
- except BaseException:
- excinfo = sys.exc_info()
-
- return cls(result, excinfo)
-
- def force_result(self, result):
- """Force the result(s) to ``result``.
-
- If the hook was marked as a ``firstresult`` a single value should
- be set otherwise set a (modified) list of results. Any exceptions
- found during invocation will be deleted.
- """
- self._result = result
- self._excinfo = None
-
- def get_result(self):
- """Get the result(s) for this hook call.
-
- If the hook was marked as a ``firstresult`` only a single value
- will be returned otherwise a list of results.
- """
- __tracebackhide__ = True
- if self._excinfo is None:
- return self._result
- else:
- ex = self._excinfo
- if _py3:
- raise ex[1].with_traceback(ex[2])
- _reraise(*ex) # noqa
-
-
-def _wrapped_call(wrap_controller, func):
- """ Wrap calling to a function with a generator which needs to yield
- exactly once. The yield point will trigger calling the wrapped function
- and return its ``_Result`` to the yield point. The generator then needs
- to finish (raise StopIteration) in order for the wrapped call to complete.
- """
- try:
- next(wrap_controller) # first yield
- except StopIteration:
- _raise_wrapfail(wrap_controller, "did not yield")
- call_outcome = _Result.from_call(func)
- try:
- wrap_controller.send(call_outcome)
- _raise_wrapfail(wrap_controller, "has second yield")
- except StopIteration:
- pass
- return call_outcome.get_result()
-
-
-class _LegacyMultiCall(object):
- """ execute a call into multiple python functions/methods. """
-
- # XXX note that the __multicall__ argument is supported only
- # for pytest compatibility reasons. It was never officially
- # supported there and is explicitely deprecated since 2.8
- # so we can remove it soon, allowing to avoid the below recursion
- # in execute() and simplify/speed up the execute loop.
-
- def __init__(self, hook_impls, kwargs, firstresult=False):
- self.hook_impls = hook_impls
- self.caller_kwargs = kwargs # come from _HookCaller.__call__()
- self.caller_kwargs["__multicall__"] = self
- self.firstresult = firstresult
-
- def execute(self):
- caller_kwargs = self.caller_kwargs
- self.results = results = []
- firstresult = self.firstresult
-
- while self.hook_impls:
- hook_impl = self.hook_impls.pop()
- try:
- args = [caller_kwargs[argname] for argname in hook_impl.argnames]
- except KeyError:
- for argname in hook_impl.argnames:
- if argname not in caller_kwargs:
- raise HookCallError(
- "hook call must provide argument %r" % (argname,)
- )
- if hook_impl.hookwrapper:
- return _wrapped_call(hook_impl.function(*args), self.execute)
- res = hook_impl.function(*args)
- if res is not None:
- if firstresult:
- return res
- results.append(res)
-
- if not firstresult:
- return results
-
- def __repr__(self):
- status = "%d meths" % (len(self.hook_impls),)
- if hasattr(self, "results"):
- status = ("%d results, " % len(self.results)) + status
- return "<_MultiCall %s, kwargs=%r>" % (status, self.caller_kwargs)
-
-
-def _legacymulticall(hook_impls, caller_kwargs, firstresult=False):
- return _LegacyMultiCall(
- hook_impls, caller_kwargs, firstresult=firstresult
- ).execute()
-
-
-def _multicall(hook_impls, caller_kwargs, firstresult=False):
- """Execute a call into multiple python functions/methods and return the
- result(s).
-
- ``caller_kwargs`` comes from _HookCaller.__call__().
- """
- __tracebackhide__ = True
- results = []
- excinfo = None
- try: # run impl and wrapper setup functions in a loop
- teardowns = []
- try:
- for hook_impl in reversed(hook_impls):
- try:
- args = [caller_kwargs[argname] for argname in hook_impl.argnames]
- except KeyError:
- for argname in hook_impl.argnames:
- if argname not in caller_kwargs:
- raise HookCallError(
- "hook call must provide argument %r" % (argname,)
- )
-
- if hook_impl.hookwrapper:
- try:
- gen = hook_impl.function(*args)
- next(gen) # first yield
- teardowns.append(gen)
- except StopIteration:
- _raise_wrapfail(gen, "did not yield")
- else:
- res = hook_impl.function(*args)
- if res is not None:
- results.append(res)
- if firstresult: # halt further impl calls
- break
- except BaseException:
- excinfo = sys.exc_info()
- finally:
- if firstresult: # first result hooks return a single value
- outcome = _Result(results[0] if results else None, excinfo)
- else:
- outcome = _Result(results, excinfo)
-
- # run all wrapper post-yield blocks
- for gen in reversed(teardowns):
- try:
- gen.send(outcome)
- _raise_wrapfail(gen, "has second yield")
- except StopIteration:
- pass
-
- return outcome.get_result()
+"""
+Call loop machinery
+"""
+import sys
+import warnings
+
+_py3 = sys.version_info > (3, 0)
+
+
+if not _py3:
+ exec(
+ """
+def _reraise(cls, val, tb):
+ raise cls, val, tb
+"""
+ )
+
+
+def _raise_wrapfail(wrap_controller, msg):
+ co = wrap_controller.gi_code
+ raise RuntimeError(
+ "wrap_controller at %r %s:%d %s"
+ % (co.co_name, co.co_filename, co.co_firstlineno, msg)
+ )
+
+
+class HookCallError(Exception):
+ """ Hook was called wrongly. """
+
+
+class _Result(object):
+ def __init__(self, result, excinfo):
+ self._result = result
+ self._excinfo = excinfo
+
+ @property
+ def excinfo(self):
+ return self._excinfo
+
+ @property
+ def result(self):
+ """Get the result(s) for this hook call (DEPRECATED in favor of ``get_result()``)."""
+ msg = "Use get_result() which forces correct exception handling"
+ warnings.warn(DeprecationWarning(msg), stacklevel=2)
+ return self._result
+
+ @classmethod
+ def from_call(cls, func):
+ __tracebackhide__ = True
+ result = excinfo = None
+ try:
+ result = func()
+ except BaseException:
+ excinfo = sys.exc_info()
+
+ return cls(result, excinfo)
+
+ def force_result(self, result):
+ """Force the result(s) to ``result``.
+
+ If the hook was marked as a ``firstresult`` a single value should
+ be set otherwise set a (modified) list of results. Any exceptions
+ found during invocation will be deleted.
+ """
+ self._result = result
+ self._excinfo = None
+
+ def get_result(self):
+ """Get the result(s) for this hook call.
+
+ If the hook was marked as a ``firstresult`` only a single value
+ will be returned otherwise a list of results.
+ """
+ __tracebackhide__ = True
+ if self._excinfo is None:
+ return self._result
+ else:
+ ex = self._excinfo
+ if _py3:
+ raise ex[1].with_traceback(ex[2])
+ _reraise(*ex) # noqa
+
+
+def _wrapped_call(wrap_controller, func):
+ """ Wrap calling to a function with a generator which needs to yield
+ exactly once. The yield point will trigger calling the wrapped function
+ and return its ``_Result`` to the yield point. The generator then needs
+ to finish (raise StopIteration) in order for the wrapped call to complete.
+ """
+ try:
+ next(wrap_controller) # first yield
+ except StopIteration:
+ _raise_wrapfail(wrap_controller, "did not yield")
+ call_outcome = _Result.from_call(func)
+ try:
+ wrap_controller.send(call_outcome)
+ _raise_wrapfail(wrap_controller, "has second yield")
+ except StopIteration:
+ pass
+ return call_outcome.get_result()
+
+
+class _LegacyMultiCall(object):
+ """ execute a call into multiple python functions/methods. """
+
+ # XXX note that the __multicall__ argument is supported only
+ # for pytest compatibility reasons. It was never officially
+ # supported there and is explicitely deprecated since 2.8
+ # so we can remove it soon, allowing to avoid the below recursion
+ # in execute() and simplify/speed up the execute loop.
+
+ def __init__(self, hook_impls, kwargs, firstresult=False):
+ self.hook_impls = hook_impls
+ self.caller_kwargs = kwargs # come from _HookCaller.__call__()
+ self.caller_kwargs["__multicall__"] = self
+ self.firstresult = firstresult
+
+ def execute(self):
+ caller_kwargs = self.caller_kwargs
+ self.results = results = []
+ firstresult = self.firstresult
+
+ while self.hook_impls:
+ hook_impl = self.hook_impls.pop()
+ try:
+ args = [caller_kwargs[argname] for argname in hook_impl.argnames]
+ except KeyError:
+ for argname in hook_impl.argnames:
+ if argname not in caller_kwargs:
+ raise HookCallError(
+ "hook call must provide argument %r" % (argname,)
+ )
+ if hook_impl.hookwrapper:
+ return _wrapped_call(hook_impl.function(*args), self.execute)
+ res = hook_impl.function(*args)
+ if res is not None:
+ if firstresult:
+ return res
+ results.append(res)
+
+ if not firstresult:
+ return results
+
+ def __repr__(self):
+ status = "%d meths" % (len(self.hook_impls),)
+ if hasattr(self, "results"):
+ status = ("%d results, " % len(self.results)) + status
+ return "<_MultiCall %s, kwargs=%r>" % (status, self.caller_kwargs)
+
+
+def _legacymulticall(hook_impls, caller_kwargs, firstresult=False):
+ return _LegacyMultiCall(
+ hook_impls, caller_kwargs, firstresult=firstresult
+ ).execute()
+
+
+def _multicall(hook_impls, caller_kwargs, firstresult=False):
+ """Execute a call into multiple python functions/methods and return the
+ result(s).
+
+ ``caller_kwargs`` comes from _HookCaller.__call__().
+ """
+ __tracebackhide__ = True
+ results = []
+ excinfo = None
+ try: # run impl and wrapper setup functions in a loop
+ teardowns = []
+ try:
+ for hook_impl in reversed(hook_impls):
+ try:
+ args = [caller_kwargs[argname] for argname in hook_impl.argnames]
+ except KeyError:
+ for argname in hook_impl.argnames:
+ if argname not in caller_kwargs:
+ raise HookCallError(
+ "hook call must provide argument %r" % (argname,)
+ )
+
+ if hook_impl.hookwrapper:
+ try:
+ gen = hook_impl.function(*args)
+ next(gen) # first yield
+ teardowns.append(gen)
+ except StopIteration:
+ _raise_wrapfail(gen, "did not yield")
+ else:
+ res = hook_impl.function(*args)
+ if res is not None:
+ results.append(res)
+ if firstresult: # halt further impl calls
+ break
+ except BaseException:
+ excinfo = sys.exc_info()
+ finally:
+ if firstresult: # first result hooks return a single value
+ outcome = _Result(results[0] if results else None, excinfo)
+ else:
+ outcome = _Result(results, excinfo)
+
+ # run all wrapper post-yield blocks
+ for gen in reversed(teardowns):
+ try:
+ gen.send(outcome)
+ _raise_wrapfail(gen, "has second yield")
+ except StopIteration:
+ pass
+
+ return outcome.get_result()
diff --git a/contrib/python/pluggy/py2/pluggy/hooks.py b/contrib/python/pluggy/py2/pluggy/hooks.py
index 66133d063e..0a1c287198 100644
--- a/contrib/python/pluggy/py2/pluggy/hooks.py
+++ b/contrib/python/pluggy/py2/pluggy/hooks.py
@@ -1,359 +1,359 @@
-"""
-Internal hook annotation, representation and calling machinery.
-"""
-import inspect
+"""
+Internal hook annotation, representation and calling machinery.
+"""
+import inspect
import sys
-import warnings
-from .callers import _legacymulticall, _multicall
-
-
-class HookspecMarker(object):
- """ Decorator helper class for marking functions as hook specifications.
-
- You can instantiate it with a project_name to get a decorator.
+import warnings
+from .callers import _legacymulticall, _multicall
+
+
+class HookspecMarker(object):
+ """ Decorator helper class for marking functions as hook specifications.
+
+ You can instantiate it with a project_name to get a decorator.
Calling :py:meth:`.PluginManager.add_hookspecs` later will discover all marked functions
if the :py:class:`.PluginManager` uses the same project_name.
- """
-
- def __init__(self, project_name):
- self.project_name = project_name
-
- def __call__(
- self, function=None, firstresult=False, historic=False, warn_on_impl=None
- ):
- """ if passed a function, directly sets attributes on the function
+ """
+
+ def __init__(self, project_name):
+ self.project_name = project_name
+
+ def __call__(
+ self, function=None, firstresult=False, historic=False, warn_on_impl=None
+ ):
+ """ if passed a function, directly sets attributes on the function
which will make it discoverable to :py:meth:`.PluginManager.add_hookspecs`.
If passed no function, returns a decorator which can be applied to a function
- later using the attributes supplied.
-
+ later using the attributes supplied.
+
If ``firstresult`` is ``True`` the 1:N hook call (N being the number of registered
- hook implementation functions) will stop at I<=N when the I'th function
+ hook implementation functions) will stop at I<=N when the I'th function
returns a non-``None`` result.
-
+
If ``historic`` is ``True`` calls to a hook will be memorized and replayed
- on later registered plugins.
-
- """
-
- def setattr_hookspec_opts(func):
- if historic and firstresult:
- raise ValueError("cannot have a historic firstresult hook")
- setattr(
- func,
- self.project_name + "_spec",
- dict(
- firstresult=firstresult,
- historic=historic,
- warn_on_impl=warn_on_impl,
- ),
- )
- return func
-
- if function is not None:
- return setattr_hookspec_opts(function)
- else:
- return setattr_hookspec_opts
-
-
-class HookimplMarker(object):
- """ Decorator helper class for marking functions as hook implementations.
-
+ on later registered plugins.
+
+ """
+
+ def setattr_hookspec_opts(func):
+ if historic and firstresult:
+ raise ValueError("cannot have a historic firstresult hook")
+ setattr(
+ func,
+ self.project_name + "_spec",
+ dict(
+ firstresult=firstresult,
+ historic=historic,
+ warn_on_impl=warn_on_impl,
+ ),
+ )
+ return func
+
+ if function is not None:
+ return setattr_hookspec_opts(function)
+ else:
+ return setattr_hookspec_opts
+
+
+class HookimplMarker(object):
+ """ Decorator helper class for marking functions as hook implementations.
+
You can instantiate with a ``project_name`` to get a decorator.
Calling :py:meth:`.PluginManager.register` later will discover all marked functions
if the :py:class:`.PluginManager` uses the same project_name.
- """
-
- def __init__(self, project_name):
- self.project_name = project_name
-
- def __call__(
- self,
- function=None,
- hookwrapper=False,
- optionalhook=False,
- tryfirst=False,
- trylast=False,
- ):
-
- """ if passed a function, directly sets attributes on the function
+ """
+
+ def __init__(self, project_name):
+ self.project_name = project_name
+
+ def __call__(
+ self,
+ function=None,
+ hookwrapper=False,
+ optionalhook=False,
+ tryfirst=False,
+ trylast=False,
+ ):
+
+ """ if passed a function, directly sets attributes on the function
which will make it discoverable to :py:meth:`.PluginManager.register`.
If passed no function, returns a decorator which can be applied to a
function later using the attributes supplied.
-
+
If ``optionalhook`` is ``True`` a missing matching hook specification will not result
- in an error (by default it is an error if no matching spec is found).
-
+ in an error (by default it is an error if no matching spec is found).
+
If ``tryfirst`` is ``True`` this hook implementation will run as early as possible
in the chain of N hook implementations for a specification.
-
+
If ``trylast`` is ``True`` this hook implementation will run as late as possible
- in the chain of N hook implementations.
-
+ in the chain of N hook implementations.
+
If ``hookwrapper`` is ``True`` the hook implementations needs to execute exactly
one ``yield``. The code before the ``yield`` is run early before any non-hookwrapper
function is run. The code after the ``yield`` is run after all non-hookwrapper
function have run. The ``yield`` receives a :py:class:`.callers._Result` object
representing the exception or result outcome of the inner calls (including other
hookwrapper calls).
-
- """
-
- def setattr_hookimpl_opts(func):
- setattr(
- func,
- self.project_name + "_impl",
- dict(
- hookwrapper=hookwrapper,
- optionalhook=optionalhook,
- tryfirst=tryfirst,
- trylast=trylast,
- ),
- )
- return func
-
- if function is None:
- return setattr_hookimpl_opts
- else:
- return setattr_hookimpl_opts(function)
-
-
-def normalize_hookimpl_opts(opts):
- opts.setdefault("tryfirst", False)
- opts.setdefault("trylast", False)
- opts.setdefault("hookwrapper", False)
- opts.setdefault("optionalhook", False)
-
-
-if hasattr(inspect, "getfullargspec"):
-
- def _getargspec(func):
- return inspect.getfullargspec(func)
-
-
-else:
-
- def _getargspec(func):
- return inspect.getargspec(func)
-
-
+
+ """
+
+ def setattr_hookimpl_opts(func):
+ setattr(
+ func,
+ self.project_name + "_impl",
+ dict(
+ hookwrapper=hookwrapper,
+ optionalhook=optionalhook,
+ tryfirst=tryfirst,
+ trylast=trylast,
+ ),
+ )
+ return func
+
+ if function is None:
+ return setattr_hookimpl_opts
+ else:
+ return setattr_hookimpl_opts(function)
+
+
+def normalize_hookimpl_opts(opts):
+ opts.setdefault("tryfirst", False)
+ opts.setdefault("trylast", False)
+ opts.setdefault("hookwrapper", False)
+ opts.setdefault("optionalhook", False)
+
+
+if hasattr(inspect, "getfullargspec"):
+
+ def _getargspec(func):
+ return inspect.getfullargspec(func)
+
+
+else:
+
+ def _getargspec(func):
+ return inspect.getargspec(func)
+
+
_PYPY3 = hasattr(sys, "pypy_version_info") and sys.version_info.major == 3
-def varnames(func):
- """Return tuple of positional and keywrord argument names for a function,
- method, class or callable.
-
- In case of a class, its ``__init__`` method is considered.
- For methods the ``self`` parameter is not included.
- """
- cache = getattr(func, "__dict__", {})
- try:
- return cache["_varnames"]
- except KeyError:
- pass
-
- if inspect.isclass(func):
- try:
- func = func.__init__
- except AttributeError:
- return (), ()
- elif not inspect.isroutine(func): # callable object?
- try:
- func = getattr(func, "__call__", func)
- except Exception:
+def varnames(func):
+ """Return tuple of positional and keywrord argument names for a function,
+ method, class or callable.
+
+ In case of a class, its ``__init__`` method is considered.
+ For methods the ``self`` parameter is not included.
+ """
+ cache = getattr(func, "__dict__", {})
+ try:
+ return cache["_varnames"]
+ except KeyError:
+ pass
+
+ if inspect.isclass(func):
+ try:
+ func = func.__init__
+ except AttributeError:
+ return (), ()
+ elif not inspect.isroutine(func): # callable object?
+ try:
+ func = getattr(func, "__call__", func)
+ except Exception:
return (), ()
-
- try: # func MUST be a function or method here or we won't parse any args
- spec = _getargspec(func)
- except TypeError:
- return (), ()
-
- args, defaults = tuple(spec.args), spec.defaults
- if defaults:
- index = -len(defaults)
+
+ try: # func MUST be a function or method here or we won't parse any args
+ spec = _getargspec(func)
+ except TypeError:
+ return (), ()
+
+ args, defaults = tuple(spec.args), spec.defaults
+ if defaults:
+ index = -len(defaults)
args, kwargs = args[:index], tuple(args[index:])
- else:
+ else:
kwargs = ()
-
- # strip any implicit instance arg
+
+ # strip any implicit instance arg
# pypy3 uses "obj" instead of "self" for default dunder methods
implicit_names = ("self",) if not _PYPY3 else ("self", "obj")
- if args:
- if inspect.ismethod(func) or (
+ if args:
+ if inspect.ismethod(func) or (
"." in getattr(func, "__qualname__", ()) and args[0] in implicit_names
- ):
- args = args[1:]
-
- try:
+ ):
+ args = args[1:]
+
+ try:
cache["_varnames"] = args, kwargs
- except TypeError:
- pass
+ except TypeError:
+ pass
return args, kwargs
-
-
-class _HookRelay(object):
- """ hook holder object for performing 1:N hook calls where N is the number
- of registered plugins.
-
- """
-
-
-class _HookCaller(object):
- def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
- self.name = name
- self._wrappers = []
- self._nonwrappers = []
- self._hookexec = hook_execute
- self.argnames = None
- self.kwargnames = None
- self.multicall = _multicall
- self.spec = None
- if specmodule_or_class is not None:
- assert spec_opts is not None
- self.set_specification(specmodule_or_class, spec_opts)
-
- def has_spec(self):
- return self.spec is not None
-
- def set_specification(self, specmodule_or_class, spec_opts):
- assert not self.has_spec()
- self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
- if spec_opts.get("historic"):
- self._call_history = []
-
- def is_historic(self):
- return hasattr(self, "_call_history")
-
- def _remove_plugin(self, plugin):
- def remove(wrappers):
- for i, method in enumerate(wrappers):
- if method.plugin == plugin:
- del wrappers[i]
- return True
-
- if remove(self._wrappers) is None:
- if remove(self._nonwrappers) is None:
- raise ValueError("plugin %r not found" % (plugin,))
-
- def get_hookimpls(self):
- # Order is important for _hookexec
- return self._nonwrappers + self._wrappers
-
- def _add_hookimpl(self, hookimpl):
- """Add an implementation to the callback chain.
- """
- if hookimpl.hookwrapper:
- methods = self._wrappers
- else:
- methods = self._nonwrappers
-
- if hookimpl.trylast:
- methods.insert(0, hookimpl)
- elif hookimpl.tryfirst:
- methods.append(hookimpl)
- else:
- # find last non-tryfirst method
- i = len(methods) - 1
- while i >= 0 and methods[i].tryfirst:
- i -= 1
- methods.insert(i + 1, hookimpl)
-
- if "__multicall__" in hookimpl.argnames:
- warnings.warn(
- "Support for __multicall__ is now deprecated and will be"
- "removed in an upcoming release.",
- DeprecationWarning,
- )
- self.multicall = _legacymulticall
-
- def __repr__(self):
- return "<_HookCaller %r>" % (self.name,)
-
- def __call__(self, *args, **kwargs):
- if args:
- raise TypeError("hook calling supports only keyword arguments")
- assert not self.is_historic()
- if self.spec and self.spec.argnames:
- notincall = (
- set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
- )
- if notincall:
- warnings.warn(
- "Argument(s) {} which are declared in the hookspec "
- "can not be found in this hook call".format(tuple(notincall)),
- stacklevel=2,
- )
- return self._hookexec(self, self.get_hookimpls(), kwargs)
-
- def call_historic(self, result_callback=None, kwargs=None, proc=None):
- """Call the hook with given ``kwargs`` for all registered plugins and
- for all plugins which will be registered afterwards.
-
- If ``result_callback`` is not ``None`` it will be called for for each
+
+
+class _HookRelay(object):
+ """ hook holder object for performing 1:N hook calls where N is the number
+ of registered plugins.
+
+ """
+
+
+class _HookCaller(object):
+ def __init__(self, name, hook_execute, specmodule_or_class=None, spec_opts=None):
+ self.name = name
+ self._wrappers = []
+ self._nonwrappers = []
+ self._hookexec = hook_execute
+ self.argnames = None
+ self.kwargnames = None
+ self.multicall = _multicall
+ self.spec = None
+ if specmodule_or_class is not None:
+ assert spec_opts is not None
+ self.set_specification(specmodule_or_class, spec_opts)
+
+ def has_spec(self):
+ return self.spec is not None
+
+ def set_specification(self, specmodule_or_class, spec_opts):
+ assert not self.has_spec()
+ self.spec = HookSpec(specmodule_or_class, self.name, spec_opts)
+ if spec_opts.get("historic"):
+ self._call_history = []
+
+ def is_historic(self):
+ return hasattr(self, "_call_history")
+
+ def _remove_plugin(self, plugin):
+ def remove(wrappers):
+ for i, method in enumerate(wrappers):
+ if method.plugin == plugin:
+ del wrappers[i]
+ return True
+
+ if remove(self._wrappers) is None:
+ if remove(self._nonwrappers) is None:
+ raise ValueError("plugin %r not found" % (plugin,))
+
+ def get_hookimpls(self):
+ # Order is important for _hookexec
+ return self._nonwrappers + self._wrappers
+
+ def _add_hookimpl(self, hookimpl):
+ """Add an implementation to the callback chain.
+ """
+ if hookimpl.hookwrapper:
+ methods = self._wrappers
+ else:
+ methods = self._nonwrappers
+
+ if hookimpl.trylast:
+ methods.insert(0, hookimpl)
+ elif hookimpl.tryfirst:
+ methods.append(hookimpl)
+ else:
+ # find last non-tryfirst method
+ i = len(methods) - 1
+ while i >= 0 and methods[i].tryfirst:
+ i -= 1
+ methods.insert(i + 1, hookimpl)
+
+ if "__multicall__" in hookimpl.argnames:
+ warnings.warn(
+ "Support for __multicall__ is now deprecated and will be"
+ "removed in an upcoming release.",
+ DeprecationWarning,
+ )
+ self.multicall = _legacymulticall
+
+ def __repr__(self):
+ return "<_HookCaller %r>" % (self.name,)
+
+ def __call__(self, *args, **kwargs):
+ if args:
+ raise TypeError("hook calling supports only keyword arguments")
+ assert not self.is_historic()
+ if self.spec and self.spec.argnames:
+ notincall = (
+ set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
+ )
+ if notincall:
+ warnings.warn(
+ "Argument(s) {} which are declared in the hookspec "
+ "can not be found in this hook call".format(tuple(notincall)),
+ stacklevel=2,
+ )
+ return self._hookexec(self, self.get_hookimpls(), kwargs)
+
+ def call_historic(self, result_callback=None, kwargs=None, proc=None):
+ """Call the hook with given ``kwargs`` for all registered plugins and
+ for all plugins which will be registered afterwards.
+
+ If ``result_callback`` is not ``None`` it will be called for for each
non-``None`` result obtained from a hook implementation.
-
- .. note::
- The ``proc`` argument is now deprecated.
- """
- if proc is not None:
- warnings.warn(
- "Support for `proc` argument is now deprecated and will be"
- "removed in an upcoming release.",
- DeprecationWarning,
- )
- result_callback = proc
-
- self._call_history.append((kwargs or {}, result_callback))
- # historizing hooks don't return results
- res = self._hookexec(self, self.get_hookimpls(), kwargs)
- if result_callback is None:
- return
- # XXX: remember firstresult isn't compat with historic
- for x in res or []:
- result_callback(x)
-
- def call_extra(self, methods, kwargs):
- """ Call the hook with some additional temporarily participating
+
+ .. note::
+ The ``proc`` argument is now deprecated.
+ """
+ if proc is not None:
+ warnings.warn(
+ "Support for `proc` argument is now deprecated and will be"
+ "removed in an upcoming release.",
+ DeprecationWarning,
+ )
+ result_callback = proc
+
+ self._call_history.append((kwargs or {}, result_callback))
+ # historizing hooks don't return results
+ res = self._hookexec(self, self.get_hookimpls(), kwargs)
+ if result_callback is None:
+ return
+ # XXX: remember firstresult isn't compat with historic
+ for x in res or []:
+ result_callback(x)
+
+ def call_extra(self, methods, kwargs):
+ """ Call the hook with some additional temporarily participating
methods using the specified ``kwargs`` as call parameters. """
- old = list(self._nonwrappers), list(self._wrappers)
- for method in methods:
- opts = dict(hookwrapper=False, trylast=False, tryfirst=False)
- hookimpl = HookImpl(None, "<temp>", method, opts)
- self._add_hookimpl(hookimpl)
- try:
- return self(**kwargs)
- finally:
- self._nonwrappers, self._wrappers = old
-
- def _maybe_apply_history(self, method):
- """Apply call history to a new hookimpl if it is marked as historic.
- """
- if self.is_historic():
- for kwargs, result_callback in self._call_history:
- res = self._hookexec(self, [method], kwargs)
- if res and result_callback is not None:
- result_callback(res[0])
-
-
-class HookImpl(object):
- def __init__(self, plugin, plugin_name, function, hook_impl_opts):
- self.function = function
- self.argnames, self.kwargnames = varnames(self.function)
- self.plugin = plugin
- self.opts = hook_impl_opts
- self.plugin_name = plugin_name
- self.__dict__.update(hook_impl_opts)
-
- def __repr__(self):
- return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin)
-
-
-class HookSpec(object):
- def __init__(self, namespace, name, opts):
- self.namespace = namespace
- self.function = function = getattr(namespace, name)
- self.name = name
- self.argnames, self.kwargnames = varnames(function)
- self.opts = opts
- self.argnames = ["__multicall__"] + list(self.argnames)
- self.warn_on_impl = opts.get("warn_on_impl")
+ old = list(self._nonwrappers), list(self._wrappers)
+ for method in methods:
+ opts = dict(hookwrapper=False, trylast=False, tryfirst=False)
+ hookimpl = HookImpl(None, "<temp>", method, opts)
+ self._add_hookimpl(hookimpl)
+ try:
+ return self(**kwargs)
+ finally:
+ self._nonwrappers, self._wrappers = old
+
+ def _maybe_apply_history(self, method):
+ """Apply call history to a new hookimpl if it is marked as historic.
+ """
+ if self.is_historic():
+ for kwargs, result_callback in self._call_history:
+ res = self._hookexec(self, [method], kwargs)
+ if res and result_callback is not None:
+ result_callback(res[0])
+
+
+class HookImpl(object):
+ def __init__(self, plugin, plugin_name, function, hook_impl_opts):
+ self.function = function
+ self.argnames, self.kwargnames = varnames(self.function)
+ self.plugin = plugin
+ self.opts = hook_impl_opts
+ self.plugin_name = plugin_name
+ self.__dict__.update(hook_impl_opts)
+
+ def __repr__(self):
+ return "<HookImpl plugin_name=%r, plugin=%r>" % (self.plugin_name, self.plugin)
+
+
+class HookSpec(object):
+ def __init__(self, namespace, name, opts):
+ self.namespace = namespace
+ self.function = function = getattr(namespace, name)
+ self.name = name
+ self.argnames, self.kwargnames = varnames(function)
+ self.opts = opts
+ self.argnames = ["__multicall__"] + list(self.argnames)
+ self.warn_on_impl = opts.get("warn_on_impl")
diff --git a/contrib/python/pluggy/py2/pluggy/manager.py b/contrib/python/pluggy/py2/pluggy/manager.py
index 2472482318..07b42cba2d 100644
--- a/contrib/python/pluggy/py2/pluggy/manager.py
+++ b/contrib/python/pluggy/py2/pluggy/manager.py
@@ -1,37 +1,37 @@
-import inspect
+import inspect
import sys
-from . import _tracing
+from . import _tracing
from .callers import _Result
-from .hooks import HookImpl, _HookRelay, _HookCaller, normalize_hookimpl_opts
-import warnings
-
+from .hooks import HookImpl, _HookRelay, _HookCaller, normalize_hookimpl_opts
+import warnings
+
if sys.version_info >= (3, 8):
from importlib import metadata as importlib_metadata
else:
import importlib_metadata
-
-
-def _warn_for_function(warning, function):
- warnings.warn_explicit(
- warning,
- type(warning),
- lineno=function.__code__.co_firstlineno,
- filename=function.__code__.co_filename,
- )
-
-
-class PluginValidationError(Exception):
- """ plugin failed validation.
-
- :param object plugin: the plugin which failed validation,
- may be a module or an arbitrary object.
- """
-
- def __init__(self, plugin, message):
- self.plugin = plugin
- super(Exception, self).__init__(message)
-
-
+
+
+def _warn_for_function(warning, function):
+ warnings.warn_explicit(
+ warning,
+ type(warning),
+ lineno=function.__code__.co_firstlineno,
+ filename=function.__code__.co_filename,
+ )
+
+
+class PluginValidationError(Exception):
+ """ plugin failed validation.
+
+ :param object plugin: the plugin which failed validation,
+ may be a module or an arbitrary object.
+ """
+
+ def __init__(self, plugin, message):
+ self.plugin = plugin
+ super(Exception, self).__init__(message)
+
+
class DistFacade(object):
"""Emulate a pkg_resources Distribution"""
@@ -49,234 +49,234 @@ class DistFacade(object):
return sorted(dir(self._dist) + ["_dist", "project_name"])
-class PluginManager(object):
+class PluginManager(object):
""" Core :py:class:`.PluginManager` class which manages registration
- of plugin objects and 1:N hook calling.
-
+ of plugin objects and 1:N hook calling.
+
You can register new hooks by calling :py:meth:`add_hookspecs(module_or_class)
<.PluginManager.add_hookspecs>`.
- You can register plugin objects (which contain hooks) by calling
+ You can register plugin objects (which contain hooks) by calling
:py:meth:`register(plugin) <.PluginManager.register>`. The :py:class:`.PluginManager`
is initialized with a prefix that is searched for in the names of the dict
of registered plugin objects.
-
+
For debugging purposes you can call :py:meth:`.PluginManager.enable_tracing`
- which will subsequently send debug information to the trace helper.
- """
-
- def __init__(self, project_name, implprefix=None):
- """If ``implprefix`` is given implementation functions
+ which will subsequently send debug information to the trace helper.
+ """
+
+ def __init__(self, project_name, implprefix=None):
+ """If ``implprefix`` is given implementation functions
will be recognized if their name matches the ``implprefix``. """
- self.project_name = project_name
- self._name2plugin = {}
- self._plugin2hookcallers = {}
- self._plugin_distinfo = []
- self.trace = _tracing.TagTracer().get("pluginmanage")
+ self.project_name = project_name
+ self._name2plugin = {}
+ self._plugin2hookcallers = {}
+ self._plugin_distinfo = []
+ self.trace = _tracing.TagTracer().get("pluginmanage")
self.hook = _HookRelay()
- if implprefix is not None:
- warnings.warn(
- "Support for the `implprefix` arg is now deprecated and will "
- "be removed in an upcoming release. Please use HookimplMarker.",
- DeprecationWarning,
+ if implprefix is not None:
+ warnings.warn(
+ "Support for the `implprefix` arg is now deprecated and will "
+ "be removed in an upcoming release. Please use HookimplMarker.",
+ DeprecationWarning,
stacklevel=2,
- )
- self._implprefix = implprefix
- self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
- methods,
- kwargs,
- firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
- )
-
- def _hookexec(self, hook, methods, kwargs):
- # called from all hookcaller instances.
- # enable_tracing will set its own wrapping function at self._inner_hookexec
- return self._inner_hookexec(hook, methods, kwargs)
-
- def register(self, plugin, name=None):
+ )
+ self._implprefix = implprefix
+ self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
+ methods,
+ kwargs,
+ firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
+ )
+
+ def _hookexec(self, hook, methods, kwargs):
+ # called from all hookcaller instances.
+ # enable_tracing will set its own wrapping function at self._inner_hookexec
+ return self._inner_hookexec(hook, methods, kwargs)
+
+ def register(self, plugin, name=None):
""" Register a plugin and return its canonical name or ``None`` if the name
is blocked from registering. Raise a :py:class:`ValueError` if the plugin
is already registered. """
- plugin_name = name or self.get_canonical_name(plugin)
-
- if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
- if self._name2plugin.get(plugin_name, -1) is None:
- return # blocked plugin, return None to indicate no registration
- raise ValueError(
- "Plugin already registered: %s=%s\n%s"
- % (plugin_name, plugin, self._name2plugin)
- )
-
- # XXX if an error happens we should make sure no state has been
- # changed at point of return
- self._name2plugin[plugin_name] = plugin
-
- # register matching hook implementations of the plugin
- self._plugin2hookcallers[plugin] = hookcallers = []
- for name in dir(plugin):
- hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
- if hookimpl_opts is not None:
- normalize_hookimpl_opts(hookimpl_opts)
- method = getattr(plugin, name)
- hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
- hook = getattr(self.hook, name, None)
- if hook is None:
- hook = _HookCaller(name, self._hookexec)
- setattr(self.hook, name, hook)
- elif hook.has_spec():
- self._verify_hook(hook, hookimpl)
- hook._maybe_apply_history(hookimpl)
- hook._add_hookimpl(hookimpl)
- hookcallers.append(hook)
- return plugin_name
-
- def parse_hookimpl_opts(self, plugin, name):
- method = getattr(plugin, name)
- if not inspect.isroutine(method):
- return
- try:
- res = getattr(method, self.project_name + "_impl", None)
- except Exception:
- res = {}
- if res is not None and not isinstance(res, dict):
- # false positive
- res = None
- # TODO: remove when we drop implprefix in 1.0
- elif res is None and self._implprefix and name.startswith(self._implprefix):
- _warn_for_function(
- DeprecationWarning(
- "The `implprefix` system is deprecated please decorate "
- "this function using an instance of HookimplMarker."
- ),
- method,
- )
- res = {}
- return res
-
- def unregister(self, plugin=None, name=None):
- """ unregister a plugin object and all its contained hook implementations
- from internal data structures. """
- if name is None:
- assert plugin is not None, "one of name or plugin needs to be specified"
- name = self.get_name(plugin)
-
- if plugin is None:
- plugin = self.get_plugin(name)
-
- # if self._name2plugin[name] == None registration was blocked: ignore
- if self._name2plugin.get(name):
- del self._name2plugin[name]
-
- for hookcaller in self._plugin2hookcallers.pop(plugin, []):
- hookcaller._remove_plugin(plugin)
-
- return plugin
-
- def set_blocked(self, name):
- """ block registrations of the given name, unregister if already registered. """
- self.unregister(name=name)
- self._name2plugin[name] = None
-
- def is_blocked(self, name):
+ plugin_name = name or self.get_canonical_name(plugin)
+
+ if plugin_name in self._name2plugin or plugin in self._plugin2hookcallers:
+ if self._name2plugin.get(plugin_name, -1) is None:
+ return # blocked plugin, return None to indicate no registration
+ raise ValueError(
+ "Plugin already registered: %s=%s\n%s"
+ % (plugin_name, plugin, self._name2plugin)
+ )
+
+ # XXX if an error happens we should make sure no state has been
+ # changed at point of return
+ self._name2plugin[plugin_name] = plugin
+
+ # register matching hook implementations of the plugin
+ self._plugin2hookcallers[plugin] = hookcallers = []
+ for name in dir(plugin):
+ hookimpl_opts = self.parse_hookimpl_opts(plugin, name)
+ if hookimpl_opts is not None:
+ normalize_hookimpl_opts(hookimpl_opts)
+ method = getattr(plugin, name)
+ hookimpl = HookImpl(plugin, plugin_name, method, hookimpl_opts)
+ hook = getattr(self.hook, name, None)
+ if hook is None:
+ hook = _HookCaller(name, self._hookexec)
+ setattr(self.hook, name, hook)
+ elif hook.has_spec():
+ self._verify_hook(hook, hookimpl)
+ hook._maybe_apply_history(hookimpl)
+ hook._add_hookimpl(hookimpl)
+ hookcallers.append(hook)
+ return plugin_name
+
+ def parse_hookimpl_opts(self, plugin, name):
+ method = getattr(plugin, name)
+ if not inspect.isroutine(method):
+ return
+ try:
+ res = getattr(method, self.project_name + "_impl", None)
+ except Exception:
+ res = {}
+ if res is not None and not isinstance(res, dict):
+ # false positive
+ res = None
+ # TODO: remove when we drop implprefix in 1.0
+ elif res is None and self._implprefix and name.startswith(self._implprefix):
+ _warn_for_function(
+ DeprecationWarning(
+ "The `implprefix` system is deprecated please decorate "
+ "this function using an instance of HookimplMarker."
+ ),
+ method,
+ )
+ res = {}
+ return res
+
+ def unregister(self, plugin=None, name=None):
+ """ unregister a plugin object and all its contained hook implementations
+ from internal data structures. """
+ if name is None:
+ assert plugin is not None, "one of name or plugin needs to be specified"
+ name = self.get_name(plugin)
+
+ if plugin is None:
+ plugin = self.get_plugin(name)
+
+ # if self._name2plugin[name] == None registration was blocked: ignore
+ if self._name2plugin.get(name):
+ del self._name2plugin[name]
+
+ for hookcaller in self._plugin2hookcallers.pop(plugin, []):
+ hookcaller._remove_plugin(plugin)
+
+ return plugin
+
+ def set_blocked(self, name):
+ """ block registrations of the given name, unregister if already registered. """
+ self.unregister(name=name)
+ self._name2plugin[name] = None
+
+ def is_blocked(self, name):
""" return ``True`` if the given plugin name is blocked. """
- return name in self._name2plugin and self._name2plugin[name] is None
-
- def add_hookspecs(self, module_or_class):
+ return name in self._name2plugin and self._name2plugin[name] is None
+
+ def add_hookspecs(self, module_or_class):
""" add new hook specifications defined in the given ``module_or_class``.
- Functions are recognized if they have been decorated accordingly. """
- names = []
- for name in dir(module_or_class):
- spec_opts = self.parse_hookspec_opts(module_or_class, name)
- if spec_opts is not None:
- hc = getattr(self.hook, name, None)
- if hc is None:
- hc = _HookCaller(name, self._hookexec, module_or_class, spec_opts)
- setattr(self.hook, name, hc)
- else:
- # plugins registered this hook without knowing the spec
- hc.set_specification(module_or_class, spec_opts)
- for hookfunction in hc.get_hookimpls():
- self._verify_hook(hc, hookfunction)
- names.append(name)
-
- if not names:
- raise ValueError(
- "did not find any %r hooks in %r" % (self.project_name, module_or_class)
- )
-
- def parse_hookspec_opts(self, module_or_class, name):
- method = getattr(module_or_class, name)
- return getattr(method, self.project_name + "_spec", None)
-
- def get_plugins(self):
- """ return the set of registered plugins. """
- return set(self._plugin2hookcallers)
-
- def is_registered(self, plugin):
+ Functions are recognized if they have been decorated accordingly. """
+ names = []
+ for name in dir(module_or_class):
+ spec_opts = self.parse_hookspec_opts(module_or_class, name)
+ if spec_opts is not None:
+ hc = getattr(self.hook, name, None)
+ if hc is None:
+ hc = _HookCaller(name, self._hookexec, module_or_class, spec_opts)
+ setattr(self.hook, name, hc)
+ else:
+ # plugins registered this hook without knowing the spec
+ hc.set_specification(module_or_class, spec_opts)
+ for hookfunction in hc.get_hookimpls():
+ self._verify_hook(hc, hookfunction)
+ names.append(name)
+
+ if not names:
+ raise ValueError(
+ "did not find any %r hooks in %r" % (self.project_name, module_or_class)
+ )
+
+ def parse_hookspec_opts(self, module_or_class, name):
+ method = getattr(module_or_class, name)
+ return getattr(method, self.project_name + "_spec", None)
+
+ def get_plugins(self):
+ """ return the set of registered plugins. """
+ return set(self._plugin2hookcallers)
+
+ def is_registered(self, plugin):
""" Return ``True`` if the plugin is already registered. """
- return plugin in self._plugin2hookcallers
-
- def get_canonical_name(self, plugin):
- """ Return canonical name for a plugin object. Note that a plugin
- may be registered under a different name which was specified
+ return plugin in self._plugin2hookcallers
+
+ def get_canonical_name(self, plugin):
+ """ Return canonical name for a plugin object. Note that a plugin
+ may be registered under a different name which was specified
by the caller of :py:meth:`register(plugin, name) <.PluginManager.register>`.
To obtain the name of an registered plugin use :py:meth:`get_name(plugin)
<.PluginManager.get_name>` instead."""
- return getattr(plugin, "__name__", None) or str(id(plugin))
-
- def get_plugin(self, name):
+ return getattr(plugin, "__name__", None) or str(id(plugin))
+
+ def get_plugin(self, name):
""" Return a plugin or ``None`` for the given name. """
- return self._name2plugin.get(name)
-
- def has_plugin(self, name):
+ return self._name2plugin.get(name)
+
+ def has_plugin(self, name):
""" Return ``True`` if a plugin with the given name is registered. """
- return self.get_plugin(name) is not None
-
- def get_name(self, plugin):
+ return self.get_plugin(name) is not None
+
+ def get_name(self, plugin):
""" Return name for registered plugin or ``None`` if not registered. """
- for name, val in self._name2plugin.items():
- if plugin == val:
- return name
-
- def _verify_hook(self, hook, hookimpl):
- if hook.is_historic() and hookimpl.hookwrapper:
- raise PluginValidationError(
- hookimpl.plugin,
- "Plugin %r\nhook %r\nhistoric incompatible to hookwrapper"
- % (hookimpl.plugin_name, hook.name),
- )
- if hook.spec.warn_on_impl:
- _warn_for_function(hook.spec.warn_on_impl, hookimpl.function)
- # positional arg checking
- notinspec = set(hookimpl.argnames) - set(hook.spec.argnames)
- if notinspec:
- raise PluginValidationError(
- hookimpl.plugin,
- "Plugin %r for hook %r\nhookimpl definition: %s\n"
- "Argument(s) %s are declared in the hookimpl but "
- "can not be found in the hookspec"
- % (
- hookimpl.plugin_name,
- hook.name,
- _formatdef(hookimpl.function),
- notinspec,
- ),
- )
-
- def check_pending(self):
- """ Verify that all hooks which have not been verified against
+ for name, val in self._name2plugin.items():
+ if plugin == val:
+ return name
+
+ def _verify_hook(self, hook, hookimpl):
+ if hook.is_historic() and hookimpl.hookwrapper:
+ raise PluginValidationError(
+ hookimpl.plugin,
+ "Plugin %r\nhook %r\nhistoric incompatible to hookwrapper"
+ % (hookimpl.plugin_name, hook.name),
+ )
+ if hook.spec.warn_on_impl:
+ _warn_for_function(hook.spec.warn_on_impl, hookimpl.function)
+ # positional arg checking
+ notinspec = set(hookimpl.argnames) - set(hook.spec.argnames)
+ if notinspec:
+ raise PluginValidationError(
+ hookimpl.plugin,
+ "Plugin %r for hook %r\nhookimpl definition: %s\n"
+ "Argument(s) %s are declared in the hookimpl but "
+ "can not be found in the hookspec"
+ % (
+ hookimpl.plugin_name,
+ hook.name,
+ _formatdef(hookimpl.function),
+ notinspec,
+ ),
+ )
+
+ def check_pending(self):
+ """ Verify that all hooks which have not been verified against
a hook specification are optional, otherwise raise :py:class:`.PluginValidationError`."""
- for name in self.hook.__dict__:
- if name[0] != "_":
- hook = getattr(self.hook, name)
- if not hook.has_spec():
- for hookimpl in hook.get_hookimpls():
- if not hookimpl.optionalhook:
- raise PluginValidationError(
- hookimpl.plugin,
- "unknown hook %r in plugin %r"
- % (name, hookimpl.plugin),
- )
-
+ for name in self.hook.__dict__:
+ if name[0] != "_":
+ hook = getattr(self.hook, name)
+ if not hook.has_spec():
+ for hookimpl in hook.get_hookimpls():
+ if not hookimpl.optionalhook:
+ raise PluginValidationError(
+ hookimpl.plugin,
+ "unknown hook %r in plugin %r"
+ % (name, hookimpl.plugin),
+ )
+
def load_setuptools_entrypoints(self, group, name=None):
""" Load modules from querying the specified setuptools ``group``.
@@ -296,40 +296,40 @@ class PluginManager(object):
or self.is_blocked(ep.name)
):
continue
- plugin = ep.load()
+ plugin = ep.load()
self.register(plugin, name=ep.name)
self._plugin_distinfo.append((plugin, DistFacade(dist)))
count += 1
return count
-
- def list_plugin_distinfo(self):
- """ return list of distinfo/plugin tuples for all setuptools registered
- plugins. """
- return list(self._plugin_distinfo)
-
- def list_name_plugin(self):
- """ return list of name/plugin pairs. """
- return list(self._name2plugin.items())
-
- def get_hookcallers(self, plugin):
- """ get all hook callers for the specified plugin. """
- return self._plugin2hookcallers.get(plugin)
-
- def add_hookcall_monitoring(self, before, after):
- """ add before/after tracing functions for all hooks
- and return an undo function which, when called,
- will remove the added tracers.
-
- ``before(hook_name, hook_impls, kwargs)`` will be called ahead
- of all hook calls and receive a hookcaller instance, a list
- of HookImpl instances and the keyword arguments for the hook call.
-
- ``after(outcome, hook_name, hook_impls, kwargs)`` receives the
+
+ def list_plugin_distinfo(self):
+ """ return list of distinfo/plugin tuples for all setuptools registered
+ plugins. """
+ return list(self._plugin_distinfo)
+
+ def list_name_plugin(self):
+ """ return list of name/plugin pairs. """
+ return list(self._name2plugin.items())
+
+ def get_hookcallers(self, plugin):
+ """ get all hook callers for the specified plugin. """
+ return self._plugin2hookcallers.get(plugin)
+
+ def add_hookcall_monitoring(self, before, after):
+ """ add before/after tracing functions for all hooks
+ and return an undo function which, when called,
+ will remove the added tracers.
+
+ ``before(hook_name, hook_impls, kwargs)`` will be called ahead
+ of all hook calls and receive a hookcaller instance, a list
+ of HookImpl instances and the keyword arguments for the hook call.
+
+ ``after(outcome, hook_name, hook_impls, kwargs)`` receives the
same arguments as ``before`` but also a :py:class:`pluggy.callers._Result` object
- which represents the result of the overall hook call.
- """
+ which represents the result of the overall hook call.
+ """
oldcall = self._inner_hookexec
-
+
def traced_hookexec(hook, hook_impls, kwargs):
before(hook.name, hook_impls, kwargs)
outcome = _Result.from_call(lambda: oldcall(hook, hook_impls, kwargs))
@@ -343,52 +343,52 @@ class PluginManager(object):
return undo
- def enable_tracing(self):
- """ enable tracing of hook calls and return an undo function. """
+ def enable_tracing(self):
+ """ enable tracing of hook calls and return an undo function. """
hooktrace = self.trace.root.get("hook")
-
- def before(hook_name, methods, kwargs):
- hooktrace.root.indent += 1
- hooktrace(hook_name, kwargs)
-
- def after(outcome, hook_name, methods, kwargs):
- if outcome.excinfo is None:
- hooktrace("finish", hook_name, "-->", outcome.get_result())
- hooktrace.root.indent -= 1
-
- return self.add_hookcall_monitoring(before, after)
-
- def subset_hook_caller(self, name, remove_plugins):
+
+ def before(hook_name, methods, kwargs):
+ hooktrace.root.indent += 1
+ hooktrace(hook_name, kwargs)
+
+ def after(outcome, hook_name, methods, kwargs):
+ if outcome.excinfo is None:
+ hooktrace("finish", hook_name, "-->", outcome.get_result())
+ hooktrace.root.indent -= 1
+
+ return self.add_hookcall_monitoring(before, after)
+
+ def subset_hook_caller(self, name, remove_plugins):
""" Return a new :py:class:`.hooks._HookCaller` instance for the named method
- which manages calls to all registered plugins except the
- ones from remove_plugins. """
- orig = getattr(self.hook, name)
- plugins_to_remove = [plug for plug in remove_plugins if hasattr(plug, name)]
- if plugins_to_remove:
- hc = _HookCaller(
- orig.name, orig._hookexec, orig.spec.namespace, orig.spec.opts
- )
- for hookimpl in orig.get_hookimpls():
- plugin = hookimpl.plugin
- if plugin not in plugins_to_remove:
- hc._add_hookimpl(hookimpl)
- # we also keep track of this hook caller so it
- # gets properly removed on plugin unregistration
- self._plugin2hookcallers.setdefault(plugin, []).append(hc)
- return hc
- return orig
-
-
-if hasattr(inspect, "signature"):
-
- def _formatdef(func):
- return "%s%s" % (func.__name__, str(inspect.signature(func)))
-
-
-else:
-
- def _formatdef(func):
- return "%s%s" % (
- func.__name__,
- inspect.formatargspec(*inspect.getargspec(func)),
- )
+ which manages calls to all registered plugins except the
+ ones from remove_plugins. """
+ orig = getattr(self.hook, name)
+ plugins_to_remove = [plug for plug in remove_plugins if hasattr(plug, name)]
+ if plugins_to_remove:
+ hc = _HookCaller(
+ orig.name, orig._hookexec, orig.spec.namespace, orig.spec.opts
+ )
+ for hookimpl in orig.get_hookimpls():
+ plugin = hookimpl.plugin
+ if plugin not in plugins_to_remove:
+ hc._add_hookimpl(hookimpl)
+ # we also keep track of this hook caller so it
+ # gets properly removed on plugin unregistration
+ self._plugin2hookcallers.setdefault(plugin, []).append(hc)
+ return hc
+ return orig
+
+
+if hasattr(inspect, "signature"):
+
+ def _formatdef(func):
+ return "%s%s" % (func.__name__, str(inspect.signature(func)))
+
+
+else:
+
+ def _formatdef(func):
+ return "%s%s" % (
+ func.__name__,
+ inspect.formatargspec(*inspect.getargspec(func)),
+ )
diff --git a/contrib/python/pluggy/py2/ya.make b/contrib/python/pluggy/py2/ya.make
index 28686a4db2..40ae96f0bf 100644
--- a/contrib/python/pluggy/py2/ya.make
+++ b/contrib/python/pluggy/py2/ya.make
@@ -1,34 +1,34 @@
PY2_LIBRARY()
-
+
OWNER(g:python-contrib)
-
+
VERSION(0.13.1)
-
+
LICENSE(MIT)
PEERDIR(
contrib/python/importlib-metadata
)
-NO_LINT()
-
-PY_SRCS(
- TOP_LEVEL
- pluggy/__init__.py
- pluggy/_tracing.py
- pluggy/_version.py
- pluggy/callers.py
- pluggy/hooks.py
- pluggy/manager.py
-)
-
+NO_LINT()
+
+PY_SRCS(
+ TOP_LEVEL
+ pluggy/__init__.py
+ pluggy/_tracing.py
+ pluggy/_version.py
+ pluggy/callers.py
+ pluggy/hooks.py
+ pluggy/manager.py
+)
+
RESOURCE_FILES(
PREFIX contrib/python/pluggy/py2/
.dist-info/METADATA
.dist-info/top_level.txt
)
-END()
+END()
RECURSE_FOR_TESTS(
tests
diff --git a/contrib/python/pluggy/ya.make b/contrib/python/pluggy/ya.make
index 7e8cf49721..43e4c7b2ab 100644
--- a/contrib/python/pluggy/ya.make
+++ b/contrib/python/pluggy/ya.make
@@ -1,9 +1,9 @@
-PY23_LIBRARY()
-
+PY23_LIBRARY()
+
LICENSE(Service-Py23-Proxy)
OWNER(g:python-contrib)
-
+
IF (PYTHON2)
PEERDIR(contrib/python/pluggy/py2)
ELSE()
@@ -11,8 +11,8 @@ ELSE()
ENDIF()
NO_LINT()
-
-END()
+
+END()
RECURSE(
py2