galaxy/test/unit/tool_util/test_parameter_specification.py

import json
from collections.abc import Callable
from functools import partial

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_job_runtime,
    validate_landing_request,
    validate_relaxed_request,
    validate_request,
    validate_test_case,
    validate_test_case_json,
    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.tool_util_models.parameters import (
    _INFINITY_SENTINEL,
    _NEG_INFINITY_SENTINEL,
    FloatParameterModel,
)
from galaxy.util.json import safe_dumps
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: ToolParameterBundleModel | None = 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 = {
        "relaxed_request_valid": _assert_relaxed_requests_validate,
        "relaxed_request_invalid": _assert_relaxed_requests_invalid,
        "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,
        "job_runtime_valid": _assert_job_runtimes_validate,
        "job_runtime_invalid": _assert_job_runtimes_invalid,
        "test_case_xml_valid": _assert_test_cases_validate,
        "test_case_xml_invalid": _assert_test_cases_invalid,
        "test_case_json_valid": _assert_test_case_jsons_validate,
        "test_case_json_invalid": _assert_test_case_jsons_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():
        if valid_or_invalid not in assertion_functions:
            continue
        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_relaxed_request_validates, _assert_relaxed_request_invalid = model_assertion_function_factory(
    validate_relaxed_request, "relaxed request"
)
_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_job_runtime_validates, _assert_job_runtime_invalid = model_assertion_function_factory(
    validate_job_runtime, "job runtime"
)
_assert_test_case_validates, _assert_test_case_invalid = model_assertion_function_factory(
    validate_test_case, "XML derived test case"
)
_assert_test_case_json_validates, _assert_test_case_json_invalid = model_assertion_function_factory(
    validate_test_case_json, "raw JSON 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_relaxed_requests_validate = partial(_for_each, _assert_relaxed_request_validates)
_assert_relaxed_requests_invalid = partial(_for_each, _assert_relaxed_request_invalid)
_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_job_runtimes_validate = partial(_for_each, _assert_job_runtime_validates)
_assert_job_runtimes_invalid = partial(_for_each, _assert_job_runtime_invalid)
_assert_test_cases_validate = partial(_for_each, _assert_test_case_validates)
_assert_test_cases_invalid = partial(_for_each, _assert_test_case_invalid)
_assert_test_case_jsons_validate = partial(_for_each, _assert_test_case_json_validates)
_assert_test_case_jsons_invalid = partial(_for_each, _assert_test_case_json_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"


def test_float_parameter_model_infinity_sentinel_roundtrip():
    """Regression test: FloatParameterModel must survive a safe_dumps/json.loads roundtrip.

    Galaxy's safe_dumps() serialises float('inf') as '__Infinity__' and
    float('-inf') as '__-Infinity__' to produce valid JSON.  When the tool
    test API (GET /api/tools/{id}/test_data) returns request_schema.parameters
    and any float parameter carries a default of inf (e.g. meme -evt), the
    resulting JSON contains that sentinel.  Planemo then calls
    ToolParameterBundleModel(parameters=<deserialized dicts>) which must not
    fail even though Pydantic sees a string rather than a float.
    """
    for value, expected in [
        (float("inf"), float("inf")),
        (float("-inf"), float("-inf")),
    ]:
        # 1. Build the model definition
        m = FloatParameterModel(type="float", name="test", value=value)
        # 2. Dump to a plain dict (contains Python float inf/-inf)
        d = m.model_dump()
        assert d["value"] == value

        # 3. Serialize through Galaxy's JSON encoder — inf becomes a sentinel string
        galaxy_json = safe_dumps([d])
        sentinel = _INFINITY_SENTINEL if value == float("inf") else _NEG_INFINITY_SENTINEL
        assert sentinel in galaxy_json

        # 4. Deserialize: sentinel is now a plain string in the dict
        dicts = json.loads(galaxy_json)
        assert dicts[0]["value"] == sentinel

        # 5. Re-parse via ToolParameterBundleModel — this must not raise
        bundle = ToolParameterBundleModel(parameters=dicts)

        # 6. The parsed model should carry the correct Python float
        parsed_param = bundle.parameters[0]
        assert isinstance(parsed_param, FloatParameterModel)
        assert parsed_param.value == expected


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)