Migrate test_multidata_param to more declarative, pytest-y test.

lib/galaxy_test/api/test_tool_execute.py

@@ -0,0 +1,131 @@
+"""Test tool execution pieces.
+
+Longer term ideally we would separate all the tool tests in test_tools.py that
+describe tool execution into this file and make sure we have parallel or matching
+tests for both the legacy tool execution API and the tool request API. We would then
+keep things like testing other tool APIs in ./test_tools.py (index, search, tool test
+files, etc..). 
+"""
+
+from typing import List
+
+from galaxy_test.base.decorators import requires_tool_id
+from galaxy_test.base.populators import (
+    RequiredTool,
+    TargetHistory,
+)
+
+
+@requires_tool_id("multi_data_param")
+def test_multidata_param(target_history: TargetHistory, required_tool: RequiredTool):
+    hda1 = target_history.with_dataset("1\t2\t3").src_dict
+    hda2 = target_history.with_dataset("4\t5\t6").src_dict
+    execution = required_tool.execute.with_inputs(
+        {
+            "f1": {"batch": False, "values": [hda1, hda2]},
+            "f2": {"batch": False, "values": [hda2, hda1]},
+        }
+    )
+    execution.assert_has_job(0).with_output("out1").with_contents("1\t2\t3\n4\t5\t6\n")
+    execution.assert_has_job(0).with_output("out2").with_contents("4\t5\t6\n1\t2\t3\n")
+
+
+@requires_tool_id("expression_forty_two")
+def test_galaxy_expression_tool_simplest(required_tool: RequiredTool):
+    required_tool.execute.assert_has_single_job.with_single_output.with_contents("42")
+
+
+@requires_tool_id("expression_parse_int")
+def test_galaxy_expression_tool_simple(required_tool: RequiredTool):
+    execution = required_tool.execute.with_inputs({"input1": "7"})
+    execution.assert_has_single_job.with_single_output.with_contents("7")
+
+
+@requires_tool_id("expression_log_line_count")
+def test_galaxy_expression_metadata(target_history: TargetHistory, required_tool: RequiredTool):
+    hda1 = target_history.with_dataset("1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\n13\n14").src_dict
+    execution = required_tool.execute.with_inputs({"input1": hda1})
+    execution.assert_has_single_job.with_single_output.with_contents("3")
+
+
+@requires_tool_id("multi_select")
+def test_multi_select_as_list(required_tool: RequiredTool):
+    execution = required_tool.execute.with_inputs({"select_ex": ["--ex1", "ex2"]})
+    execution.assert_has_single_job.with_output("output").with_contents("--ex1,ex2")
+
+
+@requires_tool_id("multi_select")
+def test_multi_select_optional(required_tool: RequiredTool):
+    execution = required_tool.execute.with_inputs(
+        {
+            "select_ex": ["--ex1"],
+            "select_optional": None,
+        }
+    )
+    job = execution.assert_has_single_job
+    job.assert_has_output("output").with_contents("--ex1")
+    job.assert_has_output("output2").with_contents_stripped("None")
+
+
+@requires_tool_id("identifier_single")
+def test_identifier_outside_map(target_history: TargetHistory, required_tool: RequiredTool):
+    hda = target_history.with_dataset("123", named="Plain HDA")
+    execute = required_tool.execute.with_inputs({"input1": hda.src_dict})
+    execute.assert_has_single_job.assert_has_single_output.with_contents_stripped("Plain HDA")
+
+
+@requires_tool_id("identifier_multiple")
+def test_identifier_in_multiple_reduce(target_history: TargetHistory, required_tool: RequiredTool):
+    hdca = target_history.with_pair(["123", "456"])
+    execute = required_tool.execute.with_inputs({"input1": hdca.src_dict})
+    execute.assert_has_single_job.assert_has_single_output.with_contents_stripped("forward\nreverse")
+
+
+@requires_tool_id("identifier_in_conditional")
+def test_identifier_map_over_multiple_input_in_conditional_legacy_format(
+    target_history: TargetHistory, required_tool: RequiredTool
+):
+    hdca = target_history.with_pair(["123", "456"])
+    execute = required_tool.execute.with_inputs(
+        {
+            "outer_cond|input1": hdca.src_dict,
+        }
+    )
+    execute.assert_has_single_job.assert_has_single_output.with_contents_stripped("forward\nreverse")
+
+
+@requires_tool_id("identifier_in_conditional")
+def test_identifier_map_over_multiple_input_in_conditional_21_01_format(
+    target_history: TargetHistory, required_tool: RequiredTool
+):
+    hdca = target_history.with_pair(["123", "456"])
+    execute = required_tool.execute.with_nested_inputs(
+        {
+            "outer_cond": {
+                "multi_input": True,
+                "input1": hdca.src_dict,
+            },
+        }
+    )
+    execute.assert_has_single_job.assert_has_single_output.with_contents_stripped("forward\nreverse")
+
+
+@requires_tool_id("identifier_multiple_in_repeat")
+def test_identifier_multiple_reduce_in_repeat_new_payload_form(
+    target_history: TargetHistory, required_tool: RequiredTool
+):
+    hdca = target_history.with_pair(["123", "456"])
+    execute = required_tool.execute.with_nested_inputs(
+        {
+            "the_repeat": [{"the_data": {"input1": hdca.src_dict}}],
+        }
+    )
+    execute.assert_has_single_job.assert_has_single_output.with_contents_stripped("forward\nreverse")
+
+
+@requires_tool_id("gx_repeat_boolean_min")
+def test_optional_repeats_with_mins_filled_id(target_history: TargetHistory, required_tool: RequiredTool):
+    # we have a tool test for this but I wanted to verify it wasn't just the
+    # tool test framework filling in a default. Creating a raw request here
+    # verifies that currently select parameters don't require a selection.
+    required_tool.execute.assert_has_single_job.with_single_output.containing("false").containing("length: 2")

