Make to_cwl compatible with pulsar

lib/galaxy/tools/evaluation.py

@@ -945,7 +945,7 @@ class UserToolEvaluator(ToolEvaluator):
         from galaxy.workflow.modules import to_cwl
 
         hda_references: List[model.HistoryDatasetAssociation] = []
-        cwl_style_inputs = to_cwl(incoming, hda_references=hda_references)
+        cwl_style_inputs = to_cwl(incoming, hda_references=hda_references, compute_environment=compute_environment)
         return {"inputs": cwl_style_inputs, "outdir": job_working_directory}
 
     def _build_command_line(self):

lib/galaxy/workflow/modules.py

@@ -32,6 +32,7 @@ from galaxy.exceptions import (
     ToolMissingException,
 )
 from galaxy.job_execution.actions.post import ActionBox
+from galaxy.job_execution.compute_environment import ComputeEnvironment
 from galaxy.model import (
     Job,
     PostJobAction,
@@ -149,7 +150,9 @@ class ConditionalStepWhen(BooleanToolParameter):
     pass
 
 
-def to_cwl(value, hda_references, step: Optional[WorkflowStep] = None):
+def to_cwl(
+    value, hda_references, step: Optional[WorkflowStep] = None, compute_environment: Optional[ComputeEnvironment] = None
+):
     element_identifier = None
     if isinstance(value, model.HistoryDatasetCollectionAssociation):
         value = value.collection
@@ -179,7 +182,7 @@ def to_cwl(value, hda_references, step: Optional[WorkflowStep] = None):
                 "class": "File",
                 "location": f"step_input://{len(hda_references)}",
                 "format": value.extension,
-                "path": value.get_file_name(),
+                "path": compute_environment.input_path_rewrite(value) if compute_environment else value.get_file_name(),
             }
             set_basename_and_derived_properties(
                 properties, value.dataset.created_from_basename or element_identifier or value.name
@@ -187,22 +190,33 @@ def to_cwl(value, hda_references, step: Optional[WorkflowStep] = None):
             return properties
     elif isinstance(value, model.DatasetCollection):
         if value.collection_type == "list":
-            return [to_cwl(dce, hda_references=hda_references, step=step) for dce in value.dataset_elements]
+            return [
+                to_cwl(dce, hda_references=hda_references, step=step, compute_environment=compute_environment)
+                for dce in value.dataset_elements
+            ]
         else:
             # Could be record or nested lists
             rval = {}
             for element in value.elements:
                 rval[element.element_identifier] = to_cwl(
-                    element.element_object, hda_references=hda_references, step=step
+                    element.element_object,
+                    hda_references=hda_references,
+                    step=step,
+                    compute_environment=compute_environment,
                 )
             return rval
     elif isinstance(value, list):
-        return [to_cwl(v, hda_references=hda_references, step=step) for v in value]
+        return [
+            to_cwl(v, hda_references=hda_references, step=step, compute_environment=compute_environment) for v in value
+        ]
     elif is_runtime_value(value):
         return None
     elif isinstance(value, dict):
         # Nested tool state, such as conditionals
-        return {k: to_cwl(v, hda_references=hda_references, step=step) for k, v in value.items()}
+        return {
+            k: to_cwl(v, hda_references=hda_references, step=step, compute_environment=compute_environment)
+            for k, v in value.items()
+        }
     else:
         return value
 

test/functional/tools/cat_user_defined.yml

@@ -0,0 +1,21 @@
+class: GalaxyUserTool
+id: cat_user_defined
+version: "0.1"
+name: cat_user_defined
+description: concatenates a file
+container: busybox
+shell_command: cat '$(inputs.input1.path)' > output.txt
+inputs:
+  - name: input1
+    type: data
+    format: txt
+outputs:
+  - name: output1
+    type: data
+    format: txt
+    from_work_dir: output.txt
+tests:
+  - inputs:
+      input1: simple_line.txt
+    outputs:
+      output1: simple_line.txt

test/functional/tools/sample_tool_conf.xml

@@ -305,6 +305,7 @@
   <tool file="explicit_container.xml" />
   <tool file="explicit_singularity_container.xml"/>
   <tool file="simple_constructs.yml" />
+  <tool file="cat_user_defined.yml" />
   <tool file="from_work_dir_glob.xml" />
 
   <!-- Tools without tool test but useful for hand-crafted, artisanal test cases. -->

test/integration/test_pulsar_embedded.py

@@ -17,6 +17,8 @@ class EmbeddedPulsarIntegrationInstance(integration_util.IntegrationInstance):
     def handle_galaxy_config_kwds(cls, config):
         super().handle_galaxy_config_kwds(config)
         config["job_config_file"] = EMBEDDED_PULSAR_JOB_CONFIG_FILE
+        config["enable_celery_tasks"] = False
+        config["metadata_strategy"] = "directory"
 
 
 instance = integration_util.integration_module_instance(EmbeddedPulsarIntegrationInstance)
@@ -24,6 +26,7 @@ instance = integration_util.integration_module_instance(EmbeddedPulsarIntegratio
 test_tools = integration_util.integration_tool_runner(
     [
         "cat_default",
+        "cat_user_defined",
         "collection_nested_default",
         "collection_creates_dynamic_nested_from_json",
         "composite",