Implement group: tags (toward multi-factor analysis with group tagging)

lib/galaxy/managers/collections.py

@@ -202,7 +202,7 @@ def _append_tags(self, dataset_collection_instance, implicit_inputs=None, tags=N
         tags = tags or {}
         implicit_inputs = implicit_inputs or []
         for _, v in implicit_inputs:
-            for tag in [t for t in v.tags if t.user_tname == 'name']:
+            for tag in v.auto_propagated_tags:
                 tags[tag.value] = tag
         for _, tag in tags.items():
             dataset_collection_instance.tags.append(tag.copy(cls=model.HistoryDatasetCollectionTagAssociation))

lib/galaxy/managers/tags.py

@@ -305,7 +305,7 @@ def _get_name_value_pair(self, tag_str):
         """Get name, value pair from a tag string."""
         # Use regular expression to parse name, value.
         reg_exp = re.compile("[" + self.key_value_separators + "]")
-        name_value_pair = reg_exp.split(tag_str)
+        name_value_pair = reg_exp.split(tag_str, 1)
         # Add empty slot if tag does not have value.
         if len(name_value_pair) < 2:
             name_value_pair.append(None)

lib/galaxy/model/__init__.py

@@ -71,6 +71,8 @@
 JOB_METRIC_MAX_LENGTH = 1023
 JOB_METRIC_PRECISION = 26
 JOB_METRIC_SCALE = 7
+# Tags that get automatically propagated from inputs to outputs when running jobs.
+AUTO_PROPAGATED_TAGS = ["name", "group"]
 
 
 class NoConverterException(Exception):
@@ -128,6 +130,10 @@ def copy_tags_from(self, target_user, source):
             new_tag_assoc.user = target_user
             self.tags.append(new_tag_assoc)
 
+    @property
+    def auto_propagated_tags(self):
+        return [t for t in self.tags if t.user_tname in AUTO_PROPAGATED_TAGS]
+
 
 class HasName(object):
 

lib/galaxy/tools/actions/__init__.py

@@ -206,7 +206,7 @@ def _collect_inputs(self, tool, trans, incoming, history, current_user_roles):
             if not data:
                 continue
 
-            for tag in [t for t in data.tags if t.user_tname == 'name']:
+            for tag in data.auto_propagated_tags:
                 preserved_tags[tag.value] = tag
 
         # grap tags from incoming HDCAs
@@ -215,7 +215,7 @@ def _collect_inputs(self, tool, trans, incoming, history, current_user_roles):
                 # if sub-collection mapping, this will be an DC not an HDCA
                 # (e.g. part of collection not a collection instance) and thus won't have tags.
                 if hasattr(collection, "tags"):
-                    for tag in [t for t in collection.tags if t.user_tname == 'name']:
+                    for tag in collection.auto_propagated_tags:
                         preserved_tags[tag.value] = tag
 
         return history, inp_data, inp_dataset_collections, preserved_tags

test/api/test_workflows.py

@@ -2321,6 +2321,54 @@ def test_run_hide_on_mapped_over_collection(self):
             assert content["history_content_type"] == "dataset_collection", content
             assert not content["visible"]
 
+    @skip_without_tool("cat")
+    def test_tag_auto_propagation(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  - id: input1
+steps:
+  - label: first_cat
+    tool_id: cat
+    state:
+      input1:
+        $link: input1
+    outputs:
+      out_file1:
+        add_tags:
+            - "name:treated1fb"
+            - "group:condition:treated"
+            - "group:type:single-read"
+            - "machine:illumina"
+  - label: second_cat
+    tool_id: cat
+    state:
+      input1:
+        $link: first_cat#out_file1
+test_data:
+  input1:
+    value: 1.fasta
+    type: File
+    name: fasta1
+""", history_id=history_id)
+
+            details0 = self.dataset_populator.get_history_dataset_details(history_id, hid=2, wait=True, assert_ok=True)
+            tags = details0["tags"]
+            assert len(tags) == 4, details0
+            assert "name:treated1fb" in tags, tags
+            assert "group:condition:treated" in tags, tags
+            assert "group:type:single-read" in tags, tags
+            assert "machine:illumina" in tags, tags
+
+            details1 = self.dataset_populator.get_history_dataset_details(history_id, hid=3, wait=True, assert_ok=True)
+            tags = details1["tags"]
+            assert len(tags) == 3, details1
+            assert "name:treated1fb" in tags, tags
+            assert "group:condition:treated" in tags, tags
+            assert "group:type:single-read" in tags, tags
+            assert "machine:illumina" not in tags, tags
+
     @skip_without_tool("collection_creates_pair")
     def test_run_add_tag_on_collection_output(self):
         with self.dataset_populator.test_history() as history_id: