Test case WIP...

Author: jmchilton

lib/galaxy/tool_util/parameters/case.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass
+from typing import (
+    Any,
+    Dict,
+    List,
+)
+
+from galaxy.tool_util.parser.interface import ToolSourceTest
+from galaxy.util import asbool
+from .models import (
+    BooleanParameterModel,
+    DataCollectionParameterModel,
+    DataParameterModel,
+    FloatParameterModel,
+    IntegerParameterModel,
+    parameters_by_name,
+    ToolParameterBundle,
+    ToolParameterT,
+)
+from .state import TestCaseToolState
+
+
+@dataclass
+class TestCaseStateAndWarnings:
+    tool_state: TestCaseToolState
+    warnings: List[str]
+
+
+def legacy_from_string(parameter: ToolParameterT, value: str, warnings: List[str], profile: str) -> Any:
+    """Convert string values in XML test cases into typed variants.
+
+    This should only be used when parsing XML test cases into a TestCaseToolState object.
+    We have to maintain backward compatibility on these for older Galaxy tool profile versions.
+    """
+    is_string = isinstance(value, str)
+    result_value: Any = value
+    if is_string and isinstance(parameter, (IntegerParameterModel,)):
+        warnings.append(
+            f"Implicitly converted {parameter.name} to an integer from a string value, please use 'value_json' to define this test input parameter value instead."
+        )
+        result_value = int(value)
+    elif is_string and isinstance(parameter, (FloatParameterModel,)):
+        warnings.append(
+            f"Implicitly converted {parameter.name} to a floating point number from a string value, please use 'value_json' to define this test input parameter value instead."
+        )
+        result_value = float(value)
+    elif is_string and isinstance(parameter, (BooleanParameterModel,)):
+        warnings.append(
+            f"Implicitly converted {parameter.name} to a boolean from a string value, please use 'value_json' to define this test input parameter value instead."
+        )
+        result_value = asbool(value)
+    return result_value
+
+
+def test_case_state(
+    test_dict: ToolSourceTest, tool_parameter_bundle: ToolParameterBundle, profile: str
+) -> TestCaseStateAndWarnings:
+    warnings: List[str] = []
+    inputs = test_dict["inputs"]
+    state = {}
+    by_name = parameters_by_name(tool_parameter_bundle)
+    for input in inputs:
+        input_name = input["name"]
+        if input_name not in by_name:
+            raise Exception(f"Cannot find tool parameter for {input_name}")
+        tool_parameter_model = by_name[input_name]
+        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
+
+    tool_state = TestCaseToolState(state)
+    tool_state.validate(tool_parameter_bundle)
+    return TestCaseStateAndWarnings(tool_state, warnings)

lib/galaxy/tool_util/parameters/models.py

@@ -37,7 +37,10 @@
 )
 
 from galaxy.exceptions import RequestParameterInvalidException
-from galaxy.tool_util.parser.interface import DrillDownOptionsDict
+from galaxy.tool_util.parser.interface import (
+    DrillDownOptionsDict,
+    TestCollectionDict,
+)
 from ._types import (
     cast_as_type,
     is_optional,
@@ -58,7 +61,7 @@
 # + request_internal: This is a pydantic model to validate what Galaxy expects to find in the database,
 # in particular dataset and collection references should be decoded integers.
 StateRepresentationT = Literal[
-    "request", "request_internal", "job_internal", "test_case", "workflow_step", "workflow_step_linked"
+    "request", "request_internal", "job_internal", "test_case_xml", "workflow_step", "workflow_step_linked"
 ]
 
 
@@ -311,9 +314,9 @@ def py_type_internal(self) -> Type:
     def py_type_test_case(self) -> Type:
         base_model: Type
         if self.multiple:
-            base_model = MultiDataRequestInternal
+            base_model = str
         else:
-            base_model = DataTestCaseValue
+            base_model = str
         return optional_if_needed(base_model, self.optional)
 
     def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
@@ -325,7 +328,7 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
             )
         elif state_representation == "job_internal":
             return dynamic_model_information_from_py_type(self, self.py_type_internal)
-        elif state_representation == "test_case":
+        elif state_representation == "test_case_xml":
             return dynamic_model_information_from_py_type(self, self.py_type_test_case)
         elif state_representation == "workflow_step":
             return dynamic_model_information_from_py_type(self, type(None), requires_value=False)
@@ -369,6 +372,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}"
@@ -529,17 +534,23 @@ class SelectParameterModel(BaseGalaxyToolParameterModelDefinition):
     options: Optional[List[LabelValue]] = None
     multiple: bool
 
