From bb7e9150ea9a3a83cc5dca902b7178b2ac97a7cf Mon Sep 17 00:00:00 2001 From: "Michael R. Crusoe" Date: Sat, 13 Nov 2021 16:30:40 +0100 Subject: [PATCH] Fix type hints --- lib/galaxy/datatypes/images.py | 13 ++++++----- lib/galaxy/tools/wrappers.py | 20 ++++++++++++---- lib/galaxy/workflow/modules.py | 28 +++++++++++++++-------- lib/galaxy/workflow/run.py | 3 +-- lib/galaxy_test/api/test_tools_cwl.py | 2 +- lib/galaxy_test/api/test_workflows_cwl.py | 1 + lib/galaxy_test/base/populators.py | 2 +- test/unit/app/tools/test_cwl.py | 5 ++-- 8 files changed, 48 insertions(+), 26 deletions(-) diff --git a/lib/galaxy/datatypes/images.py b/lib/galaxy/datatypes/images.py index d77d3484ab98..06f7ee347cbe 100644 --- a/lib/galaxy/datatypes/images.py +++ b/lib/galaxy/datatypes/images.py @@ -305,12 +305,13 @@ def sniff_prefix(self, file_prefix): ('header_size', 'i4'), ] np_dtype = np.dtype(header_def) - header = np.ndarray( - shape=(), - dtype=np_dtype, - buffer=header_raw) - if header['header_size'] == 1000 and b'TRACK' in header['magic'] and \ - header['version'] == 2 and len(header['dim']) == 3: + header: np.ndarray = np.ndarray(shape=(), dtype=np_dtype, buffer=header_raw) + if ( + header["header_size"] == 1000 + and b"TRACK" in header["magic"] + and header["version"] == 2 + and len(header["dim"]) == 3 + ): return True return False diff --git a/lib/galaxy/tools/wrappers.py b/lib/galaxy/tools/wrappers.py index 8cd6270da256..06696836dd96 100644 --- a/lib/galaxy/tools/wrappers.py +++ b/lib/galaxy/tools/wrappers.py @@ -7,6 +7,7 @@ Any, cast, Dict, + ItemsView, Iterable, Iterator, KeysView, @@ -56,7 +57,7 @@ class ToolParameterValueWrapper: Base class for object that Wraps a Tool Parameter and Value. """ - value: Union[str, List[str]] + value: Union[None, str, List[str], Dict[str, str]] input: "ToolParameter" def __bool__(self) -> bool: @@ -103,10 +104,12 @@ class InputValueWrapper(ToolParameterValueWrapper): Wraps an input so that __str__ gives the "param_dict" representation. """ + value: Optional[Dict[str, str]] + def __init__( self, input: "ToolParameter", - value: str, + value: Dict[str, str], other_values: Optional[Dict[str, str]] = None, ) -> None: self.input = input @@ -172,6 +175,7 @@ class SelectToolParameterWrapper(ToolParameterValueWrapper): """ input: "SelectToolParameter" + value: Union[str, List[str]] class SelectToolParameterFieldWrapper: """ @@ -625,9 +629,13 @@ def __init__( self.collection = collection elements = collection.elements - element_instances = {} + element_instances: Dict[ + str, Union[DatasetCollectionWrapper, DatasetFilenameWrapper] + ] = {} - element_instance_list = [] + element_instance_list: List[ + Union[DatasetCollectionWrapper, DatasetFilenameWrapper] + ] = [] for dataset_collection_element in elements: element_object = dataset_collection_element.element_object element_identifier = dataset_collection_element.element_identifier @@ -662,7 +670,9 @@ def keys(self) -> Union[List[str], KeysView[Any]]: return [] return self.__element_instances.keys() - def items(self): + def items( + self, + ) -> ItemsView[str, Union["DatasetCollectionWrapper", DatasetFilenameWrapper]]: return self.__element_instances.items() @property diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 3caf57902356..86a02f7f94ce 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -873,7 +873,7 @@ def get_inputs(self): cases = [] for param_type in ["text", "integer", "float", "boolean", "color", "field"]: - default_source: Dict[str, Union[int, float, bool, str]] = dict( + default_source: Dict[str, Union[None, int, float, bool, str]] = dict( name="default", label="Default Value", type=param_type ) if param_type == "text": @@ -885,6 +885,7 @@ def get_inputs(self): input_default_value: Union[ TextToolParameter, IntegerToolParameter, + FieldTypeToolParameter, FloatToolParameter, BooleanToolParameter, ColorToolParameter, @@ -1722,7 +1723,7 @@ def decode_runtime_state(self, runtime_state): def evaluate_value_from_expressions(self, progress, step, execution_state, extra_step_state): value_from_expressions = {} - replacements = {} + replacements: Dict[str, str] = {} for key in execution_state.inputs.keys(): step_input = step.inputs_by_name.get(key) @@ -1876,8 +1877,17 @@ def callback(input, prefixed_name, **kwargs): replacement = json.load(f) found_replacement_keys.add(prefixed_name) - is_data = isinstance(input, DataToolParameter) or isinstance(input, DataCollectionToolParameter) or isinstance(input, FieldTypeToolParameter) - if not is_data and getattr(replacement, "history_content_type", None) == "dataset" and getattr(replacement, "ext", None) == "expression.json": + is_data = ( + isinstance(input, DataToolParameter) + or isinstance(input, DataCollectionToolParameter) + or isinstance(input, FieldTypeToolParameter) + ) + if ( + not is_data + and not isinstance(replacement, NoReplacement) + and getattr(replacement, "history_content_type", None) == "dataset" + and getattr(replacement, "ext", None) == "expression.json" + ): if isinstance(replacement, model.HistoryDatasetAssociation): if not replacement.dataset.in_ready_state(): why = "dataset [%s] is needed for non-data connection and is non-ready" % replacement.id @@ -1891,11 +1901,11 @@ def callback(input, prefixed_name, **kwargs): if isinstance(input, FieldTypeToolParameter): if isinstance(replacement, model.HistoryDatasetAssociation): - replacement = {"src": "hda", "value": replacement} + return {"src": "hda", "value": replacement} elif isinstance(replacement, model.HistoryDatasetCollectionAssociation): - replacement = {"src": "hdca", "value": replacement} + return {"src": "hdca", "value": replacement} elif replacement is not NO_REPLACEMENT: - replacement = {"src": "json", "value": replacement} + return {"src": "json", "value": replacement} return replacement @@ -1938,9 +1948,9 @@ def expression_callback(input, prefixed_name, **kwargs): if prefixed_name in expression_replacements: expression_replacement = expression_replacements[prefixed_name] if isinstance(input, FieldTypeToolParameter): - replacement = {"src": "json", "value": expression_replacement} + return {"src": "json", "value": expression_replacement} else: - replacement = expression_replacement + return expression_replacement return replacement diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 3ff2c85aff64..65fb468f258e 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -367,11 +367,10 @@ def replacement_for_input_connections(self, step, input_dict, connections): else: raise NotImplementedError() - ephemeral_collection = modules.EphemeralCollection( + return modules.EphemeralCollection( collection=collection, history=self.workflow_invocation.history, ) - replacement = ephemeral_collection return replacement diff --git a/lib/galaxy_test/api/test_tools_cwl.py b/lib/galaxy_test/api/test_tools_cwl.py index 8b12d971a3c4..3e411534fbb5 100644 --- a/lib/galaxy_test/api/test_tools_cwl.py +++ b/lib/galaxy_test/api/test_tools_cwl.py @@ -327,7 +327,7 @@ def test_any1_file(self): test_data_directory="test/functional/tools/cwl_tools/v1.0/v1.0/", ) output1_content = self.dataset_populator.get_history_dataset_content(run_object.history_id) - self.dataset_populator._summarize_history_errors(run_object.history_id) + self.dataset_populator._summarize_history(run_object.history_id) assert output1_content == '"File"', "[%s]" % output1_content @skip_without_tool("any1") diff --git a/lib/galaxy_test/api/test_workflows_cwl.py b/lib/galaxy_test/api/test_workflows_cwl.py index 5cfa08e7c1a5..a5419db9b451 100644 --- a/lib/galaxy_test/api/test_workflows_cwl.py +++ b/lib/galaxy_test/api/test_workflows_cwl.py @@ -11,6 +11,7 @@ class BaseCwlWorklfowTestCase(BaseWorkflowsApiTestCase): + history_id: str allow_path_paste = True require_admin_user = True diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 5a19a7f5302b..9ad5b74e55b7 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -472,7 +472,7 @@ def run_conformance_test(self, version, doc): directory = os.path.join(CWL_TOOL_DIRECTORY, version) tool = os.path.join(directory, test["tool"]) job_path = test.get("job") - job = None + job: Optional[Dict[str, str]] = None if job_path is not None: job_path = os.path.join(directory, job_path) else: diff --git a/test/unit/app/tools/test_cwl.py b/test/unit/app/tools/test_cwl.py index 81c0e511820b..6c60e8b4ae65 100644 --- a/test/unit/app/tools/test_cwl.py +++ b/test/unit/app/tools/test_cwl.py @@ -2,6 +2,7 @@ import os import shutil import tempfile +from typing import Dict from unittest import TestCase from uuid import uuid4 @@ -546,12 +547,12 @@ def tearDown(self): def test_default_data_inputs(self): self._init_tool(tool_path=_cwl_tool_path("v1.0/v1.0/default_path.cwl")) hda = self._new_hda() - errors = {} + errors: Dict[str, str] = {} cwl_inputs = { "file1": {"src": "hda", "id": self.app.security.encode_id(hda.id)} } inputs = self.tool.inputs_from_dict({"inputs": cwl_inputs, "inputs_representation": "cwl"}) - populated_state = {} + populated_state: Dict[str, str] = {} populate_state(self.trans, self.tool.inputs, inputs, populated_state, errors) wrapped_params = WrappedParameters(self.trans, self.tool, populated_state) input_json = to_cwl_job(self.tool, wrapped_params.params, self.test_directory)