Fix inline runtag with no io and duplicate outs in compiled cwl (PolusAI#341)

Author: vjaganat90

.github/workflows/run_workflows.yml

@@ -172,11 +172,6 @@ jobs:
       run: cd sophios/ && pytest tests/test_examples.py -k test_cwl_docker_extract
       # For self-hosted runners, make sure the docker cache is up-to-date.
 
-    - name: PyTest Run Workflows
-      if: always()
-      # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
-      run: cd sophios/ && pytest tests/test_examples.py -k test_run_workflows_on_push --workers 8 --cwl_runner cwltool # --cov
-
     - name: PyTest Run REST Core Tests
       if: always()
       # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
@@ -197,6 +192,11 @@ jobs:
       # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
       run: cd sophios/ && pytest tests/test_fix_payload.py -k test_fix
 
+    - name: PyTest Run Workflows
+      if: always()
+      # NOTE: Do NOT add coverage to PYPY CI runs https://github.com/tox-dev/tox/issues/2252
+      run: cd sophios/ && pytest tests/test_examples.py -k test_run_workflows_on_push --workers 8 --cwl_runner cwltool # --cov
+
     # NOTE: The steps below are for repository_dispatch only. For all other steps, please insert above
     # this comment.
 

src/sophios/api/http/restapi.py

@@ -137,7 +137,6 @@ async def compile_wf(request: Request) -> Json:
                                                             tools_cwl, True, relative_run_path=True, testing=False)
 
     rose_tree = compiler_info.rose
-    input_output.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
     # generating cwl inline within the 'run' tag is post compile
     # and always on when compiling and preparing REST return payload
     rose_tree = cwl_inline_runtag(rose_tree)
@@ -162,14 +161,6 @@ async def compile_wf(request: Request) -> Json:
 
     cwl_tree_run.pop('steps', None)
     cwl_tree_run['steps'] = cwl_tree_run.pop('steps_dict', None)
