Source code for stressor.plugins.base

# -*- coding: utf-8 -*-
# (c) 2020-2023 Martin Wendt and contributors; see https://github.com/mar10/stressor
# Licensed under the MIT license: https://www.opensource.org/licenses/mit-license.php
"""
"""
import re
from abc import ABC, abstractmethod

from stressor.util import StressorError, assert_always, check_arg, parse_args_from_str

#: (set) all activities accept these arguments
common_args = set(
    (
        "activity",
        "assert_match",
        "assert_max_time",
        "debug",
        "mock_result",
        "monitor",
        "name",
    )
)


[docs]class ActivityError(StressorError): """Base for all errors that are explicitly raised by activities."""
[docs]class ActivityTimeoutError(ActivityError): """Activity timed out."""
[docs]class ActivityCompileError(ActivityError): """Raised when activity constructor fails."""
[docs]class ActivityAssertionError(ActivityError): """Assertion failed (e.g. `assert_match` argument, ...).""" def __init__(self, assertion_list): super().__init__("Activity assertion failed") check_arg(assertion_list, (str, list)) if isinstance(assertion_list, str): assertion_list = [assertion_list] self.assertion_list = assertion_list
[docs]class ScriptActivityError(ActivityError): """Raised when a ScriptActivity fails."""
[docs]class ActivityBase(ABC): """ Common base class for all activities. New activity plugins can be created by deriving from this class, importing it, and then calling :func:`register_activity_plugins()`. Names of derived classes should end with '...Activity'. Classes which names that begin with an underscore ('_') are ignored. """ #: (str) Name by which the actvity can be used in YAML configurations. #: Defaults to class name without trailing `...Activity`, e.g. #: 'SleepActivity' is called as ``activity: Sleep`` _script_name = None #: (set) If defined, the default implementation of `compile()` will raise #: an error if any of those args is not passed. _mandatory_args = None #: (set) If defined, the default implementation of `compile()` will raise #: an error if other args are passed to the constructor. _known_args = None #: (set) Internal use only! _all_known_args = None #: (tuple|None) List of argument names that will be displayed in path #: strings _info_args = None #: (bool) _default_monitor = False #: (bool) _default_ignore_timing = False @abstractmethod def __init__(self, config_manager, **activity_args): """ The constructor is called by the :class:`ConfigManager`, when the configuration was read. The compiler replaces the activity name in the definition's `activity` attribute with this instance. The `execute()` method is then called by the session runner on or more times. The default implementation stores `activity_args` as `self.raw_args`. however derived classes may choose to add named args explicitly for better readability and checking. An implementation should - Check args and raise errors, so we get early load-time failures when the configuration is read and compiled. - Optionally cache some information in this instance that can be re-used in the execute calls. - Optionally call `compiler.add_warning()` Attributes: compile_path_short (str): Path to location of definition in the configuration, e.g. '/main/2/PutRequest' compile_path (str): Path to location of definition in the configuration with added argument infos, e.g. '/main/2/PutRequest($(base_url)/test.html)' raw_args (dict): monitor (bool): ignore_timing (bool): Args: compile_path (:class:`PathStack`): The location breadcrumb path of this activity in the configration structure, e.g. 'main/#4'. activity_args (dict): named arguments that were passed to the activity definition. Note that the arguments are read at load-time and are not yet expanded (i.e. may contain `$(context_var)` macros). """ self.raw_args = activity_args self.compile_path_short = config_manager.stack.get_path( skip_segs=2, last_seg=self.get_script_name() ) self.compile_path = config_manager.stack.get_path( skip_segs=2, last_seg=self.get_info() ) passed = set(activity_args.keys()) if self._mandatory_args: missing = self._mandatory_args.difference(passed) if missing: raise ActivityCompileError( "Missing mandatory arguments: {}".format(missing) ) if self._all_known_args is None: if self._known_args: self._all_known_args = common_args | self._known_args else: self._all_known_args = common_args extra = passed.difference(self._all_known_args) if extra: raise ActivityCompileError("Unsupported arguments: {}".format(extra)) self.monitor = activity_args.get("monitor", self._default_monitor) self.ignore_timing = activity_args.get( "ignore_timing", self._default_ignore_timing )
[docs] def __str__(self): """Return a descriptive string.""" return self.get_info()
#: Provide nicer display for pprint(), etc. __repr__ = __str__
[docs] @classmethod def get_script_name(cls): # Note: we must check `cls.__dict__` instead of `cls._script_name`, # because we want to test the local class attribute (not a derived one): if cls.__dict__.get("_script_name") is None: assert_always(cls.__name__.endswith("Activity")) cls._script_name = cls.__name__[:-8] return cls._script_name
[docs] def get_info(self, info_args=True, expanded_args=None, session=None): """Return a descriptive string (optionally using expanded args). This method can be called with or without the additional context of the current session and expanded argument macros. Args: info_args (tuple|bool): List of argument names that should be added to display string. True: default to `self._info_args` (see `None` by default) False: don't add arguments None: all arguments expanded_args (dict, optional): optional argment dict (defaults to `self.raw_args`) session (SessionManager, optional): Running session context """ if info_args is True: info_args = self._info_args elif info_args is False: info_args = [] arg_dict = self.raw_args if expanded_args is None else expanded_args if info_args: # Add selected args args = ( "{}={!r}".format(a, arg_dict.get(a)) for a in info_args if a in arg_dict ) else: # add all args args = ( "{}={!r}".format(*kv) for kv in arg_dict.items() if kv[0] != "activity" ) return "{}({})".format(self.get_script_name(), ", ".join(args))
[docs] def prepare_execute(self, session, expanded_args): """Allow an activity to prepare the next execution. The session manager calls this for every activity instance, directly before `get_info()` and `execute()`. Normally this method does not need to be implemented. (One exception is `SleepActivity`, that calculates the next random duration per session, so it can be displayed by `get_info(..., session)`.) """ return
[docs] @abstractmethod def execute(self, session, **expanded_args): """ Called by the :class:`SessionManager`, after `$(context.var)` macros have been resolved if any. A derived class MUST implement this method. The session manager calls this methods for every activity instance: 1. prepare_execute() 2. get_info() 3. execute() - Call session.log_warning() - raise ActivityError() for errors that a user can ignore by --force flag - Honor session.stop_request - Honor session.dry_run Args: session (:class:`SessionManager`): expanded_args (dict): current global vars Raises: ActivityError: Exception: all other exceptions are considered a fatal error Returns: nothing """
[docs]class MacroBase(ABC): """ Common base class for all load-time script macros of the form ``$NAME(ARGS)``. These macros are applied at load time to the nested dict that was read from a YAML file. Typically they replace the value of that dict element (``parent[parent_key]``) with new content. For example - ``$load(PATH)`` replaces the element with a parsed YAML content or the text content of a python file. - ``$sleep(DURATION)`` replaces the element with a :class:`SleepActivity` definition Note that context variable expansions of the form ``$(CONTEXT.VAR.NAME)`` are *not* handled by this kind of macro, because this has to be dealt with at run time. TODO: This means that ARGS cannot be dynamic, so for example $load($(my_file_name)) will not work. Custom macro plugins can be defined by deriving from this class, importing it, and then calling :func:`register_macro_plugins()`. """ #: Name for use in scripts. #: Defaults to lowercase class name without trailing `...Macro`, e.g. #: 'LoadMacro' is exposed as ``$load(...)`` _script_name = None #: Regular expression that extracts '$NAME(ARGS)' #: Defaults to ... _regex = None #: List of tuples that define the supported positional arguments and #: optional default values. #: See :func:`stressor.util.parse_args_from_str` for syntax. #: If `None`, the raw cotent inside the brackets is passed as single string . _args_def = None #: Allow late evaluation (e.g. $stamp) #: TODO: Not yet implemented run_time_eval = False def __init__(self, **macro_args): """""" return
[docs] @classmethod def get_script_name(cls): # Note: we must check `cls.__dict__` instead of `cls._script_name`, # because we want to test the local class attribute (not a derived one): if cls.__dict__.get("_script_name") is None: assert_always(cls.__name__.endswith("Macro")) cls._script_name = cls.__name__[:-5].lower() return cls._script_name
[docs] @classmethod def get_regex(cls): if cls._regex is None: name = cls.get_script_name() cls._regex = re.compile(r"\s*\$" + name + r"\((.*)\)\s*") return cls._regex
[docs] def match_apply(self, context_reader, parent, parent_key): """Parse current value and call self.apply() if the macro matches. This default implementation supports Returns: (bool, any) (handled, ) """ pattern = self.get_regex() value = parent[parent_key] match = pattern.match(value) if not match: return (False, None) args = match.groups() assert len(args) == 1 arg_str = args[0] if self._args_def: kwargs = parse_args_from_str(arg_str, self._args_def) res = self.apply(context_reader, parent, parent_key, **kwargs) else: res = self.apply(context_reader, parent, parent_key, arg_str) # logger.debug("Eval {}: {} => {}".format(stack, value, res)) return (True, res)
[docs] @abstractmethod def apply(self, context_reader, parent, parent_key, args_str): """ Args: context_reader (:class:`ConfigManager`): parent (dict|list): parent_key (str|int): args_str (str|**kwargs): Returns: (any) The result that was produced (and stored into `parent[parent_key]`) """