Source code for reframe.core.hooks

# Copyright 2016-2024 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
# ReFrame Project Developers. See the top-level LICENSE file for details.
#
# SPDX-License-Identifier: BSD-3-Clause

import functools
import inspect


def is_hook(func):
    return hasattr(func, '_rfm_attach')


def is_dep_hook(func):
    return hasattr(func, '_rfm_resolve_deps')


def attach_to(phase, always_last):
    '''Backend function to attach a hook to a given phase.

    :meta private:
    '''
    def deco(func):
        if is_hook(func):
            func._rfm_attach.append((phase, always_last))
        else:
            func._rfm_attach = [(phase, always_last)]

        try:
            # no need to resolve dependencies independently; this function is
            # already attached to a different phase
            func._rfm_resolve_deps = False
        except AttributeError:
            pass

        @functools.wraps(func)
        def _fn(*args, **kwargs):
            func(*args, **kwargs)

        return _fn

    return deco


[docs] def require_deps(func): '''Decorator to denote that a function will use the test dependencies. The arguments of the decorated function must be named after the dependencies that the function intends to use. The decorator will bind the arguments to a partial realization of the :func:`~reframe.core.pipeline.RegressionTest.getdep` function, such that conceptually the new function arguments will be the following: .. code-block:: python new_arg = functools.partial(getdep, orig_arg_name) The converted arguments are essentially functions accepting a single argument, which is the target test's programming environment. Additionally, this decorator will attach the function to run *after* the test's setup phase, but *before* any other "post-setup" pipeline hook. .. warning:: .. versionchanged:: 3.7.0 Using this functionality from the :py:mod:`reframe` or :py:mod:`reframe.core.decorators` modules is now deprecated. You should use the built-in function described here. .. versionchanged:: 4.0.0 You may only use this function as framework built-in. ''' tests = inspect.getfullargspec(func).args[1:] func._rfm_resolve_deps = True @functools.wraps(func) def _fn(obj, *args): newargs = [functools.partial(obj.getdep, t) for t in tests] func(obj, *newargs) return _fn
def attach_hooks(hooks): '''Attach pipeline hooks to phase ``name''. This function returns a decorator for pipeline functions that will run the registered hooks before and after the function. If ``name'' is :class:`None`, both pre- and post-hooks will run, otherwise only the hooks of the phase ``name'' will be executed. ''' def _deco(func): def select_hooks(obj, kind): phase = kind + func.__name__ if phase not in hooks: return [] return [h for h in hooks.get(phase, []) if h.__name__ not in getattr(obj, '_disabled_hooks', [])] @functools.wraps(func) def _fn(obj, *args, **kwargs): for h in select_hooks(obj, 'pre_'): getattr(obj, h.__name__)() func(obj, *args, **kwargs) for h in select_hooks(obj, 'post_'): getattr(obj, h.__name__)() return _fn return _deco class Hook: '''A pipeline hook. This is essentially a function wrapper that hashes the functions by name, since we want hooks to be overriden by name in subclasses. ''' def __init__(self, fn): self.__fn = fn if not is_hook(fn): raise ValueError(f'{fn.__name__} is not a hook') @property def stages(self): return self._rfm_attach # return [stage for stage, _ in self._rfm_attach] def __getattr__(self, attr): return getattr(self.__fn, attr) @property def fn(self): return self.__fn def __hash__(self): return hash(self.__name__) def __eq__(self, other): if not isinstance(other, type(self)): return NotImplemented return self.__name__ == other.__name__ def __call__(self, *args, **kwargs): return self.__fn(*args, **kwargs) def __repr__(self): return repr(self.__fn) class HookRegistry: '''Global hook registry.''' def __init__(self, hooks=None): self.__hooks = [] if hooks is not None: self.update(hooks) def __contains__(self, key): return key in self.__hooks def __getattr__(self, name): return getattr(self.__hooks, name) def __iter__(self): return iter(self.__hooks) def add(self, v): '''Add value to the hook registry if it meets the conditions. Hook functions have an `_rfm_attach` attribute that specify the stages of the pipeline where they must be attached. Dependencies will be resolved first in the post-setup phase if not assigned elsewhere. ''' if is_hook(v): # Always override hooks with the same name h = Hook(v) try: pos = self.__hooks.index(h) except ValueError: self.__hooks.append(h) else: self.__hooks[pos] = h elif is_dep_hook(v): v._rfm_attach = [('post_setup', None)] self.__hooks.append(Hook(v)) def update(self, other, *, forbidden_names=None): '''Update the hook registry with the hooks from another hook registry.''' assert isinstance(other, HookRegistry) forbidden_names = forbidden_names or {} for h in other: if (h.__name__ in forbidden_names and not is_hook(forbidden_names[h.__name__])): continue try: pos = self.__hooks.index(h) except ValueError: self.__hooks.append(h) else: self.__hooks[pos] = h def __repr__(self): return repr(self.__hooks)