-
-    # currently there is a compiler bug where the output variables are duplicated
-    # this is a workaround to remove the duplicates till the compiler is fixed
-    for step in cwl_tree_run['steps']:
-
-        out_vars = cwl_tree_run['steps'][step]['out']
-        out_vars_unique = list(set(out_vars))
-        cwl_tree_run['steps'][step]['out'] = out_vars_unique
     compute_workflow: Json = {}
     compute_workflow = {
         "name": yaml_stem,

src/sophios/api/pythonapi.py

@@ -778,22 +778,10 @@ def get_cwl_workflow(self, args_dict: Dict[str, str] = {}) -> Json:
         compiler_info = self.compile(args_dict=args_dict, write_to_disk=False)
         rose_tree = compiler_info.rose
 
-        # this is unfortunately necessary for now
-        # TODO: uncouple dumping to file and getting cwl steps
-        # then remove this write_to_disk call here
-        input_output.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
-
         rose_tree = post_compile.cwl_inline_runtag(rose_tree)
         sub_node_data = rose_tree.data
         cwl_ast = sub_node_data.compiled_cwl
 
-        # Copy samee's workaround for duplicate outs
-        for step in cwl_ast['steps']:
-            out_vars = step['out']
-            out_vars_unique = list(set(out_vars))
-            out_vars_unique.sort()
-            step['out'] = out_vars_unique
-
         yaml_inputs = sub_node_data.workflow_inputs_file
         workflow_json: Json = {}
         workflow_json = {

src/sophios/main.py

@@ -166,11 +166,8 @@ def main() -> None:
     if args.partial_failure_enable:
         rose_tree = plugins.cwl_update_outputs_optional_rosetree(rose_tree)
 
-    io.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
-
     if args.cwl_inline_runtag:
         rose_tree = pc.cwl_inline_runtag(rose_tree)
-    io.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
 
     if args.graphviz:
         if shutil.which('dot'):
@@ -191,11 +188,14 @@ def main() -> None:
             print("but not the graphviz system package.)")
 
     if args.run_local or args.generate_run_script:
+        # Only now we need to write the final cwl for docker-extract
+        # and then for actually running using a cwl_runner
+        io.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
         pc.cwl_docker_extract(args.container_engine, args.pull_dir, yaml_stem)
         if args.docker_remove_entrypoints:
             rose_tree = pc.remove_entrypoints(args.container_engine, rose_tree)
-        io.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
         pc.find_and_create_output_dirs(rose_tree)
+        # No need to re-write to disk as nothing of the cwl or yaml_inputs has changed!
         if args.toil_passthrough_flags == 'yes':
             run_local.run_local(args, rose_tree, args.cachedir, args.cwl_runner, False, passthrough_args=unknown_args)
         else:
@@ -205,6 +205,11 @@ def main() -> None:
         # we need to copy the output files manually. See comment above.
         if args.cwl_runner == 'cwltool' and args.copy_output_files:
             run_local.copy_output_files(yaml_stem)
+    elif args.generate_cwl_workflow:
+        io.write_to_disk(rose_tree, Path('autogenerated/'), True, args.inputs_file)
+    else:
+        print('Please specify either --generate_cwl_workflow (compile) or --run_local (run)')
+        sys.exit(1)
 
 
 if __name__ == '__main__':

src/sophios/plugins.py

@@ -165,37 +165,6 @@ def cwl_update_outputs_optional(cwl: Cwl) -> Cwl:
     return cwl_mod
 
 
-def cwl_update_inline_runtag(cwl: Cwl, path: Path, relative_run_path: bool) -> Cwl:
-    """Updates 'run' tag with inline content
-
-    Args:
-        cwl (Cwl): A CWL Workflow
-        path (Path): The directory in which to read files from
-        relative_run_path (bool): Controls whether to use subdirectories or just one directory
-    Returns:
-        Cwl: A CWL Workflow with inline run (if any)
-    """
-    cwl_mod = copy.deepcopy(cwl)
-    for step in cwl_mod['steps']:
-        runtag_orig = step.get('run', '')
-        match runtag_orig:
-            case str():
-                if runtag_orig.endswith('.cwl'):
-                    if relative_run_path:
-                        yml_path = Path.cwd() / path / runtag_orig
-                    else:
-                        yml_path = Path(runtag_orig)  # Assume absolute path in the runtag
-                    with open(yml_path, mode='r', encoding='utf-8') as f:
-                        runtag_raw = yaml.safe_load(f.read())
-                        # local $namespace and $schema tag shouldn't be in inline cwl steps
-                        runtag_raw.pop('$namespaces', None)
-                        runtag_raw.pop('$schemas', None)
-                        step['run'] = runtag_raw
-            case _:
-                pass  # We only care if the runtag is a cwl filepath
-    return cwl_mod
-
-
 Client = Union[docker.DockerClient, podman.PodmanClient]  # type: ignore
 
 
@@ -265,30 +234,6 @@ def remove_entrypoints_podman() -> None:
         remove_entrypoints(client, podman.domain.images_build.BuildMixin())
 
 
-def cwl_update_inline_runtag_rosetree(rose_tree: RoseTree, path: Path, relative_run_path: bool) -> RoseTree:
-    """Inlines the compiled CWL files runtag
-
-    Args:
-        rose_tree (RoseTree): The data associated with compiled subworkflows
-        path (Path): The directory in which to read files from
-        relative_run_path (bool): Controls whether to use subdirectories or just one directory.
-    Returns:
-        RoseTree: rose_tree with inline cwl runtag
-    """
-    n_d: NodeData = rose_tree.data
-    if n_d.compiled_cwl['class'] == 'Workflow':
-        outputs_cwl_inline_runtag = cwl_update_inline_runtag(n_d.compiled_cwl, path, relative_run_path)
-    else:
-        outputs_cwl_inline_runtag = n_d.compiled_cwl
-
-    sub_trees_path = [cwl_update_inline_runtag_rosetree(sub_rose_tree, path, relative_run_path) for
-                      sub_rose_tree in rose_tree.sub_trees]
-    node_data_path = NodeData(n_d.namespaces, n_d.name, n_d.yml, outputs_cwl_inline_runtag, n_d.tool,
-                              n_d.workflow_inputs_file, n_d.explicit_edge_defs, n_d.explicit_edge_calls,
-                              n_d.graph, n_d.inputs_workflow, n_d.step_name_1)
-    return RoseTree(node_data_path, sub_trees_path)
-
-
 def cwl_update_outputs_optional_rosetree(rose_tree: RoseTree) -> RoseTree:
     """Updates outputs optional for every CWL CommandLineTool
 

src/sophios/post_compile.py

@@ -1,8 +1,9 @@
 from pathlib import Path
+import copy
 import subprocess as sub
 from typing import Dict, Union
 from . import plugins
-from .wic_types import RoseTree
+from .wic_types import RoseTree, NodeData
 
 
 def find_output_dirs(data: Union[RoseTree, Dict, list]) -> list:
@@ -73,10 +74,24 @@ def cwl_docker_extract(container_engine: str, pull_dir: str, file_name: str) ->
 
 
 def cwl_inline_runtag(rose_tree: RoseTree) -> RoseTree:
-    """Transform with cwl inline runtag"""
-    # this has to happen after at least one write
-    # so we can copy from local cwl_dapters in autogenerated/
-    return plugins.cwl_update_inline_runtag_rosetree(rose_tree, Path('autogenerated/'), True)
+    """Transforms the compiled CWL within the rose_tree with inline cwl of steps in the runtag
+    Args:
+        rose_tree (RoseTree): The data associated with compiled subworkflows
+    Returns:
+        RoseTree: The updated rose_tree with inline cwl in runtag
+    """
+    rose_tree_mod = copy.deepcopy(rose_tree)
+    node_data: NodeData = rose_tree_mod.data
+    cwl_tree = node_data.compiled_cwl
+
+    if cwl_tree.get('class', '') == 'Workflow':
+        for sub_rose_tree in rose_tree_mod.sub_trees:
+            sub_node_data: NodeData = sub_rose_tree.data
+            sub_step_name = sub_node_data.namespaces[-1]
+            step_to_update = next(item for item in cwl_tree['steps'] if item.get('id') == sub_step_name)
+            step_to_update['run'] = sub_node_data.compiled_cwl
+            sub_rose_tree = cwl_inline_runtag(sub_rose_tree)
+    return rose_tree_mod
 
 
 def remove_entrypoints(container_engine: str, rose_tree: RoseTree) -> RoseTree:

src/sophios/utils_cwl.py

@@ -92,6 +92,7 @@ def add_yamldict_keyval_out(steps_i: Yaml, step_key: str, strs: List[str]) -> Ya
     if steps_i:
         if 'out' in steps_i:
             new_strs = steps_i['out'] + strs
+            new_strs = list(set(new_strs))
             new_keyvals = dict([(k, v) if k != 'out' else (k, new_strs) for k, v in steps_i.items()])
         else:
             new_keyvals = dict(list(steps_i.items()) + [('out', strs)])

tests/test_cli_flags.py

@@ -1,10 +1,7 @@
 import pathlib
 import subprocess
-import yaml
 import shutil
-
-from sophios.main import main
-from sophios.cli import get_args
+import yaml
 
 
 def test_generate_cwl_workflow() -> None:
@@ -19,17 +16,16 @@ def test_generate_cwl_workflow() -> None:
 
     yaml_path = str(pathlib.Path(__file__).parent.parent.resolve() / "docs/tutorials/helloworld.wic")
 
-    args = ["sophios", "--yaml", yaml_path, "--generate_cwl_workflow"]
+    cmd = ["sophios", "--yaml", yaml_path, "--generate_cwl_workflow"]
 
     # run sophios with args
-    subprocess.run(args)
+    subprocess.run(cmd, check=False)
 
-    with open("autogenerated/helloworld.cwl", "r") as cwl_file:
+    with open("autogenerated/helloworld.cwl", "r", encoding='utf-8') as cwl_file:
         result_dict = yaml.safe_load(cwl_file)
 
-    with open(
-        str(pathlib.Path(__file__).parent.resolve() / "data/cwl/helloworld.cwl"), "r"
-    ) as cwl_file:
+    with open(str(pathlib.Path(__file__).parent.resolve() / "data/cwl/helloworld.cwl"), "r",
+              encoding='utf-8') as cwl_file:
         actual_dict = yaml.safe_load(cwl_file)
 
     assert result_dict == actual_dict

tests/test_rest_core.py

@@ -118,7 +118,6 @@ def prepare_call_rest_api(inp_path: Path) -> Json:
     """prepare payload and call rest api"""
     with open(inp_path, 'r', encoding='utf-8') as f:
         inp = json.load(f)
-    print('----------- from rest api ----------- \n\n')
     scope = {}
     scope['type'] = 'http'
 

tests/test_scattering_scaling.py

@@ -13,7 +13,7 @@
 def test_scattering_scaling(minval: int = 100, maxval: int = 500, step: int = 100) -> None:
     """Tests that execution time scales linearly with input array size"""
     # First compile the workflow
-    cmd = ['sophios', '--yaml', 'examples/scattering_scaling.wic']
+    cmd = ['sophios', '--yaml', 'examples/scattering_scaling.wic', '--generate_run_script']
     sub.run(cmd, check=True)
 
     basedir = 'autogenerated'