text: # Copyright 2016-2021 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
  
  import reframe.utility as util
  
  
  def attach_to(phase):
      '''Backend function to attach a hook to a given phase.
  
      :meta private:
      '''
      def deco(func):
          if hasattr(func, '_rfm_attach'):
              func._rfm_attach.append(phase)
          else:
              func._rfm_attach = [phase]
  
          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
  
  
  def require_deps(func):
      '''Denote that the decorated test method will use the test dependencies.
  
      See online docs for more information.
      '''
  
      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 hasattr(fn, '_rfm_attach'):
              raise ValueError(f'{fn.__name__} is not a hook')
  
      @property
      def stages(self):
          return 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 = util.OrderedSet()
          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 hasattr(v, '_rfm_attach'):
              self.__hooks.add(Hook(v))
          elif hasattr(v, '_rfm_resolve_deps'):
              v._rfm_attach = ['post_setup']
              self.__hooks.add(Hook(v))
  
      def update(self, hooks, *, denied_hooks=None):
          '''Update the hook registry with the hooks from another hook registry.
  
          The optional ``denied_hooks`` argument takes a set of disallowed
          hook names, preventing their inclusion into the current hook registry.
          '''
          denied_hooks = denied_hooks or set()
          for h in hooks:
              if h.__name__ not in denied_hooks:
                  self.__hooks.add(h)
  
      def __repr__(self):
          return repr(self.__hooks)
  