mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-13 00:44:55 +08:00
Nicola Soranzo
7f89e0eab9
Found by running: ``` make pyupgrade ruff check --fix . ruff check --preview --unsafe-fixes --select=F401 --fix . make format make update-client-api-schema ``` The bulk of the changes are replacements of typing deprecated aliases such as `Dict` and `List` with the standard library classes. These aliases became redundant in Python 3.9 when the corresponding pre-existing classes were enhanced to support `[]`, see https://docs.python.org/3/library/typing.html#deprecated-aliases Also: - Update the `pyupgrade` Makefile target to use `--py38-plus` for the packages needed by Pulsar, including the new galaxy-tool-util-models package.
257 lines
11 KiB
Python
257 lines
11 KiB
Python
from functools import partial
|
|
from typing import (
|
|
Callable,
|
|
List,
|
|
Optional,
|
|
)
|
|
|
|
import yaml
|
|
|
|
from galaxy.exceptions import RequestParameterInvalidException
|
|
from galaxy.tool_util.parameters import (
|
|
decode,
|
|
encode,
|
|
RawStateDict,
|
|
RequestInternalToolState,
|
|
RequestToolState,
|
|
ToolParameterBundleModel,
|
|
validate_internal_job,
|
|
validate_internal_landing_request,
|
|
validate_internal_request,
|
|
validate_internal_request_dereferenced,
|
|
validate_landing_request,
|
|
validate_request,
|
|
validate_test_case,
|
|
validate_workflow_step,
|
|
validate_workflow_step_linked,
|
|
ValidationFunctionT,
|
|
)
|
|
from galaxy.tool_util.parameters.json import (
|
|
to_json_schema,
|
|
to_json_schema_string,
|
|
)
|
|
from galaxy.tool_util.unittest_utils.parameters import (
|
|
parameter_bundle_for_file,
|
|
parameter_bundle_for_framework_tool,
|
|
)
|
|
from galaxy.util.resources import resource_string
|
|
|
|
|
|
def specification_object():
|
|
try:
|
|
yaml_str = resource_string(__name__, "parameter_specification.yml")
|
|
except AttributeError:
|
|
# hack for the main() function below where this file is interpreted as part of the
|
|
# Galaxy tree.
|
|
yaml_str = open("test/unit/tool_util/parameter_specification.yml").read()
|
|
return yaml.safe_load(yaml_str)
|
|
|
|
|
|
def framework_tool_checks():
|
|
# Extend parameter_specification with some extra tests against existing framework tools.
|
|
# There is something beautiful about a targeted tool for every parameter feature but realistically
|
|
# we've been doing a version of the for a decade with tool tests and we can leverage those also.
|
|
try:
|
|
yaml_str = resource_string(__name__, "framework_tool_checks.yml")
|
|
except AttributeError:
|
|
# hack for the main() function below where this file is interpreted as part of the
|
|
# Galaxy tree.
|
|
yaml_str = open("test/unit/tool_util/framework_tool_checks.yml").read()
|
|
return yaml.safe_load(yaml_str)
|
|
|
|
|
|
def test_specification():
|
|
parameter_spec = specification_object()
|
|
for file in parameter_spec.keys():
|
|
_test_file(file, parameter_spec)
|
|
|
|
|
|
def test_framework_tool_checks():
|
|
parameter_spec = framework_tool_checks()
|
|
for file in parameter_spec.keys():
|
|
_test_file(file, parameter_spec, parameter_bundle_for_framework_tool(f"{file}.xml"))
|
|
|
|
|
|
def test_single():
|
|
# _test_file("gx_int")
|
|
# _test_file("gx_float")
|
|
# _test_file("gx_boolean")
|
|
# _test_file("gx_int_optional")
|
|
# _test_file("gx_float_optional")
|
|
# _test_file("gx_conditional_boolean")
|
|
# _test_file("gx_conditional_conditional_boolean")
|
|
_test_file("gx_conditional_boolean_checked")
|
|
|
|
|
|
def _test_file(file: str, specification=None, parameter_bundle: Optional[ToolParameterBundleModel] = None):
|
|
spec = specification or specification_object()
|
|
combos = spec[file]
|
|
if parameter_bundle is None:
|
|
parameter_bundle = parameter_bundle_for_file(file)
|
|
assert parameter_bundle
|
|
|
|
assertion_functions = {
|
|
"request_valid": _assert_requests_validate,
|
|
"request_invalid": _assert_requests_invalid,
|
|
"request_internal_valid": _assert_internal_requests_validate,
|
|
"request_internal_invalid": _assert_internal_requests_invalid,
|
|
"request_internal_dereferenced_valid": _assert_internal_requests_dereferenced_validate,
|
|
"request_internal_dereferenced_invalid": _assert_internal_requests_dereferenced_invalid,
|
|
"landing_request_valid": _assert_landing_requests_validate,
|
|
"landing_request_invalid": _assert_landing_requests_invalid,
|
|
"landing_request_internal_valid": _assert_internal_landing_requests_validate,
|
|
"landing_request_internal_invalid": _assert_internal_landing_requests_invalid,
|
|
"job_internal_valid": _assert_internal_jobs_validate,
|
|
"job_internal_invalid": _assert_internal_jobs_invalid,
|
|
"test_case_xml_valid": _assert_test_cases_validate,
|
|
"test_case_xml_invalid": _assert_test_cases_invalid,
|
|
"workflow_step_valid": _assert_workflow_steps_validate,
|
|
"workflow_step_invalid": _assert_workflow_steps_invalid,
|
|
"workflow_step_linked_valid": _assert_workflow_steps_linked_validate,
|
|
"workflow_step_linked_invalid": _assert_workflow_steps_linked_invalid,
|
|
}
|
|
|
|
for valid_or_invalid, tests in combos.items():
|
|
assertion_function = assertion_functions[valid_or_invalid]
|
|
assertion_function(parameter_bundle, tests)
|
|
|
|
# Assume request validation will work here.
|
|
if "request_internal_valid" not in combos and "request_valid" in combos:
|
|
_assert_internal_requests_validate(parameter_bundle, combos["request_valid"])
|
|
if "request_internal_invalid" not in combos and "request_invalid" in combos:
|
|
_assert_internal_requests_invalid(parameter_bundle, combos["request_invalid"])
|
|
|
|
|
|
def _for_each(test: Callable, parameters: ToolParameterBundleModel, requests: List[RawStateDict]) -> None:
|
|
for request in requests:
|
|
test(parameters, request)
|
|
|
|
|
|
def model_assertion_function_factory(validate_function: ValidationFunctionT, what: str):
|
|
|
|
def _assert_validates(parameters: ToolParameterBundleModel, request: RawStateDict) -> None:
|
|
try:
|
|
validate_function(parameters, request)
|
|
except RequestParameterInvalidException as e:
|
|
raise AssertionError(f"Parameters {parameters} failed to validate {what} {request}. {e}")
|
|
|
|
def _assert_invalid(parameters: ToolParameterBundleModel, request: RawStateDict) -> None:
|
|
exc = None
|
|
try:
|
|
validate_function(parameters, request)
|
|
except RequestParameterInvalidException as e:
|
|
exc = e
|
|
|
|
if exc is None:
|
|
raise AssertionError(
|
|
f"Parameters {parameters} didn't result in validation error on {what} {request} as expected."
|
|
)
|
|
|
|
return _assert_validates, _assert_invalid
|
|
|
|
|
|
_assert_request_validates, _assert_request_invalid = model_assertion_function_factory(validate_request, "request")
|
|
_assert_internal_request_validates, _assert_internal_request_invalid = model_assertion_function_factory(
|
|
validate_internal_request, "internal request"
|
|
)
|
|
_assert_internal_request_dereferenced_validates, _assert_internal_request_dereferenced_invalid = (
|
|
model_assertion_function_factory(validate_internal_request_dereferenced, "dereferenced internal request")
|
|
)
|
|
_assert_internal_job_validates, _assert_internal_job_invalid = model_assertion_function_factory(
|
|
validate_internal_job, "internal job description"
|
|
)
|
|
_assert_test_case_validates, _assert_test_case_invalid = model_assertion_function_factory(
|
|
validate_test_case, "XML derived test case"
|
|
)
|
|
_assert_workflow_step_validates, _assert_workflow_step_invalid = model_assertion_function_factory(
|
|
validate_workflow_step, "workflow step tool state (unlinked)"
|
|
)
|
|
_assert_workflow_step_linked_validates, _assert_workflow_step_linked_invalid = model_assertion_function_factory(
|
|
validate_workflow_step_linked, "linked workflow step tool state"
|
|
)
|
|
_assert_landing_request_validates, _assert_landing_request_invalid = model_assertion_function_factory(
|
|
validate_landing_request, "landing request"
|
|
)
|
|
_assert_internal_landing_request_validates, _assert_internal_landing_request_invalid = model_assertion_function_factory(
|
|
validate_internal_landing_request, " internallanding request"
|
|
)
|
|
|
|
_assert_requests_validate = partial(_for_each, _assert_request_validates)
|
|
_assert_requests_invalid = partial(_for_each, _assert_request_invalid)
|
|
_assert_internal_requests_validate = partial(_for_each, _assert_internal_request_validates)
|
|
_assert_internal_requests_invalid = partial(_for_each, _assert_internal_request_invalid)
|
|
_assert_internal_requests_dereferenced_validate = partial(_for_each, _assert_internal_request_dereferenced_validates)
|
|
_assert_internal_requests_dereferenced_invalid = partial(_for_each, _assert_internal_request_dereferenced_invalid)
|
|
_assert_internal_jobs_validate = partial(_for_each, _assert_internal_job_validates)
|
|
_assert_internal_jobs_invalid = partial(_for_each, _assert_internal_job_invalid)
|
|
_assert_test_cases_validate = partial(_for_each, _assert_test_case_validates)
|
|
_assert_test_cases_invalid = partial(_for_each, _assert_test_case_invalid)
|
|
_assert_workflow_steps_validate = partial(_for_each, _assert_workflow_step_validates)
|
|
_assert_workflow_steps_invalid = partial(_for_each, _assert_workflow_step_invalid)
|
|
_assert_workflow_steps_linked_validate = partial(_for_each, _assert_workflow_step_linked_validates)
|
|
_assert_workflow_steps_linked_invalid = partial(_for_each, _assert_workflow_step_linked_invalid)
|
|
_assert_landing_requests_validate = partial(_for_each, _assert_landing_request_validates)
|
|
_assert_landing_requests_invalid = partial(_for_each, _assert_landing_request_invalid)
|
|
_assert_internal_landing_requests_validate = partial(_for_each, _assert_internal_landing_request_validates)
|
|
_assert_internal_landing_requests_invalid = partial(_for_each, _assert_internal_landing_request_invalid)
|
|
|
|
|
|
def decode_val(val: str) -> int:
|
|
assert val == "abcdabcd"
|
|
return 5
|
|
|
|
|
|
def test_decode_gx_data():
|
|
input_bundle = parameter_bundle_for_file("gx_data")
|
|
|
|
request_tool_state = RequestToolState({"parameter": {"src": "hda", "id": "abcdabcd"}})
|
|
request_internal_tool_state = decode(request_tool_state, input_bundle, decode_val)
|
|
assert request_internal_tool_state.input_state["parameter"]["id"] == 5
|
|
assert request_internal_tool_state.input_state["parameter"]["src"] == "hda"
|
|
|
|
|
|
def test_decode_gx_int():
|
|
input_bundle = parameter_bundle_for_file("gx_int")
|
|
|
|
request_tool_state = RequestToolState({"parameter": 5})
|
|
request_internal_tool_state = decode(request_tool_state, input_bundle, decode_val)
|
|
assert request_internal_tool_state.input_state["parameter"] == 5
|
|
|
|
|
|
def test_json_schema_for_conditional():
|
|
input_bundle = parameter_bundle_for_file("gx_conditional_boolean")
|
|
tool_state = RequestToolState.parameter_model_for(input_bundle)
|
|
json_schema = to_json_schema(tool_state)
|
|
assert json_schema["$defs"]["When_test_parameter_False"]["properties"]["test_parameter"]["const"] is False
|
|
assert json_schema["$defs"]["When_test_parameter_True"]["properties"]["test_parameter"]["const"] is True
|
|
assert "test_parameter" not in json_schema["$defs"]["When_test_parameter___absent"]["properties"]
|
|
assert to_json_schema_string(tool_state)
|
|
|
|
|
|
def test_encode_gx_data():
|
|
input_bundle = parameter_bundle_for_file("gx_data")
|
|
|
|
def encode_val(val: int) -> str:
|
|
assert val == 5
|
|
return "abcdabcd"
|
|
|
|
request_internal_tool_state = RequestInternalToolState({"parameter": {"src": "hda", "id": 5}})
|
|
request_tool_state = encode(request_internal_tool_state, input_bundle, encode_val)
|
|
assert request_tool_state.input_state["parameter"]["id"] == "abcdabcd"
|
|
assert request_tool_state.input_state["parameter"]["src"] == "hda"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parameter_spec = specification_object()
|
|
parameter_models_json = {}
|
|
for file in parameter_spec.keys():
|
|
tool_parameter_model = parameter_bundle_for_file(file)
|
|
parameter_models_json[file] = tool_parameter_model.model_dump()
|
|
yaml_str = yaml.safe_dump(parameter_models_json)
|
|
with open("client/src/components/Tool/parameter_models.yml", "w") as f:
|
|
f.write("# auto generated file for JavaScript testing, do not modify manually\n")
|
|
f.write("# -----\n")
|
|
f.write('# PYTHONPATH="lib" python test/unit/tool_util/test_parameter_specification.py\n')
|
|
f.write("# -----\n")
|
|
f.write(yaml_str)
|