From b30fed83eb03b1bf737060ee611e75df7dd329a2 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Wed, 7 Aug 2024 13:41:01 +0200
Subject: [PATCH] Add extraction test for empty implicit collection

---
 .../api/test_workflow_extraction.py           | 40 +++++++++++++++++++
 1 file changed, 40 insertions(+)

diff --git a/lib/galaxy_test/api/test_workflow_extraction.py b/lib/galaxy_test/api/test_workflow_extraction.py
index 4d48887cc30d..ada641171af6 100644
--- a/lib/galaxy_test/api/test_workflow_extraction.py
+++ b/lib/galaxy_test/api/test_workflow_extraction.py
@@ -224,6 +224,46 @@ def test_extract_workflows_with_dataset_collections(self, history_id):
         collection_step_state = loads(collection_step["tool_state"])
         assert collection_step_state["collection_type"] == "paired"
 
+    def test_empty_collection_map_over_extract_workflow(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_workflow(
+                """class: GalaxyWorkflow
+inputs:
+  input: collection
+  filter_file: data
+steps:
+  filter_collection:
+    tool_id: __FILTER_FROM_FILE__
+    in:
+       input: input
+       how|filter_source: filter_file
+    state:
+       how:
+         how_filter: remove_if_present
+  concat:
+    tool_id: cat1
+    in:
+      input1: filter_collection/output_filtered
+test_data:
+  input:
+    collection_type: list
+    elements:
+      - identifier: i1
+        content: "0"
+  filter_file: i1""",
+                history_id,
+                wait=True,
+            )
+            response = self._post(
+                "workflows", data={"from_history_id": history_id, "workflow_name": "extract with empty collection test"}
+            )
+            assert response.status_code == 200
+            workflow_id = response.json()["id"]
+            workflow = self.workflow_populator.download_workflow(workflow_id)
+            assert workflow
+            # TODO: after adding request models we should be able to recover implicit collection job requests.
+            # assert len(workflow["steps"]) == 4
+
     @skip_without_tool("cat_collection")
     def test_subcollection_mapping(self, history_id):
         jobs_summary = self._run_workflow(