text: # Copyright 2016-2022 Swiss National Supercomputing Centre (CSCS/ETH Zurich)
  # ReFrame Project Developers. See the top-level LICENSE file for details.
  #
  # SPDX-License-Identifier: BSD-3-Clause
  
  #
  # Decorators used for the definition of tests
  #
  
  __all__ = [
      'parameterized_test', 'simple_test', 'required_version',
      'require_deps', 'run_before', 'run_after'
  ]
  
  
  import collections
  import inspect
  import sys
  import traceback
  
  import reframe.utility.osext as osext
  import reframe.core.warnings as warn
  import reframe.core.hooks as hooks
  from reframe.core.exceptions import ReframeSyntaxError, SkipTestError, what
  from reframe.core.fixtures import FixtureRegistry
  from reframe.core.logging import getlogger
  from reframe.core.pipeline import RegressionTest
  from reframe.utility.versioning import VersionValidator
  
  
  # NOTE: we should consider renaming this module in 4.0; it practically takes
  # care of the registration and instantiation of the tests.
  
  
  class TestRegistry:
      '''Regression test registry.
  
      The tests are stored in a dictionary where the test class is the key
      and the constructor arguments for the different instantiations of the
      test are stored as the dictionary value as a list of (args, kwargs)
      tuples.
  
      For backward compatibility reasons, the registry also contains a set of
      tests to be skipped. The machinery related to this should be dropped with
      the ``required_version`` decorator.
      '''
  
      def __init__(self):
          self._tests = dict()
          self._skip_tests = set()
  
      @classmethod
      def create(cls, test, *args, **kwargs):
          obj = cls()
          obj.add(test, *args, **kwargs)
          return obj
  
      def add(self, test, *args, **kwargs):
          self._tests.setdefault(test, [])
          self._tests[test].append((args, kwargs))
  
      # FIXME: To drop with the required_version decorator
      def skip(self, test):
          '''Add a test to the skip set.'''
          self._skip_tests.add(test)
  
      def instantiate_all(self, reset_sysenv=0):
          '''Instantiate all the registered tests.
  
          :param reset_sysenv: Reset valid_systems and valid_prog_environs after
              instantiating the tests. Bit 0 resets the valid_systems, bit 1
              resets the valid_prog_environs.
          '''
  
          # We first instantiate the leaf tests and then walk up their
          # dependencies to instantiate all the fixtures. Fixtures can only
          # establish their exact dependencies at instantiation time, so the
          # dependency graph grows dynamically.
  
          leaf_tests = []
          for test, variants in self._tests.items():
              if test in self._skip_tests:
                  continue
  
              for args, kwargs in variants:
                  try:
                      kwargs['reset_sysenv'] = reset_sysenv
                      leaf_tests.append(test(*args, **kwargs))
                  except SkipTestError as e:
                      getlogger().warning(
                          f'skipping test {test.__qualname__!r}: {e}'
                      )
                  except Exception:
                      exc_info = sys.exc_info()
                      getlogger().warning(
                          f"skipping test {test.__qualname__!r}: "
                          f"{what(*exc_info)} "
                          f"(rerun with '-v' for more information)"
                      )
                      getlogger().verbose(traceback.format_exc())
  
          # Instantiate fixtures
  
          # Do a level-order traversal of the fixture registries of all leaf
          # tests, instantiate all fixtures and generate the final set of
          # candidate tests; the leaf tests are consumed at the end of the
          # traversal and all instantiated tests (including fixtures) are stored
          # in `final_tests`.
          final_tests = []
          fixture_registry = FixtureRegistry()
          while leaf_tests:
              tmp_registry = FixtureRegistry()
              while leaf_tests:
                  c = leaf_tests.pop()
                  reg = getattr(c, '_rfm_fixture_registry', None)
                  final_tests.append(c)
                  if reg:
                      tmp_registry.update(reg)
  
              # Instantiate the new fixtures and update the registry
              new_fixtures = tmp_registry.difference(fixture_registry)
              leaf_tests = new_fixtures.instantiate_all()
              fixture_registry.update(new_fixtures)
  
          return final_tests
  
      def __iter__(self):
          '''Iterate over the registered test classes.'''
          return iter(self._tests.keys())
  
      def __contains__(self, test):
          return test in self._tests
  
  
  def _register_test(cls, *args, **kwargs):
      '''Register a test and its construction arguments into the registry.'''
  
      mod = inspect.getmodule(cls)
      if not hasattr(mod, '_rfm_test_registry'):
          mod._rfm_test_registry = TestRegistry.create(cls, *args, **kwargs)
      else:
          mod._rfm_test_registry.add(cls, *args, **kwargs)
  
  
  def _register_parameterized_test(cls, args=None):
      '''Register the test.
  
      Register the test with _rfm_use_params=True. This additional argument flags
      this case to consume the parameter space. Otherwise, the regression test
      parameters would simply be initialized to None.
      '''
      def _instantiate(cls, args):
          if isinstance(args, collections.abc.Sequence):
              return cls(*args)
          elif isinstance(args, collections.abc.Mapping):
              return cls(**args)
          elif args is None:
              return cls()
  
      def _instantiate_all():
          ret = []
          for cls, args in mod.__rfm_test_registry:
              try:
                  if cls in mod.__rfm_skip_tests:
                      continue
              except AttributeError:
                  mod.__rfm_skip_tests = set()
  
              try:
                  ret.append(_instantiate(cls, args))
              except SkipTestError as e:
                  getlogger().warning(f'skipping test {cls.__qualname__!r}: {e}')
              except Exception:
                  exc_info = sys.exc_info()
                  getlogger().warning(
                      f"skipping test {cls.__qualname__!r}: {what(*exc_info)} "
                      f"(rerun with '-v' for more information)"
                  )
                  getlogger().verbose(traceback.format_exc())
  
          return ret
  
      mod = inspect.getmodule(cls)
      if not hasattr(mod, '_rfm_gettests'):
          mod._rfm_gettests = _instantiate_all
  
      try:
          mod.__rfm_test_registry.append((cls, args))
      except AttributeError:
          mod.__rfm_test_registry = [(cls, args)]
  
  
  def _validate_test(cls):
      if not issubclass(cls, RegressionTest):
          raise ReframeSyntaxError('the decorated class must be a '
                                   'subclass of RegressionTest')
  
      if (cls.is_abstract()):
          getlogger().warning(
              f'skipping test {cls.__qualname__!r}: '
              f'test has one or more undefined parameters'
          )
          return False
  
      conditions = [VersionValidator(v) for v in cls._rfm_required_version]
      if (cls._rfm_required_version and
          not any(c.validate(osext.reframe_version()) for c in conditions)):
  
          getlogger().warning(f"skipping incompatible test "
                              f"'{cls.__qualname__}': not valid for ReFrame "
                              f"version {osext.reframe_version().split('-')[0]}")
          return False
  
      return True
  
  
  def simple_test(cls):
      '''Class decorator for registering tests with ReFrame.
  
      The decorated class must derive from
      :class:`reframe.core.pipeline.RegressionTest`. This decorator is also
      available directly under the :mod:`reframe` module.
  
      .. versionadded:: 2.13
      '''
      if _validate_test(cls):
          for n in range(cls.num_variants):
              _register_test(cls, variant_num=n)
  
      return cls
  
  
  def parameterized_test(*inst):
      '''Class decorator for registering multiple instantiations of a test class.
  
     The decorated class must derive from
     :class:`reframe.core.pipeline.RegressionTest`. This decorator is also
     available directly under the :mod:`reframe` module.
  
     :arg inst: The different instantiations of the test. Each instantiation
          argument may be either a sequence or a mapping.
  
     .. versionadded:: 2.13
  
     .. note::
        This decorator does not instantiate any test.  It only registers them.
        The actual instantiation happens during the loading phase of the test.
  
     .. deprecated:: 3.6.0
  
        Please use the :func:`~reframe.core.pipeline.RegressionTest.parameter`
        built-in instead.
  
      '''
  
      warn.user_deprecation_warning(
          'the @parameterized_test decorator is deprecated; '
          'please use the parameter() built-in instead',
          from_version='3.6.0'
      )
  
      def _do_register(cls):
          if _validate_test(cls):
              if not cls.param_space.is_empty():
                  raise ReframeSyntaxError(
                      f'{cls.__qualname__!r} is already a parameterized test'
                  )
  
              for args in inst:
                  _register_parameterized_test(cls, args)
  
          return cls
  
      return _do_register
  
  
  def required_version(*versions):
      '''Class decorator for specifying the required ReFrame versions for the
      following test.
  
      If the test is not compatible with the current ReFrame version it will be
      skipped.
  
      :arg versions: A list of ReFrame version specifications that this test is
        allowed to run. A version specification string can have one of the
        following formats:
  
        1. ``VERSION``: Specifies a single version.
        2. ``{OP}VERSION``, where ``{OP}`` can be any of ``>``, ``>=``, ``<``,
           ``<=``, ``==`` and ``!=``. For example, the version specification
           string ``'>=3.5.0'`` will allow the following test to be loaded only
           by ReFrame 3.5.0 and higher. The ``==VERSION`` specification is the
           equivalent of ``VERSION``.
        3. ``V1..V2``: Specifies a range of versions.
  
        You can specify multiple versions with this decorator, such as
        ``@required_version('3.5.1', '>=3.5.6')``, in which case the test will be
        selected if *any* of the versions is satisfied, even if the versions
        specifications are conflicting.
  
      .. versionadded:: 2.13
  
      .. versionchanged:: 3.5.0
  
         Passing ReFrame version numbers that do not comply with the `semantic
         versioning <https://semver.org/>`__ specification is deprecated.
         Examples of non-compliant version numbers are ``3.5`` and ``3.5-dev0``.
         These should be written as ``3.5.0`` and ``3.5.0-dev.0``.
  
      .. deprecated:: 3.5.0
         Please set the ``require_version`` parameter in the class definition
         instead.
  
      '''
      warn.user_deprecation_warning(
          "the '@required_version' decorator is deprecated; please set "
          "the 'require_version' parameter in the class definition instead",
          from_version='3.7.0'
      )
  
      if not versions:
          raise ReframeSyntaxError('no versions specified')
  
      conditions = [VersionValidator(v) for v in versions]
  
      def _skip_tests(cls):
          mod = inspect.getmodule(cls)
          if not hasattr(mod, '__rfm_skip_tests'):
              mod.__rfm_skip_tests = set()
  
          if not hasattr(mod, '_rfm_test_registry'):
              mod._rfm_test_registry = TestRegistry()
  
          if not any(c.validate(osext.reframe_version()) for c in conditions):
              getlogger().warning(
                  f"skipping incompatible test '{cls.__qualname__}': not valid "
                  f"for ReFrame version {osext.reframe_version().split('-')[0]}"
              )
              if cls in mod._rfm_test_registry:
                  mod._rfm_test_registry.skip(cls)
              else:
                  mod.__rfm_skip_tests.add(cls)
  
          return cls
  
      return _skip_tests
  
  
  # Valid pipeline stages that users can specify in the `run_before()` and
  # `run_after()` decorators
  _USER_PIPELINE_STAGES = (
      'init', 'setup', 'compile', 'run', 'sanity', 'performance', 'cleanup'
  )
  
  
  def run_before(stage):
      '''Decorator for attaching a test method to a pipeline stage.
  
      .. deprecated:: 3.7.0
         Please use the :func:`~reframe.core.pipeline.RegressionMixin.run_before`
         built-in function.
  
      '''
      warn.user_deprecation_warning(
          'using the @rfm.run_before decorator from the rfm module is '
          'deprecated; please use the built-in decorator @run_before instead.',
          from_version='3.7.0'
      )
      if stage not in _USER_PIPELINE_STAGES:
          raise ReframeSyntaxError(
              f'invalid pipeline stage specified: {stage!r}')
  
      if stage == 'init':
          raise ReframeSyntaxError('pre-init hooks are not allowed')
  
      return hooks.attach_to('pre_' + stage)
  
  
  def run_after(stage):
      '''Decorator for attaching a test method to a pipeline stage.
  
      .. deprecated:: 3.7.0
         Please use the :func:`~reframe.core.pipeline.RegressionMixin.run_after`
         built-in function.
  
      '''
      warn.user_deprecation_warning(
          'using the @rfm.run_after decorator from the rfm module is '
          'deprecated; please use the built-in decorator @run_after instead.',
          from_version='3.7.0'
      )
      if stage not in _USER_PIPELINE_STAGES:
          raise ReframeSyntaxError(
              f'invalid pipeline stage specified: {stage!r}')
  
      # Map user stage names to the actual pipeline functions if needed
      if stage == 'init':
          stage = '__init__'
      elif stage == 'compile':
          stage = 'compile_wait'
      elif stage == 'run':
          stage = 'run_wait'
  
      return hooks.attach_to('post_' + stage)
  
  
  def require_deps(fn):
      '''Decorator to denote that a function will use the test dependencies.
  
      .. versionadded:: 2.21
  
      .. deprecated:: 3.7.0
         Please use the
         :func:`~reframe.core.pipeline.RegressionTest.require_deps` built-in
         function.
  
      '''
      warn.user_deprecation_warning(
          'using the @rfm.require_deps decorator from the rfm module is '
          'deprecated; please use the built-in decorator @require_deps instead.',
          from_version='3.7.0'
      )
      return hooks.require_deps(fn)
  