Merge pull request #1395 from mvdbeek/fix_workflow_lint

Fix workflow_lint for json output values

planemo/workflow_lint.py

@@ -290,16 +290,7 @@ def _lint_case(path: str, test_case: TestCase, lint_context: WorkflowLintContext
         # TODO: validate structure of test expectations
 
         output_expectations = test_case.output_expectations[test_output_id]
-        all_assertion_definitions = []
-        if "element_tests" in output_expectations:
-            # This is a collection
-            for element_id in output_expectations["element_tests"]:
-                all_assertion_definitions.append(output_expectations["element_tests"][element_id].get("asserts"))
-        else:
-            all_assertion_definitions.append(output_expectations.get("asserts"))
-        for assertion_definitions in all_assertion_definitions:
-            if not _check_test_assertions(lint_context, assertion_definitions):
-                test_valid = False
+        test_valid = is_valid_output_expectations(lint_context=lint_context, output_expectations=output_expectations)
 
     if not found_valid_expectation:
         lint_context.warn("Found no valid test expectations for workflow test")
@@ -308,6 +299,23 @@ def _lint_case(path: str, test_case: TestCase, lint_context: WorkflowLintContext
     return test_valid
 
 
+def is_valid_output_expectations(lint_context, output_expectations):
+    all_assertion_definitions = []
+    if isinstance(output_expectations, (int, str, float, bool)):
+        # CWL style parameter output
+        return True
+    elif "element_tests" in output_expectations:
+        # This is a collection
+        for element_id in output_expectations["element_tests"]:
+            all_assertion_definitions.append(output_expectations["element_tests"][element_id].get("asserts"))
+    else:
+        all_assertion_definitions.append(output_expectations.get("asserts"))
+    for assertion_definitions in all_assertion_definitions:
+        if not _check_test_assertions(lint_context, assertion_definitions):
+            return False
+    return True
+
+
 def _check_test_assertions(
     lint_context: WorkflowLintContext, assertion_definitions: Optional[Dict[str, Dict[str, Any]]]
 ) -> bool:

tests/data/json_value_out-tests.yml

@@ -0,0 +1,5 @@
+- doc: Test outline for Param-value-from-file-workflow
+  job:
+    int_in: 2
+  outputs:
+    int_out: 2

tests/data/json_value_out.yml

@@ -0,0 +1,27 @@
+class: GalaxyWorkflow
+label: Param value from file workflow
+inputs:
+  int_in:
+    optional: false
+    type: int
+outputs:
+  int_out:
+    outputSource: pick/integer_param
+steps:
+  pick:
+    tool_id: pick_value
+    tool_version: 0.1.0
+    tool_state:
+      style_cond:
+        __current_case__: 0
+        pick_style: first
+        type_cond:
+          __current_case__: 1
+          param_type: integer
+          pick_from:
+          - __index__: 0
+            value:
+              __class__: ConnectedValue
+    in:
+      style_cond|type_cond|pick_from_0|value:
+        source: int_in

tests/test_cmd_workflow_lint.py

@@ -179,6 +179,11 @@ def test_assertion_linting(self):
             in result.output
         )
 
+    def test_json_value_out(self):
+        workflow_path = "/".join((TEST_DATA_DIR, "json_value_out.yml"))
+        lint_cmd = ["workflow_lint", workflow_path, "--fail_level", "error"]
+        self._check_exit_code(lint_cmd, exit_code=0)
+
     def test_tool_id_linting_wrong_version(self):
         workflow_path = "/".join(
             (TEST_DATA_DIR, "wf_repos", "autoupdate_tests", "workflow_with_unexisting_version_of_tool.ga")