Merge 0c9fc8b into bc65530

reana_commons/errors.py

@@ -27,3 +27,11 @@ def __str__(self):
 
 class REANASecretAlreadyExists(Exception):
     """The referenced secret already exists."""
+
+
+class REANAValidationError(Exception):
+    """Validation error."""
+
+    def __init__(self, message):
+        """Initialize REANAValidationError exception."""
+        self.message = message

reana_commons/operational_options.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of REANA.
+# Copyright (C) 2020 CERN.
+#
+# REANA is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+"""REANA Workflow Operational Options implementation utils."""
+
+import copy
+import subprocess
+import sys
+
+import click
+
+from reana_commons.errors import REANAValidationError
+
+available_options = {
+    'CACHE': {'serial': 'CACHE'},
+    'FROM': {'serial': 'FROM'},
+    'TARGET': {'serial': 'TARGET', 'cwl': '--target'},
+    'toplevel': {'yadage': 'toplevel'},
+    'initdir': {'yadage': 'initdir'}
+}
+
+
+def validate_operational_options(workflow_type, parsed_options):
+    """Validate and return  workflow operational options.
+
+    :param workflow_type: A supported workflow specification type.
+    :param parsed_options: A dict with the parsed operational parameters.
+    :returns: A dictionary which represents the valid workflow specification.
+    """
+    if not isinstance(parsed_options, dict):
+        raise REANAValidationError(
+            '==> ERROR: Operational options must be a dictionary.')
+    elif not parsed_options:
+        return parsed_options
+
+    validated_options = copy.deepcopy(parsed_options)
+    for option in parsed_options.keys():
+        if option not in available_options:
+            raise REANAValidationError(
+                '==> ERROR: Operational option "{0}" not supported.'
+                .format(option))
+        translation = available_options[option].get(workflow_type)
+        if not translation:
+            raise REANAValidationError(
+                '==> ERROR: Operational option "{0}" not supported for'
+                ' {1} workflows.'
+                .format(option, workflow_type))
+        # Override engine specific options
+        if translation not in available_options:
+            validated_options[translation] = validated_options[option]
+            del validated_options[option]
+
+    return validated_options

reana_commons/serial.py

@@ -81,7 +81,8 @@
 }
 
 
-def serial_load(workflow_file, specification, parameters=None, original=None):
+def serial_load(workflow_file, specification, parameters=None, original=None,
+                **kwargs):
     """Validate and return a expanded REANA Serial workflow specification.
 
     :param workflow_file: A specification file compliant with
@@ -105,33 +106,32 @@ def serial_load(workflow_file, specification, parameters=None, original=None):
 
 
 def _expand_parameters(specification, parameters, original=None):
-    """Expand parameters inside comands for Serial workflow specifications.
+    """Expand parameters inside commands for Serial workflow specifications.
 
     :param specification: Full valid Serial workflow specification.
     :param parameters: Parameters to be extended on a Serial specification.
-    :param original: Flag which, determins type of specifications to return.
+    :param original: Flag which, determines type of specifications to return.
     :returns: If 'original' parameter is set, a copy of the specification
         whithout expanded parametrers will be returned. If 'original' is not
         set, a copy of the specification with expanded parameters (all $varname
         and ${varname} will be expanded with their value). Otherwise an error
         will be thrown if the parameters can not be expanded.
     :raises: jsonschema.ValidationError
     """
-    expanded_specification = deepcopy(specification)
-
-    try:
-        for step_num, step in enumerate(expanded_specification['steps']):
-            current_step = expanded_specification['steps'][step_num]
-            for command_num, command in enumerate(step['commands']):
-                current_step['commands'][command_num] = \
-                    Template(command).substitute(parameters)
-        # if call is done from client, original==True and original
-        # specifications withtout applied parameters are returned.
-        if original:
-            return specification
-        else:
+    # if call is done from client, original==True and original
+    # specifications withtout applied parameters are returned.
+    if original:
+        return specification
+    else:
+        try:
+            expanded_specification = deepcopy(specification)
+            for step_num, step in enumerate(expanded_specification['steps']):
+                current_step = expanded_specification['steps'][step_num]
+                for command_num, command in enumerate(step['commands']):
+                    current_step['commands'][command_num] = \
+                        Template(command).substitute(parameters)
             return expanded_specification
-    except KeyError as e:
-        raise ValidationError('Workflow parameter(s) could not '
-                              'be expanded. Please take a look '
-                              'to {params}'.format(params=str(e)))
+        except KeyError as e:
+            raise ValidationError('Workflow parameter(s) could not '
+                                  'be expanded. Please take a look '
+                                  'to {params}'.format(params=str(e)))

reana_commons/version.py

@@ -14,4 +14,4 @@
 
 from __future__ import absolute_import, print_function
 
-__version__ = "0.7.0.dev20200319"
+__version__ = "0.7.0.dev20200402"

tests/test_operational_options.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of REANA.
+# Copyright (C) 2020 CERN.
+#
+# REANA is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+
+"""REANA-Commons utilities testing."""
+
+import pytest
+
+from reana_commons.errors import REANAValidationError
+from reana_commons.operational_options import validate_operational_options
+
+
+@pytest.mark.parametrize(
+    'workflow_type, options, except_msg',
+    [
+        ('serial', 'CACHE=on', 'must be a dict'),
+        ('yadage', {'unsupported': 'yes'}, 'not supported'),
+        ('foo', {'TARGET': 'gendata'},
+         'not supported for {workflow_type} workflows'),
+        ('serial', {'toplevel': 'github:reanahub/awesome-workflow'},
+         'not supported for {workflow_type} workflows'),
+    ],
+)
+def test_unsupported(workflow_type, options, except_msg):
+    """Test unsupported operational options cases."""
+    with pytest.raises(REANAValidationError) as e:
+        validate_operational_options(workflow_type, options)
+    assert except_msg.format(workflow_type=workflow_type) in e.value.message
+
+
+@pytest.mark.parametrize(
+    'workflow_type, options, option',
+    [
+        ('serial', {'FROM': 'fitdata'}, 'FROM'),
+        ('yadage', {'toplevel': 'github:reanahub/awesome-workflow'},
+         'toplevel'),
+        ('cwl', {'TARGET': 'gendata'}, '--target')
+    ],
+)
+def test_successful(workflow_type, options, option):
+    """Test successful operational options cases."""
+    validated_options = validate_operational_options(workflow_type, options)
+    assert validated_options[option]