lib/galaxy_test/api/test_tools.py

@@ -963,45 +963,6 @@ def test_drill_down_first_by_default(self):
             output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
             assert output1_content.strip() == "parameter: aba"
 
-    @skip_without_tool("multi_select")
-    def test_multi_select_as_list(self):
-        with self.dataset_populator.test_history(require_new=False) as history_id:
-            inputs = {
-                "select_ex": ["--ex1", "ex2"],
-            }
-            response = self._run("multi_select", history_id, inputs, assert_ok=True)
-            output = response["outputs"][0]
-            output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
-
-            assert output1_content == "--ex1,ex2"
-
-    @skip_without_tool("multi_select")
-    def test_multi_select_optional(self):
-        with self.dataset_populator.test_history(require_new=False) as history_id:
-            inputs = {
-                "select_ex": ["--ex1"],
-                "select_optional": None,
-            }
-            response = self._run("multi_select", history_id, inputs, assert_ok=True)
-            output = response["outputs"]
-            output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output[0])
-            output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output[1])
-            assert output1_content.strip() == "--ex1"
-            assert output2_content.strip() == "None", output2_content
-
-    @skip_without_tool("gx_repeat_boolean_min")
-    def test_optional_repeats_with_mins_filled_id(self):
-        # we have a tool test for this but I wanted to verify it wasn't just the
-        # tool test framework filling in a default. Creating a raw request here
-        # verifies that currently select parameters don't require a selection.
-        with self.dataset_populator.test_history(require_new=False) as history_id:
-            inputs: Dict[str, Any] = {}
-            response = self._run("gx_repeat_boolean_min", history_id, inputs, assert_ok=True)
-            output = response["outputs"][0]
-            output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
-            assert "false" in output1_content
-            assert "length: 2" in output1_content
-
     def test_data_column_defaults(self):
         for input_format in ["legacy", "21.01"]:
             tabular_contents = "1\t2\t3\t\n4\t5\t6\n"
@@ -1078,23 +1039,6 @@ def test_library_data_param(self):
             output_multiple_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output[1])
             assert output_multiple_content == "TestData\nTestData\n", output_multiple_content
 
-    @skip_without_tool("multi_data_param")
-    def test_multidata_param(self):
-        with self.dataset_populator.test_history(require_new=False) as history_id:
-            hda1 = dataset_to_param(self.dataset_populator.new_dataset(history_id, content="1\t2\t3"))
-            hda2 = dataset_to_param(self.dataset_populator.new_dataset(history_id, content="4\t5\t6"))
-            inputs = {
-                "f1": {"batch": False, "values": [hda1, hda2]},
-                "f2": {"batch": False, "values": [hda2, hda1]},
-            }
-            response = self._run("multi_data_param", history_id, inputs, assert_ok=True)
-            output1 = response["outputs"][0]
-            output2 = response["outputs"][1]
-            output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
-            output2_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output2)
-            assert output1_content == "1\t2\t3\n4\t5\t6\n", output1_content
-            assert output2_content == "4\t5\t6\n1\t2\t3\n", output2_content
-
     @skip_without_tool("cat1")
     def test_run_cat1(self):
         with self.dataset_populator.test_history(require_new=False) as history_id:
@@ -2032,85 +1976,6 @@ def test_list_selectable_in_multidata_input(self, history_id):
         build = self.dataset_populator.build_tool_state("identifier_multiple", history_id)
         assert len(build["inputs"][0]["options"]["hdca"]) == 1
 
