Add abstraction for popping connection dictionary to model.

Author: jmchilton

gxformat2/converter.py

@@ -14,6 +14,7 @@
     convert_dict_to_id_list_if_needed,
     ensure_step_position,
     inputs_as_native_steps,
+    pop_connect_from_step_dict,
     with_step_ids,
 )
 from .yaml import ordered_load
@@ -89,13 +90,13 @@ def rename_arg(argument):
     return argument
 
 
-def clean_connection(value):
+def clean_connection(value: str) -> str:
     if value and "#" in value and SUPPORT_LEGACY_CONNECTIONS:
         # Hope these are just used by Galaxy testing workflows and such, and not in production workflows.
         log.warn(f"Legacy workflow syntax for connections [{value}] will not be supported in the future")
         value = value.replace("#", "/", 1)
-    else:
-        return value
+
+    return value
 
 
 class ImportOptions:
@@ -381,7 +382,7 @@ def transform_pause(context, step, default_name="Pause for dataset review"):
         "name": name
     }
 
-    connect = _init_connect_dict(step)
+    connect = pop_connect_from_step_dict(step)
     _populate_input_connections(context, step, connect)
     _populate_tool_state(step, tool_state)
 
@@ -398,7 +399,7 @@ def transform_subworkflow(context, step):
     tool_state = {
     }
 
-    connect = _init_connect_dict(step)
+    connect = pop_connect_from_step_dict(step)
     _populate_input_connections(context, step, connect)
     _populate_tool_state(step, tool_state)
 
@@ -428,7 +429,7 @@ def transform_tool(context, step):
         "__page__": 0,
     }
 
-    connect = _init_connect_dict(step)
+    connect = pop_connect_from_step_dict(step)
 
     def append_link(key, value):
         if key not in connect:
@@ -641,38 +642,6 @@ def _join_prefix(prefix, key):
     return new_key
 
 
-def _init_connect_dict(step):
-    if "connect" not in step:
-        step["connect"] = {}
-
-    connect = step["connect"]
-    del step["connect"]
-
-    # handle CWL-style in dict connections.
-    if "in" in step:
-        step_in = step["in"]
-        assert isinstance(step_in, dict)
-        connection_keys = set()
-        for key, value in step_in.items():
-            # TODO: this can be a list right?
-            if isinstance(value, dict) and 'source' in value:
-                value = value["source"]
-            elif isinstance(value, dict) and 'default' in value:
-                continue
-            elif isinstance(value, dict):
-                raise KeyError(f'step input must define either source or default {value}')
-            connect[key] = [value]
-            connection_keys.add(key)
-
-        for key in connection_keys:
-            del step_in[key]
-
-        if len(step_in) == 0:
-            del step['in']
-
-    return connect
-
-
 def _populate_input_connections(context, step, connect):
     _ensure_inputs_connections(step)
     input_connections = step["input_connections"]

gxformat2/model.py

@@ -4,6 +4,38 @@
 DictOrList = Union[Dict, List]
 
 
+def pop_connect_from_step_dict(step: dict) -> dict:
+    if "connect" not in step:
+        step["connect"] = {}
+
+    connect = step["connect"]
+    del step["connect"]
+
+    # handle CWL-style in dict connections.
+    if "in" in step:
+        step_in = step["in"]
+        assert isinstance(step_in, dict)
+        connection_keys = set()
+        for key, value in step_in.items():
+            # TODO: this can be a list right?
+            if isinstance(value, dict) and 'source' in value:
+                value = value["source"]
+            elif isinstance(value, dict) and 'default' in value:
+                continue
+            elif isinstance(value, dict):
+                raise KeyError(f'step input must define either source or default {value}')
+            connect[key] = [value]
+            connection_keys.add(key)
+
+        for key in connection_keys:
+            del step_in[key]
+
+        if len(step_in) == 0:
+            del step['in']
+
+    return connect
+
+
 def convert_dict_to_id_list_if_needed(
     dict_or_list: DictOrList,
     add_label: bool = False,

tests/test_model_helpers.py

@@ -0,0 +1,27 @@
+from gxformat2.model import pop_connect_from_dict
+
+
+def test_pop_connect():
+    raw_step = {
+        "in": {
+            "bar": {
+                "source": "foo/moo",
+            },
+        },
+    }
+    connect = pop_connect_from_dict(raw_step)
+    assert connect["bar"] == ["foo/moo"]
+    assert "in" not in raw_step
+
+
+def test_pop_connect_preserves_defaults():
+    raw_step = {
+        "in": {
+            "bar": {
+                "default": 7,
+            },
+        },
+    }
+    connect = pop_connect_from_dict(raw_step)
+    assert "bar" not in connect
+    assert "in" in raw_step