-    def py_type_if_required(self, allow_connections=False) -> Type:
+    def py_type_if_required(self, allow_connections: bool = False, expect_list: bool = True) -> Type:
         if self.options is not None:
             literal_options: List[Type] = [cast_as_type(Literal[o.value]) for o in self.options]
             py_type = union_type(literal_options)
         else:
             py_type = StrictStr
         if self.multiple:
             if allow_connections:
-                py_type = list_type(allow_connected_value(py_type))
+                if expect_list:
+                    py_type = list_type(allow_connected_value(py_type))
+                else:
+                    py_type = allow_connected_value(py_type)
             else:
-                py_type = list_type(py_type)
+                if expect_list:
+                    py_type = list_type(py_type)
+                else:
+                    py_type = py_type
         elif allow_connections:
             py_type = allow_connected_value(py_type)
         return py_type
@@ -559,6 +570,10 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
         elif state_representation == "workflow_step_linked":
             py_type = self.py_type_if_required(allow_connections=True)
             return dynamic_model_information_from_py_type(self, optional_if_needed(py_type, self.optional))
+        elif state_representation == "test_case_xml":
+            # in a YAML test case representation this can be string, in XML we are still expecting a comma separated string
+            py_type = self.py_type_if_required(allow_connections=False, expect_list=False)
+            return dynamic_model_information_from_py_type(self, optional_if_needed(py_type, self.optional))
         else:
             return dynamic_model_information_from_py_type(self, self.py_type)
 
@@ -1018,9 +1033,13 @@ class ToolParameterBundleModel(BaseModel):
     input_models: List[ToolParameterT]
 
 
-def parameters_by_name(tool_parameter_bundle: ToolParameterBundle) -> Dict[str, ToolParameterT]:
+def parameters_by_name(inputs: Union[Iterable[ToolParameterModel], Iterable[ToolParameterT], ToolParameterBundle]) -> Dict[str, ToolParameterT]:
     as_dict = {}
-    for input_model in simple_input_models(tool_parameter_bundle.input_models):
+    if hasattr(inputs, "input_models"):
+        inputs_list = simple_input_models(cast(ToolParameterBundle, inputs.input_models))
+    else:
+        inputs_list = cast(Union[Iterable[ToolParameterModel], Iterable[ToolParameterT]], inputs)
+    for input_model in inputs_list:
         as_dict[input_model.name] = input_model
     return as_dict
 
@@ -1058,7 +1077,7 @@ def create_job_internal_model(tool: ToolParameterBundle, name: str = "DynamicMod
 
 
 def create_test_case_model(tool: ToolParameterBundle, name: str = "DynamicModelForTool") -> Type[BaseModel]:
-    return create_field_model(tool.input_models, name, "test_case")
+    return create_field_model(tool.input_models, name, "test_case_xml")
 
 
 def create_workflow_step_model(tool: ToolParameterBundle, name: str = "DynamicModelForTool") -> Type[BaseModel]:

lib/galaxy/tool_util/parameters/state.py

@@ -17,6 +17,7 @@
     create_job_internal_model,
     create_request_internal_model,
     create_request_model,
+    create_test_case_model,
     create_workflow_step_linked_model,
     create_workflow_step_model,
     StateRepresentationT,
@@ -91,12 +92,12 @@ def _parameter_model_for(cls, input_models: ToolParameterBundle) -> Type[BaseMod
 
 
 class TestCaseToolState(ToolState):
-    state_representation: Literal["test_case"] = "test_case"
+    state_representation: Literal["test_case_xml"] = "test_case_xml"
 
     @classmethod
     def _parameter_model_for(cls, input_models: ToolParameterBundle) -> Type[BaseModel]:
         # implement a test case model...
-        return create_request_internal_model(input_models)
+        return create_test_case_model(input_models)
 
 
 class WorkflowStepToolState(ToolState):

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):
@@ -545,6 +546,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

@@ -654,15 +654,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
 
@@ -719,11 +719,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")),
@@ -738,9 +738,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:
@@ -986,15 +986,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(",")
@@ -1032,7 +1032,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)
 

lib/galaxy/tool_util/unittest_utils/__init__.py

@@ -31,5 +31,9 @@ def get_content(filename: Optional[str]) -> bytes:
     return get_content
 
 
+def functional_test_tool_directory() -> str:
+    return os.path.join(galaxy_directory(), "test/functional/tools")
+
+
 def functional_test_tool_path(test_path: str) -> str:
-    return os.path.join(galaxy_directory(), "test/functional/tools", test_path)
+    return os.path.join(functional_test_tool_directory(), test_path)