From 2b2fc7721f4bc0c05c7384170ddc82278aad5875 Mon Sep 17 00:00:00 2001 From: Tim Pillinger <26465611+wxtim@users.noreply.github.com> Date: Thu, 22 Feb 2024 08:17:53 +0000 Subject: [PATCH] Xtrig arg validate (#5955) * Xtrigger function arg validation. * Add integration tests for xtrigger validation, which provides simple examples for each of the built in xtriggers. - Xrandom validate function - Init test xrandom validate function - Add unit tests for validation of built in xtriggers - Automatically validate xtrigger function signature --------- Co-authored-by: Hilary James Oliver * Apply suggestions from code review Co-authored-by: Ronnie Dutta <61982285+MetRonnie@users.noreply.github.com> * fix flake8 * Improve xtrigger validation (#60) * Improve xtrigger validation * wall_clock: use placeholder function for signature validation & autodocs * Fix docstring for autodoc [skip ci] --------- Co-authored-by: Hilary Oliver Co-authored-by: Ronnie Dutta <61982285+MetRonnie@users.noreply.github.com> --- changes.d/5955.feat.md | 1 + cylc/flow/config.py | 42 ++-- cylc/flow/exceptions.py | 7 +- cylc/flow/scripts/function_run.py | 11 +- cylc/flow/subprocpool.py | 77 +++++--- cylc/flow/xtrigger_mgr.py | 93 ++++++--- cylc/flow/xtriggers/echo.py | 45 ++++- cylc/flow/xtriggers/wall_clock.py | 46 ++++- cylc/flow/xtriggers/xrandom.py | 55 ++++-- setup.cfg | 14 +- tests/functional/runahead/04-no-final-cycle.t | 2 +- tests/functional/runahead/no_final/flow.cylc | 6 +- .../spawn-on-demand/15-stop-flow-3/flow.cylc | 4 +- .../35-pass-special-tasks-non-word-names.t | 40 ---- tests/functional/xtriggers/03-sequence.t | 4 +- tests/integration/test_config.py | 179 +++++++++++++++++- tests/integration/test_xtrigger_mgr.py | 1 + tests/unit/test_subprocpool.py | 23 +-- tests/unit/test_xtrigger_mgr.py | 19 +- .../xtriggers/test_echo.py} | 32 ++-- tests/unit/xtriggers/test_wall_clock.py | 61 ++++++ tests/unit/xtriggers/test_xrandom.py | 39 ++++ 22 files changed, 602 insertions(+), 199 deletions(-) create mode 100644 changes.d/5955.feat.md delete mode 100755 tests/functional/validate/35-pass-special-tasks-non-word-names.t rename tests/{functional/validate/69-bare-clock-xtrigger.t => unit/xtriggers/test_echo.py} (55%) create mode 100644 tests/unit/xtriggers/test_wall_clock.py create mode 100644 tests/unit/xtriggers/test_xrandom.py diff --git a/changes.d/5955.feat.md b/changes.d/5955.feat.md new file mode 100644 index 00000000000..c9fc9b4d879 --- /dev/null +++ b/changes.d/5955.feat.md @@ -0,0 +1 @@ +Support xtrigger argument validation. diff --git a/cylc/flow/config.py b/cylc/flow/config.py index 76c8b4e1980..f871531adeb 100644 --- a/cylc/flow/config.py +++ b/cylc/flow/config.py @@ -57,6 +57,7 @@ from cylc.flow.id import Tokens from cylc.flow.cycling.integer import IntegerInterval from cylc.flow.cycling.iso8601 import ingest_time, ISO8601Interval + from cylc.flow.exceptions import ( CylcError, InputError, @@ -1718,28 +1719,25 @@ def generate_triggers(self, lexpression, left_nodes, right, seq, if label != 'wall_clock': raise WorkflowConfigError(f"xtrigger not defined: {label}") else: - # Allow "@wall_clock" in the graph as an undeclared - # zero-offset clock xtrigger. - xtrig = SubFuncContext( - 'wall_clock', 'wall_clock', [], {}) + # Allow "@wall_clock" in graph as implicit zero-offset. + xtrig = SubFuncContext('wall_clock', 'wall_clock', [], {}) - if xtrig.func_name == 'wall_clock': - if self.cycling_type == INTEGER_CYCLING_TYPE: - raise WorkflowConfigError( - "Clock xtriggers require datetime cycling:" - f" {label} = {xtrig.get_signature()}" - ) - else: - # Convert offset arg to kwarg for certainty later. - if "offset" not in xtrig.func_kwargs: - xtrig.func_kwargs["offset"] = None - with suppress(IndexError): - xtrig.func_kwargs["offset"] = xtrig.func_args[0] + if ( + xtrig.func_name == 'wall_clock' + and self.cycling_type == INTEGER_CYCLING_TYPE + ): + raise WorkflowConfigError( + "Clock xtriggers require datetime cycling:" + f" {label} = {xtrig.get_signature()}" + ) - if self.xtrigger_mgr is None: - XtriggerManager.validate_xtrigger(label, xtrig, self.fdir) - else: + # Generic xtrigger validation. + XtriggerManager.check_xtrigger(label, xtrig, self.fdir) + + if self.xtrigger_mgr: + # (not available during validation) self.xtrigger_mgr.add_trig(label, xtrig, self.fdir) + self.taskdefs[right].add_xtrig_label(label, seq) def get_actual_first_point(self, start_point): @@ -2427,10 +2425,10 @@ def upgrade_clock_triggers(self): # Derive an xtrigger label. label = '_'.join(('_cylc', 'wall_clock', task_name)) # Define the xtrigger function. - xtrig = SubFuncContext(label, 'wall_clock', [], {}) - xtrig.func_kwargs["offset"] = offset + args = [] if offset is None else [offset] + xtrig = SubFuncContext(label, 'wall_clock', args, {}) if self.xtrigger_mgr is None: - XtriggerManager.validate_xtrigger(label, xtrig, self.fdir) + XtriggerManager.check_xtrigger(label, xtrig, self.fdir) else: self.xtrigger_mgr.add_trig(label, xtrig, self.fdir) # Add it to the task, for each sequence that the task appears in. diff --git a/cylc/flow/exceptions.py b/cylc/flow/exceptions.py index 43591e7be9c..3ad8c093757 100644 --- a/cylc/flow/exceptions.py +++ b/cylc/flow/exceptions.py @@ -241,13 +241,12 @@ class XtriggerConfigError(WorkflowConfigError): """ - def __init__(self, label: str, trigger: str, message: str): + def __init__(self, label: str, message: str): self.label: str = label - self.trigger: str = trigger self.message: str = message - def __str__(self): - return f'[{self.label}] {self.message}' + def __str__(self) -> str: + return f'[@{self.label}] {self.message}' class ClientError(CylcError): diff --git a/cylc/flow/scripts/function_run.py b/cylc/flow/scripts/function_run.py index 7ada8805de3..63029a97782 100755 --- a/cylc/flow/scripts/function_run.py +++ b/cylc/flow/scripts/function_run.py @@ -18,10 +18,13 @@ (This command is for internal use.) -Run a Python function "(*args, **kwargs)" in the process pool. It must be -defined in a module of the same name. Positional and keyword arguments must be -passed in as JSON strings. is the workflow source dir, needed to find -local xtrigger modules. +Run a Python xtrigger function "(*args, **kwargs)" in the process pool. +It must be in a module of the same name. Positional and keyword arguments must +be passed in as JSON strings. + +Python entry points are the preferred way to make xtriggers available to the +scheduler, but local xtriggers can be stored in . + """ import sys diff --git a/cylc/flow/subprocpool.py b/cylc/flow/subprocpool.py index 9ac3845bb54..1a49eed680d 100644 --- a/cylc/flow/subprocpool.py +++ b/cylc/flow/subprocpool.py @@ -41,7 +41,8 @@ from cylc.flow.task_proxy import TaskProxy from cylc.flow.wallclock import get_current_time_string -_XTRIG_FUNCS: dict = {} +_XTRIG_MOD_CACHE: dict = {} +_XTRIG_FUNC_CACHE: dict = {} def _killpg(proc, signal): @@ -66,46 +67,58 @@ def _killpg(proc, signal): return True -def get_func(func_name, src_dir): - """Find and return an xtrigger function from a module of the same name. +def get_xtrig_mod(mod_name, src_dir): + """Find, cache, and return a named xtrigger module. - These locations are checked in this order: - - /lib/python/ - - `$CYLC_PYTHONPATH` - - defined via a `cylc.xtriggers` entry point for an - installed Python package. + Locations checked in this order: + - /lib/python (prepend to sys.path) + - $CYLC_PYTHONPATH (already in sys.path) + - `cylc.xtriggers` entry point - Workflow source directory passed in as this is executed in an independent - process in the command pool and therefore doesn't know about the workflow. + (Check entry point last so users can override with local implementations). + + Workflow source dir passed in - this executes in an independent subprocess. + + Raises: + ImportError, if the module is not found """ - if func_name in _XTRIG_FUNCS: - return _XTRIG_FUNCS[func_name] + if mod_name in _XTRIG_MOD_CACHE: + # Found and cached already. + return _XTRIG_MOD_CACHE[mod_name] # First look in /lib/python. sys.path.insert(0, os.path.join(src_dir, 'lib', 'python')) - mod_name = func_name try: - mod_by_name = __import__(mod_name, fromlist=[mod_name]) + _XTRIG_MOD_CACHE[mod_name] = __import__(mod_name, fromlist=[mod_name]) except ImportError: - # Look for xtriggers via entry_points for external sources. - # Do this after the lib/python and PYTHONPATH approaches to allow - # users to override entry_point definitions with local/custom - # implementations. + # Then entry point. for entry_point in iter_entry_points('cylc.xtriggers'): - if func_name == entry_point.name: - _XTRIG_FUNCS[func_name] = entry_point.load() - return _XTRIG_FUNCS[func_name] - + if mod_name == entry_point.name: + _XTRIG_MOD_CACHE[mod_name] = entry_point.load() + return _XTRIG_MOD_CACHE[mod_name] # Still unable to find anything so abort raise - try: - _XTRIG_FUNCS[func_name] = getattr(mod_by_name, func_name) - except AttributeError: - # Module func_name has no function func_name, nor an entry_point entry. - raise - return _XTRIG_FUNCS[func_name] + return _XTRIG_MOD_CACHE[mod_name] + + +def get_xtrig_func(mod_name, func_name, src_dir): + """Find, cache, and return a function from an xtrigger module. + + Raises: + ImportError, if the module is not found + AttributeError, if the function is not found in the module + + """ + if (mod_name, func_name) in _XTRIG_FUNC_CACHE: + return _XTRIG_FUNC_CACHE[(mod_name, func_name)] + + mod = get_xtrig_mod(mod_name, src_dir) + + _XTRIG_FUNC_CACHE[(mod_name, func_name)] = getattr(mod, func_name) + + return _XTRIG_FUNC_CACHE[(mod_name, func_name)] def run_function(func_name, json_args, json_kwargs, src_dir): @@ -113,6 +126,8 @@ def run_function(func_name, json_args, json_kwargs, src_dir): func_name(*func_args, **func_kwargs) + The function is presumed to be in a module of the same name. + Redirect any function stdout to stderr (and workflow log in debug mode). Return value printed to stdout as a JSON string - allows use of the existing process pool machinery as-is. src_dir is for local modules. @@ -120,14 +135,18 @@ def run_function(func_name, json_args, json_kwargs, src_dir): """ func_args = json.loads(json_args) func_kwargs = json.loads(json_kwargs) + # Find and import then function. - func = get_func(func_name, src_dir) + func = get_xtrig_func(func_name, func_name, src_dir) + # Redirect stdout to stderr. orig_stdout = sys.stdout sys.stdout = sys.stderr res = func(*func_args, **func_kwargs) + # Restore stdout. sys.stdout = orig_stdout + # Write function return value as JSON to stdout. sys.stdout.write(json.dumps(res)) diff --git a/cylc/flow/xtrigger_mgr.py b/cylc/flow/xtrigger_mgr.py index 6ced0023a56..b891165dbd3 100644 --- a/cylc/flow/xtrigger_mgr.py +++ b/cylc/flow/xtrigger_mgr.py @@ -16,6 +16,7 @@ from contextlib import suppress from enum import Enum +from inspect import signature import json import re from copy import deepcopy @@ -26,10 +27,11 @@ from cylc.flow.exceptions import XtriggerConfigError import cylc.flow.flags from cylc.flow.hostuserutil import get_user -from cylc.flow.subprocpool import get_func -from cylc.flow.xtriggers.wall_clock import wall_clock +from cylc.flow.subprocpool import get_xtrig_func +from cylc.flow.xtriggers.wall_clock import _wall_clock if TYPE_CHECKING: + from inspect import BoundArguments from cylc.flow.broadcast_mgr import BroadcastMgr from cylc.flow.data_store_mgr import DataStoreMgr from cylc.flow.subprocctx import SubFuncContext @@ -240,12 +242,16 @@ def __init__( self.do_housekeeping = False @staticmethod - def validate_xtrigger( + def check_xtrigger( label: str, fctx: 'SubFuncContext', fdir: str, ) -> None: - """Validate an Xtrigger function. + """Generic xtrigger validation: check existence, string templates and + function signature. + + Xtrigger modules may also supply a specific `validate` function + which will be run here. Args: label: xtrigger label @@ -259,29 +265,39 @@ def validate_xtrigger( * If the function is not callable. * If any string template in the function context arguments are not present in the expected template values. + * If the arguments do not match the function signature. """ fname: str = fctx.func_name + try: - func = get_func(fname, fdir) + func = get_xtrig_func(fname, fname, fdir) except ImportError: raise XtriggerConfigError( - label, - fname, - f"xtrigger module '{fname}' not found", + label, f"xtrigger module '{fname}' not found", ) except AttributeError: raise XtriggerConfigError( - label, - fname, - f"'{fname}' not found in xtrigger module '{fname}'", + label, f"'{fname}' not found in xtrigger module '{fname}'", ) + if not callable(func): raise XtriggerConfigError( - label, - fname, - f"'{fname}' not callable in xtrigger module '{fname}'", + label, f"'{fname}' not callable in xtrigger module '{fname}'", + ) + + # Validate args and kwargs against the function signature + sig_str = fctx.get_signature() + try: + bound_args = signature(func).bind( + *fctx.func_args, **fctx.func_kwargs ) + except TypeError as exc: + raise XtriggerConfigError(label, f"{sig_str}: {exc}") + # Specific xtrigger.validate(), if available. + XtriggerManager.try_xtrig_validate_func( + label, fname, fdir, bound_args, sig_str + ) # Check any string templates in the function arg values (note this # won't catch bad task-specific values - which are added dynamically). @@ -297,9 +313,7 @@ def validate_xtrigger( template_vars.add(TemplateVariables(match)) except ValueError: raise XtriggerConfigError( - label, - fname, - f"Illegal template in xtrigger: {match}", + label, f"Illegal template in xtrigger: {match}", ) # check for deprecated template variables @@ -315,16 +329,41 @@ def validate_xtrigger( f' {", ".join(t.value for t in deprecated_variables)}' ) + @staticmethod + def try_xtrig_validate_func( + label: str, + fname: str, + fdir: str, + bound_args: 'BoundArguments', + signature_str: str, + ): + """Call an xtrigger's `validate()` function if it is implemented. + + Raise XtriggerConfigError if validation fails. + """ + try: + xtrig_validate_func = get_xtrig_func(fname, 'validate', fdir) + except (AttributeError, ImportError): + return + bound_args.apply_defaults() + try: + xtrig_validate_func(bound_args.arguments) + except Exception as exc: # Note: catch all errors + raise XtriggerConfigError( + label, f"{signature_str} validation failed: {exc}" + ) + def add_trig(self, label: str, fctx: 'SubFuncContext', fdir: str) -> None: """Add a new xtrigger function. - Check the xtrigger function exists here (e.g. during validation). + Call check_xtrigger before this, during validation. + Args: label: xtrigger label fctx: function context fdir: function module directory + """ - self.validate_xtrigger(label, fctx, fdir) self.functx_map[label] = fctx def mutate_trig(self, label, kwargs): @@ -395,20 +434,18 @@ def get_xtrig_ctx( args = [] kwargs = {} if ctx.func_name == "wall_clock": - if "trigger_time" in ctx.func_kwargs: + if "trigger_time" in ctx.func_kwargs: # noqa: SIM401 (readabilty) # Internal (retry timer): trigger_time already set. kwargs["trigger_time"] = ctx.func_kwargs["trigger_time"] - elif "offset" in ctx.func_kwargs: # noqa: SIM106 + else: # External (clock xtrigger): convert offset to trigger_time. # Datetime cycling only. kwargs["trigger_time"] = itask.get_clock_trigger_time( itask.point, - ctx.func_kwargs["offset"] - ) - else: - # Should not happen! - raise ValueError( - "wall_clock function kwargs needs trigger time or offset" + ctx.func_kwargs.get( + "offset", + ctx.func_args[0] if ctx.func_args else None + ) ) else: # Other xtrig functions: substitute template values. @@ -440,7 +477,7 @@ def call_xtriggers_async(self, itask: 'TaskProxy'): if sig in self.sat_xtrig: # Already satisfied, just update the task itask.state.xtriggers[label] = True - elif wall_clock(*ctx.func_args, **ctx.func_kwargs): + elif _wall_clock(*ctx.func_args, **ctx.func_kwargs): # Newly satisfied itask.state.xtriggers[label] = True self.sat_xtrig[sig] = {} diff --git a/cylc/flow/xtriggers/echo.py b/cylc/flow/xtriggers/echo.py index e3c5880cab0..d200e2389bb 100644 --- a/cylc/flow/xtriggers/echo.py +++ b/cylc/flow/xtriggers/echo.py @@ -16,22 +16,49 @@ """A Cylc xtrigger function.""" -from contextlib import suppress +from typing import Any, Dict, Tuple +from cylc.flow.exceptions import WorkflowConfigError -def echo(*args, **kwargs): - """Prints args to stdout and return success only if kwargs['succeed'] - is True. +def echo(*args, **kwargs) -> Tuple: + """Print arguments to stdout, return kwargs['succeed'] and kwargs. This may be a useful aid to understanding how xtriggers work. + Args: + succeed: Set the success of failure of this xtrigger. + *args: Print to stdout. + **kwargs: Print to stdout, and return as output. + + Examples: + + >>> echo('Breakfast Time', succeed=True, egg='poached') + echo: ARGS: ('Breakfast Time',) + echo: KWARGS: {'succeed': True, 'egg': 'poached'} + (True, {'succeed': True, 'egg': 'poached'}) + Returns - tuple: (True/False, kwargs) + (True/False, kwargs) """ print("echo: ARGS:", args) print("echo: KWARGS:", kwargs) - result = False - with suppress(KeyError): - result = kwargs["succeed"] is True - return result, kwargs + + return kwargs["succeed"], kwargs + + +def validate(all_args: Dict[str, Any]): + """ + Validate the xtrigger function arguments parsed from the workflow config. + + This is separate from the xtrigger to allow parse-time validation. + + """ + # NOTE: with (*args, **kwargs) pattern, all_args looks like: + # { + # 'args': (arg1, arg2, ...), + # 'kwargs': {kwarg1: val, kwarg2: val, ...} + # } + succeed = all_args['kwargs'].get("succeed") + if not isinstance(succeed, bool): + raise WorkflowConfigError("Requires 'succeed=True/False' arg") diff --git a/cylc/flow/xtriggers/wall_clock.py b/cylc/flow/xtriggers/wall_clock.py index 3c50d956f0a..d5d1a009154 100644 --- a/cylc/flow/xtriggers/wall_clock.py +++ b/cylc/flow/xtriggers/wall_clock.py @@ -17,13 +17,53 @@ """xtrigger function to trigger off of a wall clock time.""" from time import time +from typing import Any, Dict +from cylc.flow.cycling.iso8601 import interval_parse +from cylc.flow.exceptions import WorkflowConfigError -def wall_clock(trigger_time=None): - """Return True after the desired wall clock time, False. +def wall_clock(offset: str = 'PT0S'): + """Trigger at a specific real "wall clock" time relative to the cycle point + in the graph. + + Clock triggers, unlike other trigger functions, are executed synchronously + in the main process. + + Args: + offset: + ISO 8601 interval to wait after the cycle point is reached in real + time before triggering. May be negative, in which case it will + trigger before the real time reaches the cycle point. + """ + # NOTE: This is just a placeholder for the actual implementation. + # This is only used for validating the signature and for autodocs. + ... + + +def _wall_clock(trigger_time: int) -> bool: + """Actual implementation of wall_clock. + + Return True after the desired wall clock time, or False before. Args: - trigger_time (int): + trigger_time: Trigger time as seconds since Unix epoch. """ return time() > trigger_time + + +def validate(args: Dict[str, Any]): + """Validate and manipulate args parsed from the workflow config. + + NOTE: the xtrigger signature is different to the function signature above + + wall_clock() # infer zero interval + wall_clock(PT1H) + wall_clock(offset=PT1H) + + The offset must be a valid ISO 8601 interval. + """ + try: + interval_parse(args["offset"]) + except (ValueError, AttributeError): + raise WorkflowConfigError(f"Invalid offset: {args['offset']}") diff --git a/cylc/flow/xtriggers/xrandom.py b/cylc/flow/xtriggers/xrandom.py index 53d98865ef1..867035d46f8 100644 --- a/cylc/flow/xtriggers/xrandom.py +++ b/cylc/flow/xtriggers/xrandom.py @@ -16,39 +16,46 @@ from random import random, randint from time import sleep +from typing import Any, Dict, Tuple + +from cylc.flow.exceptions import WorkflowConfigError COLORS = ["red", "orange", "yellow", "green", "blue", "indigo", "violet"] SIZES = ["tiny", "small", "medium", "large", "huge", "humongous"] -def xrandom(percent, secs=0, _=None): +def xrandom( + percent: float, secs: int = 0, _: Any = None +) -> Tuple[bool, Dict[str, str]]: """Random xtrigger, with configurable sleep and percent success. - Sleep for ``sec`` seconds, and report satisfied with ~``percent`` + Sleep for ``sec`` seconds, and report satisfied with ``percent`` likelihood. The ``_`` argument is not used in the function code, but can be used to specialize the function signature to cycle point or task. Args: - percent (float): + percent: Percent likelihood of passing. - secs (int): + secs: Seconds to sleep before starting the trigger. - _ (object): + _: Used to allow users to specialize the trigger with extra parameters. Examples: If the percent is zero, it returns that the trigger condition was not satisfied, and an empty dictionary. + >>> xrandom(0, 0) (False, {}) If the percent is not zero, but the random percent success is not met, then it also returns that the trigger condition was not satisfied, and an empty dictionary. + >>> import sys >>> mocked_random = lambda: 0.3 >>> sys.modules[__name__].random = mocked_random @@ -57,7 +64,8 @@ def xrandom(percent, secs=0, _=None): Finally, if the percent is not zero, and the random percent success is met, then it returns that the trigger condition was satisfied, and a - dictionary containing random color and size as result. + dictionary containing random colour and size as result. + >>> import sys >>> mocked_random = lambda: 0.9 >>> sys.modules[__name__].random = mocked_random @@ -69,23 +77,44 @@ def xrandom(percent, secs=0, _=None): Returns: tuple: (satisfied, results) - satisfied (bool): + satisfied: True if ``satisfied`` else ``False``. - results (dict): + results: A dictionary containing the following keys: ``COLOR`` - A random color (e.g. red, orange, ...) + A random colour (e.g. red, orange, ...). ``SIZE`` - A random size (e.g. small, medium, ...) as the result. + A random size (e.g. small, medium, ...). """ sleep(float(secs)) results = {} - satisfied = random() < float(percent) / 100 # nosec + satisfied = random() < float(percent) / 100 # nosec: B311 if satisfied: results = { - 'COLOR': COLORS[randint(0, len(COLORS) - 1)], # nosec - 'SIZE': SIZES[randint(0, len(SIZES) - 1)] # nosec + 'COLOR': COLORS[randint(0, len(COLORS) - 1)], # nosec: B311 + 'SIZE': SIZES[randint(0, len(SIZES) - 1)] # nosec: B311 } return satisfied, results + + +def validate(args: Dict[str, Any]): + """Validate and manipulate args parsed from the workflow config. + + The rules for args are: + * percent: Must be 0 ≤ x ≤ 100 + * secs: Must be an integer. + """ + percent = args['percent'] + if ( + not isinstance(percent, (float, int)) + or not (0 <= percent <= 100) + ): + raise WorkflowConfigError( + "'percent' should be a float between 0 and 100" + ) + + secs = args['secs'] + if not isinstance(secs, int): + raise WorkflowConfigError("'secs' should be an integer") diff --git a/setup.cfg b/setup.cfg index 849f4ae71b8..bb90ebd1ed6 100644 --- a/setup.cfg +++ b/setup.cfg @@ -215,13 +215,15 @@ cylc.main_loop = cylc.pre_configure = cylc.post_install = log_vc_info = cylc.flow.install_plugins.log_vc_info:main -# NOTE: Built-in xtriggers +# NOTE: Built-in xtrigger modules +# - must contain a function (the xtrigger) with the same name as the module +# - and may contain a "validate" function to check arguments cylc.xtriggers = - echo = cylc.flow.xtriggers.echo:echo - wall_clock = cylc.flow.xtriggers.wall_clock:wall_clock - workflow_state = cylc.flow.xtriggers.workflow_state:workflow_state - suite_state = cylc.flow.xtriggers.suite_state:suite_state - xrandom = cylc.flow.xtriggers.xrandom:xrandom + echo = cylc.flow.xtriggers.echo + wall_clock = cylc.flow.xtriggers.wall_clock + workflow_state = cylc.flow.xtriggers.workflow_state + suite_state = cylc.flow.xtriggers.suite_state + xrandom = cylc.flow.xtriggers.xrandom [bdist_rpm] requires = diff --git a/tests/functional/runahead/04-no-final-cycle.t b/tests/functional/runahead/04-no-final-cycle.t index 1c9609033df..ae1ecebc500 100644 --- a/tests/functional/runahead/04-no-final-cycle.t +++ b/tests/functional/runahead/04-no-final-cycle.t @@ -32,7 +32,7 @@ TEST_NAME=${TEST_NAME_BASE}-check-fail DB="$RUN_DIR/${WORKFLOW_NAME}/log/db" TASKS=$(sqlite3 "${DB}" 'select count(*) from task_states where status=="failed"') # manual comparison for the test -if ((TASKS==4)); then +if ((TASKS==2)); then ok "${TEST_NAME}" else fail "${TEST_NAME}" diff --git a/tests/functional/runahead/no_final/flow.cylc b/tests/functional/runahead/no_final/flow.cylc index 89e23f7af1c..3dd1d7e9734 100644 --- a/tests/functional/runahead/no_final/flow.cylc +++ b/tests/functional/runahead/no_final/flow.cylc @@ -1,11 +1,13 @@ -# Should shutdown stalled with 4 failed tasks. +# Should shutdown stalled with 2 failed tasks due to P1 runahead limit [scheduler] cycle point time zone = Z [[events]] + abort on stall timeout = True + stall timeout = PT0S abort on inactivity timeout = True inactivity timeout = PT20S [scheduling] - runahead limit = P3 + runahead limit = P1 initial cycle point = 20100101T00 [[xtriggers]] never = wall_clock(P100Y) diff --git a/tests/functional/spawn-on-demand/15-stop-flow-3/flow.cylc b/tests/functional/spawn-on-demand/15-stop-flow-3/flow.cylc index fb796f784f5..4731fee49ea 100644 --- a/tests/functional/spawn-on-demand/15-stop-flow-3/flow.cylc +++ b/tests/functional/spawn-on-demand/15-stop-flow-3/flow.cylc @@ -12,11 +12,11 @@ abort on inactivity timeout = True [scheduling] [[xtriggers]] - x = xrandom("0", "10") # never succeeds + x = xrandom(0, 10) # never succeeds [[graph]] R1 = """ @x => never - c => never_again + c => never_again """ [runtime] [[c]] diff --git a/tests/functional/validate/35-pass-special-tasks-non-word-names.t b/tests/functional/validate/35-pass-special-tasks-non-word-names.t deleted file mode 100755 index c523ea25b78..00000000000 --- a/tests/functional/validate/35-pass-special-tasks-non-word-names.t +++ /dev/null @@ -1,40 +0,0 @@ -#!/usr/bin/env bash -# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE. -# Copyright (C) NIWA & British Crown (Met Office) & Contributors. -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU General Public License as published by -# the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program. If not, see . -#------------------------------------------------------------------------------- -# Test validation of special tasks names with non-word characters -. "$(dirname "$0")/test_header" -set_test_number 1 -cat >'flow.cylc' <<'__FLOW_CONFIG__' -[scheduling] - initial cycle point = 20200202 - final cycle point = 20300303 - [[special tasks]] - clock-trigger = t-1, t+1, t%1, t@1 - [[graph]] - P1D = """ - t-1 - t+1 - t%1 - t@1 - """ - -[runtime] - [[t-1, t+1, t%1, t@1]] - script = true -__FLOW_CONFIG__ -run_ok "${TEST_NAME_BASE}" cylc validate "${PWD}" -exit diff --git a/tests/functional/xtriggers/03-sequence.t b/tests/functional/xtriggers/03-sequence.t index f45af9b1caf..63c360f66c5 100644 --- a/tests/functional/xtriggers/03-sequence.t +++ b/tests/functional/xtriggers/03-sequence.t @@ -31,7 +31,7 @@ init_workflow "${TEST_NAME_BASE}" << '__FLOW_CONFIG__' final cycle point = +P1Y [[xtriggers]] e1 = echo(name='bob', succeed=True) - e2 = echo(name='alice') + e2 = echo(name='alice', succeed=False) [[dependencies]] [[[R1]]] graph = "start" @@ -55,7 +55,7 @@ cylc show "${WORKFLOW_NAME}//2026/foo" | grep -E '^ - xtrigger' > 2026.foo.log # 2026/foo should get only xtrigger e2. cmp_ok 2026.foo.log - <<__END__ - - xtrigger "e2 = echo(name=alice)" + - xtrigger "e2 = echo(name=alice, succeed=False)" __END__ cylc stop --now --max-polls=10 --interval=2 "${WORKFLOW_NAME}" diff --git a/tests/integration/test_config.py b/tests/integration/test_config.py index acf24d17eaf..581ec4d83fb 100644 --- a/tests/integration/test_config.py +++ b/tests/integration/test_config.py @@ -16,12 +16,19 @@ from pathlib import Path import sqlite3 +from typing import Any import pytest -from cylc.flow.exceptions import ServiceFileError, WorkflowConfigError +from cylc.flow.exceptions import ( + ServiceFileError, + WorkflowConfigError, + XtriggerConfigError, +) from cylc.flow.parsec.exceptions import ListValueError from cylc.flow.pathutil import get_workflow_run_pub_db_path +Fixture = Any + @pytest.mark.parametrize( 'task_name,valid', [ @@ -341,3 +348,173 @@ def test_validate_incompatible_db(one_conf, flow, validate): finally: conn.close() assert tables == ['suite_params'] + + +def test_xtrig_validation_wall_clock( + flow: 'Fixture', + validate: 'Fixture', +): + """If an xtrigger module has a `validate()` function is called. + + https://github.com/cylc/cylc-flow/issues/5448 + """ + id_ = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'initial cycle point': '1012', + 'xtriggers': {'myxt': 'wall_clock(offset=PT7MH)'}, + 'graph': {'R1': '@myxt => foo'}, + } + }) + with pytest.raises(WorkflowConfigError, match=( + r'\[@myxt\] wall_clock\(offset=PT7MH\) validation failed: ' + r'Invalid offset: PT7MH' + )): + validate(id_) + + +def test_xtrig_implicit_wall_clock(flow: Fixture, validate: Fixture): + """Test @wall_clock is allowed in graph without explicit + xtrigger definition. + """ + wid = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'initial cycle point': '2024', + 'graph': {'R1': '@wall_clock => foo'}, + } + }) + validate(wid) + + +def test_xtrig_validation_echo( + flow: 'Fixture', + validate: 'Fixture', +): + """If an xtrigger module has a `validate()` function is called. + + https://github.com/cylc/cylc-flow/issues/5448 + """ + id_ = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'xtriggers': {'myxt': 'echo()'}, + 'graph': {'R1': '@myxt => foo'}, + } + }) + with pytest.raises( + WorkflowConfigError, + match=r'echo.* Requires \'succeed=True/False\' arg' + ): + validate(id_) + + +def test_xtrig_validation_xrandom( + flow: 'Fixture', + validate: 'Fixture', +): + """If an xtrigger module has a `validate()` function it is called. + + https://github.com/cylc/cylc-flow/issues/5448 + """ + id_ = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'xtriggers': {'myxt': 'xrandom(200)'}, + 'graph': {'R1': '@myxt => foo'}, + } + }) + with pytest.raises( + XtriggerConfigError, + match=r"'percent' should be a float between 0 and 100" + ): + validate(id_) + + +def test_xtrig_validation_custom( + flow: 'Fixture', + validate: 'Fixture', + monkeypatch: 'Fixture', +): + """If an xtrigger module has a `validate()` function + an exception is raised if that validate function fails. + + https://github.com/cylc/cylc-flow/issues/5448 + """ + # Rather than create our own xtrigger module on disk + # and attempt to trigger a validation failure we + # mock our own exception, xtrigger and xtrigger + # validation functions and inject these into the + # appropriate locations: + def kustom_xt(feature): + return True, {} + + def kustom_validate(args): + raise Exception('This is only a test.') + + # Patch xtrigger func & its validate func + monkeypatch.setattr( + 'cylc.flow.xtrigger_mgr.get_xtrig_func', + lambda *args: kustom_validate if "validate" in args else kustom_xt + ) + + id_ = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'initial cycle point': '1012', + 'xtriggers': {'myxt': 'kustom_xt(feature=42)'}, + 'graph': {'R1': '@myxt => foo'}, + } + }) + + Path(id_) + with pytest.raises(XtriggerConfigError, match=r'This is only a test.'): + validate(id_) + + +@pytest.mark.parametrize('xtrig_call, expected_msg', [ + pytest.param( + 'xrandom()', + r"xrandom.* missing a required argument: 'percent'", + id="missing-arg" + ), + pytest.param( + 'wall_clock(alan_grant=1)', + r"wall_clock.* unexpected keyword argument 'alan_grant'", + id="unexpected-arg" + ), +]) +def test_xtrig_signature_validation( + flow: 'Fixture', validate: 'Fixture', + xtrig_call: str, expected_msg: str +): + """Test automatic xtrigger function signature validation.""" + id_ = flow({ + 'scheduler': {'allow implicit tasks': True}, + 'scheduling': { + 'initial cycle point': '2024', + 'xtriggers': {'myxt': xtrig_call}, + 'graph': {'R1': '@myxt => foo'}, + } + }) + with pytest.raises(XtriggerConfigError, match=expected_msg): + validate(id_) + + +def test_special_task_non_word_names(flow: Fixture, validate: Fixture): + """Test validation of special tasks names with non-word characters""" + wid = flow({ + 'scheduling': { + 'initial cycle point': '2020', + 'special tasks': { + 'clock-trigger': 't-1, t+1, t%1, t@1', + }, + 'graph': { + 'P1D': 't-1 & t+1 & t%1 & t@1', + }, + }, + 'runtime': { + 't-1, t+1, t%1, t@1': {'script': True}, + }, + }) + validate(wid) diff --git a/tests/integration/test_xtrigger_mgr.py b/tests/integration/test_xtrigger_mgr.py index c116e6968a5..612049163cc 100644 --- a/tests/integration/test_xtrigger_mgr.py +++ b/tests/integration/test_xtrigger_mgr.py @@ -21,6 +21,7 @@ from cylc.flow.pathutil import get_workflow_run_dir + async def test_2_xtriggers(flow, start, scheduler, monkeypatch): """Test that if an itask has 2 wall_clock triggers with different offsets that xtrigger manager gets both of them. diff --git a/tests/unit/test_subprocpool.py b/tests/unit/test_subprocpool.py index 981ca8e75f9..9e2d4af212b 100644 --- a/tests/unit/test_subprocpool.py +++ b/tests/unit/test_subprocpool.py @@ -29,7 +29,7 @@ from cylc.flow.cycling.iso8601 import ISO8601Point from cylc.flow.task_events_mgr import TaskJobLogsRetrieveContext from cylc.flow.subprocctx import SubProcContext -from cylc.flow.subprocpool import SubProcPool, _XTRIG_FUNCS, get_func +from cylc.flow.subprocpool import SubProcPool, _XTRIG_FUNC_CACHE, _XTRIG_MOD_CACHE, get_xtrig_func from cylc.flow.task_proxy import TaskProxy @@ -155,7 +155,8 @@ def test_xfunction(self): with the_answer_file.open(mode="w") as f: f.write("""the_answer = lambda: 42""") f.flush() - fn = get_func("the_answer", temp_dir) + f_name = "the_answer" + fn = get_xtrig_func(f_name, f_name, temp_dir) result = fn() self.assertEqual(42, result) @@ -166,19 +167,18 @@ def test_xfunction_cache(self): python_dir.mkdir(parents=True) amandita_file = python_dir / "amandita.py" with amandita_file.open(mode="w") as f: - f.write("""amandita = lambda: 'chocolate'""") + f.write("""choco = lambda: 'chocolate'""") f.flush() - fn = get_func("amandita", temp_dir) + m_name = "amandita" # module + f_name = "choco" # function + fn = get_xtrig_func(m_name, f_name, temp_dir) result = fn() self.assertEqual('chocolate', result) # is in the cache - self.assertTrue('amandita' in _XTRIG_FUNCS) + self.assertTrue((m_name, f_name) in _XTRIG_FUNC_CACHE) # returned from cache - self.assertEqual(fn, get_func("amandita", temp_dir)) - del _XTRIG_FUNCS['amandita'] - # is not in the cache - self.assertFalse('amandita' in _XTRIG_FUNCS) + self.assertEqual(fn, get_xtrig_func(m_name, f_name, temp_dir)) def test_xfunction_import_error(self): """Test for error on importing a xtrigger function. @@ -189,7 +189,7 @@ def test_xfunction_import_error(self): """ with TemporaryDirectory() as temp_dir: with self.assertRaises(ModuleNotFoundError): - get_func("invalid-module-name", temp_dir) + get_xtrig_func("invalid-module-name", "func-name", temp_dir) def test_xfunction_attribute_error(self): """Test for error on looking for an attribute in a xtrigger script.""" @@ -200,8 +200,9 @@ def test_xfunction_attribute_error(self): with the_answer_file.open(mode="w") as f: f.write("""the_droid = lambda: 'excalibur'""") f.flush() + f_name = "the_sword" with self.assertRaises(AttributeError): - get_func("the_sword", temp_dir) + get_xtrig_func(f_name, f_name, temp_dir) @pytest.fixture diff --git a/tests/unit/test_xtrigger_mgr.py b/tests/unit/test_xtrigger_mgr.py index bba77c1f7f1..78f8fe6d969 100644 --- a/tests/unit/test_xtrigger_mgr.py +++ b/tests/unit/test_xtrigger_mgr.py @@ -68,7 +68,7 @@ def test_add_xtrigger_with_params(xtrigger_mgr): assert xtrig == xtrigger_mgr.functx_map["xtrig"] -def test_add_xtrigger_with_unknown_params(xtrigger_mgr): +def test_check_xtrigger_with_unknown_params(xtrigger_mgr): """Test for adding an xtrigger with an unknown parameter. The XTriggerManager contains a list of specific parameters that are @@ -84,22 +84,25 @@ def test_add_xtrigger_with_unknown_params(xtrigger_mgr): label="echo", func_name="echo", func_args=[1, "name", "%(what_is_this)s"], - func_kwargs={"location": "soweto"} + func_kwargs={"succeed": True} ) - with pytest.raises(XtriggerConfigError): - xtrigger_mgr.add_trig("xtrig", xtrig, 'fdir') + with pytest.raises( + XtriggerConfigError, + match="Illegal template in xtrigger: what_is_this" + ): + xtrigger_mgr.check_xtrigger("xtrig", xtrig, 'fdir') -def test_add_xtrigger_with_deprecated_params(xtrigger_mgr, caplog): +def test_check_xtrigger_with_deprecated_params(xtrigger_mgr, caplog): """It should flag deprecated template variables.""" xtrig = SubFuncContext( label="echo", func_name="echo", func_args=[1, "name", "%(suite_name)s"], - func_kwargs={"location": "soweto"} + func_kwargs={"succeed": True} ) caplog.set_level(logging.WARNING, CYLC_LOG) - xtrigger_mgr.add_trig("xtrig", xtrig, 'fdir') + xtrigger_mgr.check_xtrigger("xtrig", xtrig, 'fdir') assert caplog.messages == [ 'Xtrigger "xtrig" uses deprecated template variables: suite_name' ] @@ -139,7 +142,6 @@ def test_housekeeping_nothing_satisfied(xtrigger_mgr): def test_housekeeping_with_xtrigger_satisfied(xtrigger_mgr): """The housekeeping method makes sure only satisfied xtrigger function are kept.""" - xtrigger_mgr.validate_xtrigger = lambda *a, **k: True # Ignore validation xtrig = SubFuncContext( label="get_name", func_name="get_name", @@ -171,7 +173,6 @@ def test_housekeeping_with_xtrigger_satisfied(xtrigger_mgr): def test__call_xtriggers_async(xtrigger_mgr): """Test _call_xtriggers_async""" - xtrigger_mgr.validate_xtrigger = lambda *a, **k: True # Ignore validation # the echo1 xtrig (not satisfied) echo1_xtrig = SubFuncContext( label="echo1", diff --git a/tests/functional/validate/69-bare-clock-xtrigger.t b/tests/unit/xtriggers/test_echo.py similarity index 55% rename from tests/functional/validate/69-bare-clock-xtrigger.t rename to tests/unit/xtriggers/test_echo.py index 4408a064dba..7b1f4d140ee 100644 --- a/tests/functional/validate/69-bare-clock-xtrigger.t +++ b/tests/unit/xtriggers/test_echo.py @@ -1,4 +1,3 @@ -#!/usr/bin/env bash # THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE. # Copyright (C) NIWA & British Crown (Met Office) & Contributors. # @@ -15,18 +14,25 @@ # You should have received a copy of the GNU General Public License # along with this program. If not, see . -# Test that undeclared zero-offset clock xtriggers are allowed. -. "$(dirname "$0")/test_header" +from cylc.flow.exceptions import WorkflowConfigError +from cylc.flow.xtriggers.echo import validate +import pytest +from pytest import param -set_test_number 1 -cat >'flow.cylc' <<'__FLOW_CONFIG__' -[scheduler] - allow implicit tasks = True -[scheduling] - initial cycle point = now - [[graph]] - T00 = "@wall_clock => foo" -__FLOW_CONFIG__ +def test_validate_good(): + validate({ + 'args': (), + 'kwargs': {'succeed': False, 'egg': 'fried', 'potato': 'baked'} + }) -run_ok "${TEST_NAME_BASE}-val" cylc validate . + +@pytest.mark.parametrize( + 'all_args', ( + param({'args': (False,), 'kwargs': {}}, id='no-kwarg'), + param({'args': (), 'kwargs': {'spud': 'mashed'}}, id='no-succeed-kwarg'), + ) +) +def test_validate_exceptions(all_args): + with pytest.raises(WorkflowConfigError, match='^Requires'): + validate(all_args) diff --git a/tests/unit/xtriggers/test_wall_clock.py b/tests/unit/xtriggers/test_wall_clock.py new file mode 100644 index 00000000000..44bba2619fc --- /dev/null +++ b/tests/unit/xtriggers/test_wall_clock.py @@ -0,0 +1,61 @@ +# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE. +# Copyright (C) NIWA & British Crown (Met Office) & Contributors. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +from cylc.flow.xtriggers.wall_clock import _wall_clock, validate +from metomi.isodatetime.parsers import DurationParser +import pytest +from pytest import param + + +@pytest.mark.parametrize('trigger_time, expected', [ + (499, True), + (500, False), +]) +def test_wall_clock( + monkeypatch: pytest.MonkeyPatch, trigger_time: int, expected: bool +): + monkeypatch.setattr( + 'cylc.flow.xtriggers.wall_clock.time', lambda: 500 + ) + assert _wall_clock(trigger_time) == expected + + +@pytest.fixture +def monkeypatch_interval_parser(monkeypatch): + """Interval parse only works normally if a WorkflowSpecifics + object identify the parser to be used. + """ + monkeypatch.setattr( + 'cylc.flow.xtriggers.wall_clock.interval_parse', + DurationParser().parse + ) + + +def test_validate_good(monkeypatch_interval_parser): + validate({'offset': 'PT1H'}) + + +@pytest.mark.parametrize( + 'args, err', ( + param({'offset': 1}, "^Invalid", id='invalid-offset-int'), + param({'offset': 'Zaphod'}, "^Invalid", id='invalid-offset-str'), + ) +) +def test_validate_exceptions( + monkeypatch_interval_parser, args, err +): + with pytest.raises(Exception, match=err): + validate(args) diff --git a/tests/unit/xtriggers/test_xrandom.py b/tests/unit/xtriggers/test_xrandom.py new file mode 100644 index 00000000000..21a9fdb776d --- /dev/null +++ b/tests/unit/xtriggers/test_xrandom.py @@ -0,0 +1,39 @@ +# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE. +# Copyright (C) NIWA & British Crown (Met Office) & Contributors. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import pytest +from pytest import param + +from cylc.flow.xtriggers.xrandom import validate +from cylc.flow.exceptions import WorkflowConfigError + + +def test_validate_good(): + validate({'percent': 1, 'secs': 0, '_': 'HelloWorld'}) + + +@pytest.mark.parametrize( + 'args, err', ( + param({'percent': 'foo'}, r"'percent", id='percent-not-numeric'), + param({'percent': 101}, r"'percent", id='percent>100'), + param({'percent': -1}, r"'percent", id='percent<0'), + param({'percent': 100, 'secs': 1.1}, r"'secs'", id='secs-not-int'), + ) +) +def test_validate_exceptions(args, err): + """Illegal args and kwargs cause a WorkflowConfigError raised.""" + with pytest.raises(WorkflowConfigError, match=f'^{err}'): + validate(args)