Skip to content

Commit

Permalink
Rework test case validation schema.
Browse files Browse the repository at this point in the history
  • Loading branch information
jmchilton committed Aug 27, 2024
1 parent 10ef0a0 commit fd4b182
Show file tree
Hide file tree
Showing 10 changed files with 391 additions and 55 deletions.
2 changes: 2 additions & 0 deletions lib/galaxy/tool_util/parameters/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@
repeat_inputs_to_array,
validate_explicit_conditional_test_value,
visit_input_values,
VISITOR_NO_REPLACEMENT,
)

__all__ = (
Expand Down Expand Up @@ -116,6 +117,7 @@
"keys_starting_with",
"visit_input_values",
"repeat_inputs_to_array",
"VISITOR_NO_REPLACEMENT",
"decode",
"encode",
"WorkflowStepToolState",
Expand Down
24 changes: 23 additions & 1 deletion lib/galaxy/tool_util/parameters/case.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@
from packaging.version import Version

from galaxy.tool_util.parser.interface import (
TestCollectionDef,
ToolSource,
ToolSourceTest,
ToolSourceTestInput,
ToolSourceTestInputs,
xml_data_input_to_json,
XmlTestCollectionDefDict,
)
from galaxy.util import asbool
from .factory import input_models_for_tool_source
Expand All @@ -25,6 +28,7 @@
ConditionalWhen,
DataCollectionParameterModel,
DataColumnParameterModel,
DataParameterModel,
FloatParameterModel,
IntegerParameterModel,
RepeatParameterModel,
Expand Down Expand Up @@ -249,8 +253,26 @@ def _merge_into_state(
else:
test_input = _input_for(state_path, inputs)
if test_input is not None:
input_value: Any
if isinstance(tool_input, (DataCollectionParameterModel,)):
input_value = test_input.get("attributes", {}).get("collection")
input_value = TestCollectionDef.from_dict(
cast(XmlTestCollectionDefDict, test_input.get("attributes", {}).get("collection"))
).test_format_to_dict()
elif isinstance(tool_input, (DataParameterModel,)):
data_tool_input = cast(DataParameterModel, tool_input)
if data_tool_input.multiple:
value = test_input["value"]
input_value_list = []
if value:
test_input_values = cast(str, value).split(",")
for test_input_value in test_input_values:
instance_test_input = test_input.copy()
instance_test_input["value"] = test_input_value
input_value = xml_data_input_to_json(test_input)
input_value_list.append(input_value)
input_value = input_value_list
else:
input_value = xml_data_input_to_json(test_input)
else:
input_value = test_input["value"]
input_value = legacy_from_string(tool_input, input_value, warnings, profile)
Expand Down
13 changes: 6 additions & 7 deletions lib/galaxy/tool_util/parameters/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@
from galaxy.exceptions import RequestParameterInvalidException
from galaxy.tool_util.parser.interface import (
DrillDownOptionsDict,
TestCollectionDefDict,
JsonTestCollectionDefDict,
JsonTestDatasetDefDict,
)
from ._types import (
cast_as_type,
Expand Down Expand Up @@ -312,9 +313,9 @@ def py_type_internal(self) -> Type:
def py_type_test_case(self) -> Type:
base_model: Type
if self.multiple:
base_model = str
base_model = list_type(JsonTestDatasetDefDict)
else:
base_model = str
base_model = JsonTestDatasetDefDict
return optional_if_needed(base_model, self.optional)

def pydantic_template(self, state_representation: StateRepresentationT) -> DynamicModelInformation:
Expand Down Expand Up @@ -372,7 +373,7 @@ def pydantic_template(self, state_representation: StateRepresentationT) -> Dynam
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, TestCollectionDefDict)
return dynamic_model_information_from_py_type(self, JsonTestCollectionDefDict)
else:
raise NotImplementedError(
f"Have not implemented data collection parameter models for state representation {state_representation}"
Expand Down Expand Up @@ -1179,9 +1180,7 @@ def to_simple_model(input_parameter: Union[ToolParameterModel, ToolParameterT])
return cast(ToolParameterT, input_parameter)


def simple_input_models(
parameters: Union[List[ToolParameterModel], List[ToolParameterT]]
) -> Iterable[ToolParameterT]:
def simple_input_models(parameters: Union[List[ToolParameterModel], List[ToolParameterT]]) -> Iterable[ToolParameterT]:
return [to_simple_model(m) for m in parameters]


Expand Down
202 changes: 182 additions & 20 deletions lib/galaxy/tool_util/parser/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from os.path import join
from typing import (
Any,
cast,
Dict,
List,
Optional,
Expand Down Expand Up @@ -550,7 +551,8 @@ def parse_input_sources(self) -> List[InputSource]:
"""Return a list of InputSource objects."""


TestCollectionDefElementObject = Union["TestCollectionDefDict", "ToolSourceTestInput"]
AnyTestCollectionDefDict = Union["JsonTestCollectionDefDict", "XmlTestCollectionDefDict"]
TestCollectionDefElementObject = Union[AnyTestCollectionDefDict, "ToolSourceTestInput"]
TestCollectionAttributeDict = Dict[str, Any]
CollectionType = str

Expand All @@ -560,24 +562,154 @@ class TestCollectionDefElementDict(TypedDict):
element_definition: TestCollectionDefElementObject


class TestCollectionDefDict(TypedDict):
class TestCollectionDefElementInternal(TypedDict):
element_identifier: str
element_definition: Union["TestCollectionDef", "ToolSourceTestInput"]


# two versions of collection inputs can be parsed out, XmlTestCollectionDefDict is historically
# used by tools and Galaxy internals and exposed in the API via the test definition endpoints for
# tool execution. JsonTestCollectionDefDict is the format consumed by Planemo that mirrors a CWL
# way of defining inputs.
class XmlTestCollectionDefDict(TypedDict):
model_class: Literal["TestCollectionDef"]
attributes: TestCollectionAttributeDict
collection_type: CollectionType
elements: List[TestCollectionDefElementDict]
name: str


JsonTestDatasetDefDict = TypedDict(
"JsonTestDatasetDefDict",
{
"class": Literal["File"],
"path": NotRequired[Optional[str]],
"location": NotRequired[Optional[str]],
"name": NotRequired[Optional[str]],
"dbkey": NotRequired[Optional[str]],
"filetype": NotRequired[Optional[str]],
"composite_data": NotRequired[Optional[List[str]]],
"tags": NotRequired[Optional[List[str]]],
},
)

JsonTestCollectionDefElementDict = Union[
"JsonTestCollectionDefDatasetElementDict", "JsonTestCollectionDefCollectionElementDict"
]
JsonTestCollectionDefDatasetElementDict = TypedDict(
"JsonTestCollectionDefDatasetElementDict",
{
"identifier": str,
"class": Literal["File"],
"path": NotRequired[Optional[str]],
"location": NotRequired[Optional[str]],
"name": NotRequired[Optional[str]],
"dbkey": NotRequired[Optional[str]],
"filetype": NotRequired[Optional[str]],
"composite_data": NotRequired[Optional[List[str]]],
"tags": NotRequired[Optional[List[str]]],
},
)
BaseJsonTestCollectionDefCollectionElementDict = TypedDict(
"BaseJsonTestCollectionDefCollectionElementDict",
{
"class": Literal["Collection"],
"collection_type": str,
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
},
)
JsonTestCollectionDefCollectionElementDict = TypedDict(
"JsonTestCollectionDefCollectionElementDict",
{
"identifier": str,
"class": Literal["Collection"],
"collection_type": str,
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
},
)
JsonTestCollectionDefDict = TypedDict(
"JsonTestCollectionDefDict",
{
"class": Literal["Collection"],
"collection_type": str,
"elements": NotRequired[Optional[List[JsonTestCollectionDefElementDict]]],
"name": NotRequired[Optional[str]],
},
)


def xml_data_input_to_json(xml_input: ToolSourceTestInput) -> "JsonTestDatasetDefDict":
attributes = xml_input["attributes"]
as_dict: JsonTestDatasetDefDict = {
"class": "File",
}
value = xml_input["value"]
if value:
as_dict["path"] = value
_copy_if_exists(attributes, as_dict, "location")
_copy_if_exists(attributes, as_dict, "dbkey")
_copy_if_exists(attributes, as_dict, "ftype", "filetype")
_copy_if_exists(attributes, as_dict, "composite_data", only_if_value=True)
tags = attributes.get("tags")
if tags:
as_dict["tags"] = [t.strip() for t in tags.split(",")]
return as_dict


def _copy_if_exists(attributes, as_dict, name: str, as_name: Optional[str] = None, only_if_value: bool = False):
if name in attributes:
value = attributes[name]
if not value and only_if_value:
return
if as_name is None:
as_name = name
as_dict[as_name] = value


class TestCollectionDef:
__test__ = False # Prevent pytest from discovering this class (issue #12071)
elements: List[TestCollectionDefElementInternal]

def __init__(self, attrib, name, collection_type, elements):
self.attrib = attrib
self.collection_type = collection_type
self.elements = elements
self.name = name

def to_dict(self) -> TestCollectionDefDict:
def _test_format_to_dict(self) -> "BaseJsonTestCollectionDefCollectionElementDict":

def to_element(xml_element_dict: "TestCollectionDefElementInternal") -> "JsonTestCollectionDefElementDict":
identifier = xml_element_dict["element_identifier"]
element_object = xml_element_dict["element_definition"]
as_dict: JsonTestCollectionDefElementDict

if isinstance(element_object, TestCollectionDef):
as_dict = JsonTestCollectionDefCollectionElementDict(
identifier=identifier, **element_object._test_format_to_dict()
)
else:
as_dict = JsonTestCollectionDefDatasetElementDict(
identifier=identifier,
**xml_data_input_to_json(cast(ToolSourceTestInput, element_object)),
)
return as_dict

test_format_dict = BaseJsonTestCollectionDefCollectionElementDict(
{
"class": "Collection",
"elements": list(map(to_element, self.elements)),
"collection_type": self.collection_type,
}
)
return test_format_dict

def test_format_to_dict(self) -> JsonTestCollectionDefDict:
test_format_dict = JsonTestCollectionDefDict(**self._test_format_to_dict())
if self.name:
test_format_dict["name"] = self.name
return test_format_dict

def to_dict(self) -> XmlTestCollectionDefDict:
def element_to_dict(element_dict):
element_identifier, element_def = element_dict["element_identifier"], element_dict["element_definition"]
if isinstance(element_def, TestCollectionDef):
Expand All @@ -596,23 +728,53 @@ def element_to_dict(element_dict):
}

@staticmethod
def from_dict(as_dict: TestCollectionDefDict):
assert as_dict["model_class"] == "TestCollectionDef"

def element_from_dict(element_dict):
if "element_definition" not in element_dict:
raise Exception(f"Invalid element_dict {element_dict}")
element_def = element_dict["element_definition"]
if element_def.get("model_class", None) == "TestCollectionDef":
element_def = TestCollectionDef.from_dict(element_def)
return {"element_identifier": element_dict["element_identifier"], "element_definition": element_def}

return TestCollectionDef(
attrib=as_dict["attributes"],
name=as_dict["name"],
elements=list(map(element_from_dict, as_dict["elements"] or [])),
collection_type=as_dict["collection_type"],
)
def from_dict(
as_dict: Union[AnyTestCollectionDefDict, JsonTestCollectionDefCollectionElementDict]
) -> "TestCollectionDef":
if "model_class" in as_dict:
xml_as_dict = cast(XmlTestCollectionDefDict, as_dict)
assert xml_as_dict["model_class"] == "TestCollectionDef"

def element_from_dict(element_dict) -> TestCollectionDefElementInternal:
if "element_definition" not in element_dict:
raise Exception(f"Invalid element_dict {element_dict}")
element_def = element_dict["element_definition"]
if element_def.get("model_class", None) == "TestCollectionDef":
element_def = TestCollectionDef.from_dict(element_def)
return {"element_identifier": element_dict["element_identifier"], "element_definition": element_def}

return TestCollectionDef(
attrib=xml_as_dict["attributes"],
name=xml_as_dict.get("name", "Unnamed Collection"),
elements=list(map(element_from_dict, xml_as_dict["elements"] or [])),
collection_type=xml_as_dict["collection_type"],
)
else:
json_as_dict = cast(JsonTestCollectionDefDict, as_dict)

def element_from_dict_json(
element_dict: JsonTestCollectionDefElementDict,
) -> TestCollectionDefElementInternal:
element_class = element_dict.get("class")
identifier = element_dict["identifier"]
element_def: Union[TestCollectionDef, ToolSourceTestInput]
if element_class == "Collection":
collection_element_dict = cast(JsonTestCollectionDefCollectionElementDict, element_dict)
element_def = TestCollectionDef.from_dict(collection_element_dict)
else:
dataset_element_dict = cast(JsonTestCollectionDefDatasetElementDict, element_dict)
value = dataset_element_dict["path"] # todo handle location
name = dataset_element_dict.get("name") or "Unnamed Collection"
element_def = {"name": name, "value": value, "attributes": {}}
return TestCollectionDefElementInternal(element_identifier=identifier, element_definition=element_def)

elements = list(map(element_from_dict_json, json_as_dict.get("elements") or []))
return TestCollectionDef(
attrib={},
name=json_as_dict.get("name") or "Unnamed Collection",
elements=elements,
collection_type=json_as_dict["collection_type"],
)

def collect_inputs(self):
inputs = []
Expand Down
6 changes: 3 additions & 3 deletions lib/galaxy/tool_util/parser/xml.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@
PageSource,
PagesSource,
RequiredFiles,
TestCollectionDefDict,
TestCollectionDefElementDict,
TestCollectionDefElementObject,
TestCollectionOutputDef,
Expand All @@ -58,6 +57,7 @@
ToolSourceTestOutputAttributes,
ToolSourceTestOutputs,
ToolSourceTests,
XmlTestCollectionDefDict,
XrefDict,
)
from .output_actions import ToolOutputActionGroup
Expand Down Expand Up @@ -1001,7 +1001,7 @@ def __parse_inputs_elems(test_elem, i) -> ToolSourceTestInputs:
return raw_inputs


def _test_collection_def_dict(elem: Element) -> TestCollectionDefDict:
def _test_collection_def_dict(elem: Element) -> XmlTestCollectionDefDict:
elements: List[TestCollectionDefElementDict] = []
attrib: Dict[str, Any] = _element_to_dict(elem)
collection_type = attrib["type"]
Expand All @@ -1017,7 +1017,7 @@ def _test_collection_def_dict(elem: Element) -> TestCollectionDefDict:
element_definition = __parse_param_elem(element)
elements.append({"element_identifier": element_identifier, "element_definition": element_definition})

return TestCollectionDefDict(
return XmlTestCollectionDefDict(
model_class="TestCollectionDef",
attributes=attrib,
collection_type=collection_type,
Expand Down
Loading

0 comments on commit fd4b182

Please sign in to comment.