-    @skip_without_tool("identifier_multiple")
-    def test_identifier_in_multiple_reduce(self, history_id):
-        hdca_id = self._build_pair(history_id, ["123", "456"])
-        inputs = {
-            "input1": {"src": "hdca", "id": hdca_id},
-        }
-        create_response = self._run("identifier_multiple", history_id, inputs)
-        self._assert_status_code_is(create_response, 200)
-        create = create_response.json()
-        outputs = create["outputs"]
-        jobs = create["jobs"]
-        implicit_collections = create["implicit_collections"]
-        assert len(jobs) == 1
-        assert len(outputs) == 1
-        assert len(implicit_collections) == 0
-        output1 = outputs[0]
-        output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
-        assert output1_content.strip() == "forward\nreverse"
-
-    @skip_without_tool("identifier_in_conditional")
-    def test_identifier_map_over_multiple_input_in_conditional(self, history_id):
-        hdca_id = self._build_pair(history_id, ["123", "456"])
-        inputs = {
-            "outer_cond|input1": {"src": "hdca", "id": hdca_id},
-        }
-        create_response = self._run("identifier_in_conditional", history_id, inputs)
-        self._assert_status_code_is(create_response, 200)
-        create = create_response.json()
-        outputs = create["outputs"]
-        jobs = create["jobs"]
-        implicit_collections = create["implicit_collections"]
-        assert len(jobs) == 1
-        assert len(outputs) == 1
-        assert len(implicit_collections) == 0
-        output1 = outputs[0]
-        output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
-        assert output1_content.strip() == "forward\nreverse"
-
-    @skip_without_tool("identifier_in_conditional")
-    def test_identifier_map_over_multiple_input_in_conditional_new_payload_form(self, history_id):
-        hdca_id = self._build_pair(history_id, ["123", "456"])
-        inputs = {
-            "outer_cond": {
-                "multi_input": True,
-                "input1": {"id": hdca_id, "src": "hdca"},
-            },
-        }
-        create_response = self._run("identifier_in_conditional", history_id, inputs, input_format="21.01")
-        self._assert_status_code_is(create_response, 200)
-        create = create_response.json()
-        outputs = create["outputs"]
-        jobs = create["jobs"]
-        implicit_collections = create["implicit_collections"]
-        assert len(jobs) == 1
-        assert len(outputs) == 1
-        assert len(implicit_collections) == 0
-        output1 = outputs[0]
-        output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
-        assert output1_content.strip() == "forward\nreverse"
-
-    @skip_without_tool("identifier_multiple_in_repeat")
-    def test_identifier_multiple_reduce_in_repeat_new_payload_form(self, history_id):
-        hdca_id = self._build_pair(history_id, ["123", "456"])
-        inputs = {
-            "the_repeat": [{"the_data": {"input1": {"src": "hdca", "id": hdca_id}}}],
-        }
-        create_response = self._run("identifier_multiple_in_repeat", history_id, inputs, input_format="21.01")
-        self._assert_status_code_is(create_response, 200)
-        create = create_response.json()
-        outputs = create["outputs"]
-        jobs = create["jobs"]
-        implicit_collections = create["implicit_collections"]
-        assert len(jobs) == 1
-        assert len(outputs) == 1
-        assert len(implicit_collections) == 0
-        output1 = outputs[0]
-        output1_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output1)
-        assert output1_content.strip() == "forward\nreverse"
-
     @skip_without_tool("identifier_in_conditional")
     def test_identifier_map_over_input_in_conditional(self, history_id):
         # Run cat tool, so HDA names are different from element identifiers
@@ -2903,42 +2768,6 @@ def test_group_tag_selection_multiple(self, history_id):
         output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output)
         assert output_content.strip() == "123\n456\n456\n0ab"
 
-    @skip_without_tool("expression_forty_two")
-    def test_galaxy_expression_tool_simplest(self):
-        history_id = self.dataset_populator.new_history()
-        run_response = self._run("expression_forty_two", history_id)
-        self._assert_status_code_is(run_response, 200)
-        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
-        output_content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert output_content == "42"
-
-    @skip_without_tool("expression_parse_int")
-    def test_galaxy_expression_tool_simple(self):
-        history_id = self.dataset_populator.new_history()
-        inputs = {
-            "input1": "7",
-        }
-        run_response = self._run("expression_parse_int", history_id, inputs)
-        self._assert_status_code_is(run_response, 200)
-        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
-        output_content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert output_content == "7"
-
-    @skip_without_tool("expression_log_line_count")
-    def test_galaxy_expression_metadata(self):
-        history_id = self.dataset_populator.new_history()
-        new_dataset1 = self.dataset_populator.new_dataset(
-            history_id, content="1\n2\n3\n4\n5\n6\n7\n8\n9\n10\n11\n12\n13\n14"
-        )
-        inputs = {
-            "input1": dataset_to_param(new_dataset1),
-        }
-        run_response = self._run("expression_log_line_count", history_id, inputs)
-        self._assert_status_code_is(run_response, 200)
-        self.dataset_populator.wait_for_history(history_id, assert_ok=True)
-        output_content = self.dataset_populator.get_history_dataset_content(history_id)
-        assert output_content == "3"
-
     @skip_without_tool("cat1")
     def test_run_deferred_dataset(self, history_id):
         details = self.dataset_populator.create_deferred_hda(