WIP: test case stuff...

Author: jmchilton

lib/galaxy/tool_util/parameters/case.py

@@ -57,12 +57,11 @@ def test_case_state(
         if input_name not in by_name:
             raise Exception(f"Cannot find tool parameter for {input_name}")
         tool_parameter_model = by_name[input_name]
-        input_value = input["value"]
-        input_value = legacy_from_string(tool_parameter_model, input_value, warnings, profile)
-        if isinstance(tool_parameter_model, (DataParameterModel,)):
-            pass
-        elif isinstance(tool_parameter_model, (DataCollectionParameterModel,)):
-            pass
+        if isinstance(tool_parameter_model, (DataCollectionParameterModel,)):
+            input_value = input.get("attributes", {}).get("collection")
+        else:
+            input_value = input["value"]
+            input_value = legacy_from_string(tool_parameter_model, input_value, warnings, profile)
 
         state[input_name] = input_value
 

lib/galaxy/tool_util/parameters/models.py

@@ -37,6 +37,7 @@
 )
 
 from galaxy.exceptions import RequestParameterInvalidException
+from galaxy.tool_util.parser.interface import TestCollectionDict
 from ._types import (
     cast_as_type,
     is_optional,
@@ -368,6 +369,8 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
             return dynamic_model_information_from_py_type(self, type(None), requires_value=False)
         elif state_representation == "workflow_step_linked":
             return dynamic_model_information_from_py_type(self, ConnectedValue)
+        elif state_representation == "test_case_xml":
+            return dynamic_model_information_from_py_type(self, TestCollectionDict)
         else:
             raise NotImplementedError(f"Have not implemented data collection parameter models for state representation {state_representation}")
 

lib/galaxy/tool_util/parser/interface.py

@@ -19,6 +19,7 @@
 import packaging.version
 from pydantic import BaseModel
 from typing_extensions import (
+    Literal,
     NotRequired,
     TypedDict,
 )
@@ -376,7 +377,7 @@ def paths_and_modtimes(self):
             paths_and_modtimes[self.source_path] = os.path.getmtime(self.source_path)
         return paths_and_modtimes
 
-    def parse_tests_to_dict(self) -> ToolSourceTests:
+    def parse_tests_to_dict(self, for_json: bool = False) -> ToolSourceTests:
         return {"tests": []}
 
     def __str__(self):
@@ -525,6 +526,23 @@ def parse_input_sources(self) -> List[InputSource]:
         """Return a list of InputSource objects."""
 
 
+TestCollectionAttributeDict = Dict[str, Any]
+CollectionType = str
+
+
+class TestCollectionDictElement(TypedDict):
+    element_identifier: str
+    element_definition: Union["TestCollectionDict", "ToolSourceTestInput"]
+
+
+class TestCollectionDict(TypedDict):
+    model_class: Literal["TestCollectionDef"] = "TestCollectionDef"
+    attributes: TestCollectionAttributeDict
+    collection_type: CollectionType
+    elements: List[TestCollectionDictElement]
+    name: str
+
+
 class TestCollectionDef:
     __test__ = False  # Prevent pytest from discovering this class (issue #12071)
 

lib/galaxy/tool_util/parser/xml.py

@@ -650,15 +650,15 @@ def macro_paths(self):
     def source_path(self):
         return self._source_path
 
-    def parse_tests_to_dict(self) -> ToolSourceTests:
+    def parse_tests_to_dict(self, for_json: bool = False) -> ToolSourceTests:
         tests_elem = self.root.find("tests")
         tests: List[ToolSourceTest] = []
         rval: ToolSourceTests = dict(tests=tests)
 
         if tests_elem is not None:
             for i, test_elem in enumerate(tests_elem.findall("test")):
                 profile = self.parse_profile()
-                tests.append(_test_elem_to_dict(test_elem, i, profile))
+                tests.append(_test_elem_to_dict(test_elem, i, profile, for_json=for_json))
 
         return rval
 
@@ -715,11 +715,11 @@ def parse_creator(self):
         return creators
 
 
-def _test_elem_to_dict(test_elem, i, profile=None) -> ToolSourceTest:
+def _test_elem_to_dict(test_elem, i, profile=None, for_json=False) -> ToolSourceTest:
     rval: ToolSourceTest = dict(
         outputs=__parse_output_elems(test_elem),
         output_collections=__parse_output_collection_elems(test_elem, profile=profile),
-        inputs=__parse_input_elems(test_elem, i),
+        inputs=__parse_input_elems(test_elem, i, for_json=for_json),
         expect_num_outputs=test_elem.get("expect_num_outputs"),
         command=__parse_assert_list_from_elem(test_elem.find("assert_command")),
         command_version=__parse_assert_list_from_elem(test_elem.find("assert_command_version")),
@@ -734,9 +734,9 @@ def _test_elem_to_dict(test_elem, i, profile=None) -> ToolSourceTest:
     return rval
 
 
-def __parse_input_elems(test_elem, i) -> ToolSourceTestInputs:
+def __parse_input_elems(test_elem, i, for_json=False) -> ToolSourceTestInputs:
     __expand_input_elems(test_elem)
-    return __parse_inputs_elems(test_elem, i)
+    return __parse_inputs_elems(test_elem, i, for_json=for_json)
 
 
 def __parse_output_elems(test_elem) -> ToolSourceTestOutputs:
@@ -982,15 +982,15 @@ def _copy_to_dict_if_present(elem, rval, attributes):
     return rval
 
 
-def __parse_inputs_elems(test_elem, i) -> ToolSourceTestInputs:
+def __parse_inputs_elems(test_elem, i, for_json=False) -> ToolSourceTestInputs:
     raw_inputs: ToolSourceTestInputs = []
     for param_elem in test_elem.findall("param"):
-        raw_inputs.append(__parse_param_elem(param_elem, i))
+        raw_inputs.append(__parse_param_elem(param_elem, i, for_json=for_json))
 
     return raw_inputs
 
 
-def __parse_param_elem(param_elem, i=0) -> ToolSourceTestInput:
+def __parse_param_elem(param_elem, i=0, for_json=False) -> ToolSourceTestInput:
     attrib: ToolSourceTestInputAttributes = dict(param_elem.attrib)
     if "values" in attrib:
         value = attrib["values"].split(",")
@@ -1028,7 +1028,8 @@ def __parse_param_elem(param_elem, i=0) -> ToolSourceTestInput:
             elif child.tag == "edit_attributes":
                 attrib["edit_attributes"].append(child)
             elif child.tag == "collection":
-                attrib["collection"] = TestCollectionDef.from_xml(child, __parse_param_elem)
+                collection = TestCollectionDef.from_xml(child, lambda elem: __parse_param_elem(elem, for_json=for_json))
+                attrib["collection"] = collection if not for_json else collection.to_dict()
         if composite_data_name:
             # Composite datasets need implicit renaming;
             # inserted at front of list so explicit declarations

lib/galaxy/tool_util/parser/yaml.py

@@ -189,7 +189,7 @@ def _parse_output_collection(self, tool, name, output_dict):
         )
         return output_collection
 
-    def parse_tests_to_dict(self) -> ToolSourceTests:
+    def parse_tests_to_dict(self, for_json: bool = False) -> ToolSourceTests:
         tests: List[ToolSourceTest] = []
         rval: ToolSourceTests = dict(tests=tests)
 

test/unit/tool_util/test_parameter_test_cases.py

@@ -35,7 +35,7 @@ def test_validate_framework_test_tools():
         tool_source = get_tool_source(tool_path)
         parsed_tool = parse_tool(tool_source)
         profile = tool_source.parse_profile()
-        test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict()["tests"]
+        test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict(for_json=True)["tests"]
         for test_case in test_cases:
             test_case_state_and_warnings = case_state(test_case, parsed_tool.inputs, profile)
             tool_state = test_case_state_and_warnings.tool_state
@@ -47,7 +47,7 @@ def validate_test_cases_for(tool_name: str) -> List[List[str]]:
     tool_parameter_bundle = parameter_bundle_for_file(tool_name)
     tool_source = parameter_tool_source(tool_name)
     profile = tool_source.parse_profile()
-    test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict()["tests"]
+    test_cases: List[ToolSourceTest] = tool_source.parse_tests_to_dict(for_json=True)["tests"]
     warnings_by_test = []
     for test_case in test_cases:
         test_case_state_and_warnings = case_state(test_case, tool_parameter_bundle, profile)