diff --git a/changes.d/5452.feat.md b/changes.d/5452.feat.md new file mode 100644 index 00000000000..c9fc9b4d879 --- /dev/null +++ b/changes.d/5452.feat.md @@ -0,0 +1 @@ +Support xtrigger argument validation. diff --git a/cylc/flow/config.py b/cylc/flow/config.py index d80456266bf..b61edfdf771 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, @@ -82,6 +83,7 @@ from cylc.flow.print_tree import print_tree from cylc.flow.simulation import configure_sim_modes from cylc.flow.subprocctx import SubFuncContext +from cylc.flow.subprocpool import get_xtrig_func from cylc.flow.task_events_mgr import ( EventData, get_event_handler_data @@ -1711,28 +1713,33 @@ 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) + + # Specific xtrigger.validate(), if available. + with suppress(AttributeError, ImportError): + get_xtrig_func(xtrig.func_name, "validate", self.fdir)( + xtrig.func_args, + xtrig.func_kwargs, + xtrig.get_signature() + ) + + 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): @@ -2423,7 +2430,7 @@ def upgrade_clock_triggers(self): xtrig = SubFuncContext(label, 'wall_clock', [], {}) xtrig.func_kwargs["offset"] = offset 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/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..890918c49dc 100644 --- a/cylc/flow/xtrigger_mgr.py +++ b/cylc/flow/xtrigger_mgr.py @@ -26,7 +26,7 @@ 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.subprocpool import get_xtrig_func from cylc.flow.xtriggers.wall_clock import wall_clock if TYPE_CHECKING: @@ -240,12 +240,14 @@ 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 and string templates. + + Xtrigger modules may also supply a specific "validate" function. Args: label: xtrigger label @@ -262,8 +264,9 @@ def validate_xtrigger( """ fname: str = fctx.func_name + try: - func = get_func(fname, fdir) + func = get_xtrig_func(fname, fname, fdir) except ImportError: raise XtriggerConfigError( label, @@ -318,13 +321,14 @@ def validate_xtrigger( 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): diff --git a/cylc/flow/xtriggers/echo.py b/cylc/flow/xtriggers/echo.py index e3c5880cab0..adfb9bc9f83 100644 --- a/cylc/flow/xtriggers/echo.py +++ b/cylc/flow/xtriggers/echo.py @@ -16,12 +16,11 @@ """A Cylc xtrigger function.""" -from contextlib import suppress +from cylc.flow.exceptions import WorkflowConfigError def echo(*args, **kwargs): - """Prints args to stdout and return success only if kwargs['succeed'] - is True. + """Print arguments to stdout, return kwargs['succeed'] and kwargs. This may be a useful aid to understanding how xtriggers work. @@ -31,7 +30,18 @@ def echo(*args, **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(f_args, f_kwargs, f_signature): + """ + Validate the xtrigger function arguments parsed from the workflow config. + + This is separate from the xtrigger to allow parse-time validation. + + """ + if "succeed" not in f_kwargs or not type(f_kwargs["succeed"]) is bool: + raise WorkflowConfigError( + f"Requires 'succeed=True/False' arg: {f_signature}" + ) diff --git a/cylc/flow/xtriggers/wall_clock.py b/cylc/flow/xtriggers/wall_clock.py index 3c50d956f0a..0e0c40c733c 100644 --- a/cylc/flow/xtriggers/wall_clock.py +++ b/cylc/flow/xtriggers/wall_clock.py @@ -17,6 +17,8 @@ """xtrigger function to trigger off of a wall clock time.""" from time import time +from cylc.flow.cycling.iso8601 import interval_parse +from cylc.flow.exceptions import WorkflowConfigError def wall_clock(trigger_time=None): @@ -27,3 +29,44 @@ def wall_clock(trigger_time=None): Trigger time as seconds since Unix epoch. """ return time() > trigger_time + + +def validate(f_args, f_kwargs, f_signature): + """Validate and manipulate args parsed from the workflow config. + + wall_clock() # infer zero interval + wall_clock(PT1H) + wall_clock(offset=PT1H) + + The offset must be a valid ISO 8601 interval. + + If f_args used, convert to f_kwargs for clarity. + + """ + + n_args = len(f_args) + n_kwargs = len(f_kwargs) + + if n_args + n_kwargs > 1: + raise WorkflowConfigError(f"Too many args: {f_signature}") + + if n_kwargs: + # sole kwarg must be "offset" + kw = next(iter(f_kwargs)) + if kw != "offset": + raise WorkflowConfigError(f"Illegal arg '{kw}': {f_signature}") + + elif n_args: + # convert to kwarg + f_kwargs["offset"] = f_args[0] + del f_args[0] + + else: + # no args, infer zero interval + f_kwargs["offset"] = "P0Y" + + # must be a valid interval + try: + interval_parse(f_kwargs["offset"]) + except ValueError: + raise WorkflowConfigError(f"Invalid offset: {f_signature}") diff --git a/setup.cfg b/setup.cfg index 7db9eb8bf4e..eca9140b181 100644 --- a/setup.cfg +++ b/setup.cfg @@ -215,12 +215,14 @@ 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 - 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 + xrandom = cylc.flow.xtriggers.xrandom [bdist_rpm] requires = 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/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..9b497d8e5ab 100644 --- a/tests/unit/test_xtrigger_mgr.py +++ b/tests/unit/test_xtrigger_mgr.py @@ -44,7 +44,7 @@ def test_extract_templates(): ) -def test_add_xtrigger(xtrigger_mgr): +def test_check_xtrigger(xtrigger_mgr): """Test for adding an xtrigger.""" xtrig = SubFuncContext( label="echo", @@ -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 @@ -87,10 +87,10 @@ def test_add_xtrigger_with_unknown_params(xtrigger_mgr): func_kwargs={"location": "soweto"} ) with pytest.raises(XtriggerConfigError): - xtrigger_mgr.add_trig("xtrig", xtrig, 'fdir') + 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", @@ -99,7 +99,7 @@ def test_add_xtrigger_with_deprecated_params(xtrigger_mgr, caplog): func_kwargs={"location": "soweto"} ) 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 +139,7 @@ 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 + xtrigger_mgr.check_xtrigger = lambda *a, **k: True # Ignore validation xtrig = SubFuncContext( label="get_name", func_name="get_name", @@ -171,7 +171,7 @@ 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 + xtrigger_mgr.check_xtrigger = lambda *a, **k: True # Ignore validation # the echo1 xtrig (not satisfied) echo1_xtrig = SubFuncContext( label="echo1",