galaxy/test/unit/jobs/test_mapper.py

import uuid

from galaxy.jobs import (
    HasResourceParameters,
    JobDestination,
)
from galaxy.jobs.mapper import (
    ERROR_MESSAGE_NO_RULE_FUNCTION,
    ERROR_MESSAGE_RULE_FUNCTION_NOT_FOUND,
    JobRunnerMapper,
)
from galaxy.util import bunch
from . import test_rules

WORKFLOW_UUID = uuid.uuid1().hex
TOOL_JOB_DESTINATION = JobDestination()
DYNAMICALLY_GENERATED_DESTINATION = JobDestination()


def test_static_mapping():
    mapper = __mapper()
    assert mapper.get_job_destination({}) is TOOL_JOB_DESTINATION


def test_caching():
    mapper = __mapper()
    mapper.get_job_destination({})
    mapper.get_job_destination({})
    assert mapper.job_wrapper.tool.call_count == 1


def test_dynamic_mapping():
    mapper = __mapper(__dynamic_destination(dict(function="upload")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == "local_runner"


def test_dynamic_mapping_priorities():
    mapper = __mapper(__dynamic_destination(dict(function="tophat")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    # Next line verifies we using definition in 20_instance.py instead of
    # 10_site.py.
    assert mapper.job_config.rule_response == "instance_dest_id"


def test_dynamic_mapping_defaults_to_tool_id_as_rule():
    mapper = __mapper(__dynamic_destination())
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == "tool1_dest_id"


def test_dynamic_mapping_job_conf_params():
    mapper = __mapper(__dynamic_destination(dict(function="check_job_conf_params", param1="7")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == "sent_7_dest_id"


def test_dynamic_mapping_function_parameters():
    mapper = __mapper(__dynamic_destination(dict(function="check_rule_params")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == "all_passed"


def test_dynamic_mapping_resource_parameters():
    mapper = __mapper(__dynamic_destination(dict(function="check_resource_params")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == "have_resource_params"


def test_dynamic_mapping_workflow_invocation_parameter():
    mapper = __mapper(__dynamic_destination(dict(function="check_workflow_invocation_uuid")))
    assert mapper.get_job_destination({}) is DYNAMICALLY_GENERATED_DESTINATION
    assert mapper.job_config.rule_response == WORKFLOW_UUID


def test_dynamic_mapping_no_function():
    dest = __dynamic_destination(dict())
    mapper = __mapper(dest)
    mapper.job_wrapper.tool.all_ids = ["no_such_function"]
    error_message = ERROR_MESSAGE_NO_RULE_FUNCTION % dest
    __assert_mapper_errors_with_message(mapper, error_message)


def test_dynamic_mapping_missing_function():
    dest = __dynamic_destination(dict(function="missing_func"))
    mapper = __mapper(dest)
    mapper.job_wrapper.tool.all_ids = ["no_such_function"]
    error_message = ERROR_MESSAGE_RULE_FUNCTION_NOT_FOUND % ("missing_func")
    __assert_mapper_errors_with_message(mapper, error_message)


def __assert_mapper_errors_with_message(mapper, message):
    exception = None
    try:
        mapper.get_job_destination({})
    except Exception as e:
        exception = e
    assert exception
    assert str(exception) == message, "%s != %s" % (str(exception), message)


def __mapper(tool_job_destination=TOOL_JOB_DESTINATION):
    job_wrapper = MockJobWrapper(tool_job_destination)
    job_config = MockJobConfig()

    mapper = JobRunnerMapper(
        job_wrapper,
        {},
        job_config
    )
    mapper.rules_module = test_rules
    return mapper


def __dynamic_destination(params={}):
    return JobDestination(runner="dynamic", params=params)


class MockJobConfig(object):

    def __init__(self):
        self.rule_response = None
        self.dynamic_params = None

    def get_destination(self, rep):
        # Called to transform dynamic job destination rule response
        # from destination id/runner url into a dynamic job destination.
        self.rule_response = rep
        return DYNAMICALLY_GENERATED_DESTINATION


class MockJobWrapper(HasResourceParameters):

    def __init__(self, tool_job_destination):
        self.tool = MockTool(tool_job_destination)
        self.job_id = 12345
        self.app = object()

    def is_mock_job_wrapper(self):
        return True

    def get_job(self):
        raw_params = {
            "threshold": 8,
            "__workflow_invocation_uuid__": WORKFLOW_UUID,
        }

        def get_param_values(app, ignore_errors):
            assert app == self.app
            params = raw_params.copy()
            params["__job_resource"] = {
                "__job_resource__select": "True",
                "memory": "8gb"
            }
            return params

        return bunch.Bunch(
            user=bunch.Bunch(
                id=6789,
                email="test@example.com"
            ),
            raw_param_dict=lambda: raw_params,
            get_param_values=get_param_values
        )


class MockTool(object):

    def __init__(self, tool_job_destination):
        self.id = "testtoolshed/devteam/tool1/23abcd13123"
        self.call_count = 0
        self.tool_job_destination = tool_job_destination
        self.all_ids = ["testtoolshed/devteam/tool1/23abcd13123", "tool1"]

    def get_job_destination(self, params):
        self.call_count += 1
        return self.tool_job_destination

    def is_mock_tool(self):
        return True