opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-11-02 03:33:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Move `get_param_values() method from Job to Tool` class

Browse Source 
because the ``Job`` model class shouldn't have to deal with ``app``.

Also:
- Remove ``app`` parameter from ``Tool.params_from_strings()``
- Improve type annotations
- Clean up unused code
This commit is contained in:
Nicola Soranzo
2025-06-10 02:30:44 +01:00
parent 9bd2a328f1
commit 010bbdd0ae
15 changed files with 157 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/__init__.py
+7 -5
View File
@@ -984,6 +984,8 @@ class JobConfiguration(ConfiguresHandlers):
class HasResourceParameters:
tool: Optional["Tool"]
def get_resource_parameters(self, job=None):
# Find the dymically inserted resource parameters and give them
# to rule.
@@ -991,8 +993,8 @@ class HasResourceParameters:
if job is None:
job = self.get_job()
app = self.app
param_values = job.get_param_values(app, ignore_errors=True)
assert self.tool
param_values = self.tool.get_param_values(job, ignore_errors=True)
resource_params = {}
try:
resource_params_raw = param_values["__job_resource"]
@@ -1234,7 +1236,7 @@ class MinimalJobWrapper(HasResourceParameters):
"""
job = _job or self.get_job()
param_dict = {p.name: p.value for p in job.parameters}
param_dict = self.tool.params_from_strings(param_dict, self.app)
param_dict = self.tool.params_from_strings(param_dict)
return param_dict
@property
@@ -1297,7 +1299,7 @@ class MinimalJobWrapper(HasResourceParameters):
# search again, now we know tool doesn't require name match
param_dump = {p.name: p.value for p in job.parameters if not p.name.startswith("__")}
assert self.tool
params = self.tool.params_from_strings(param_dump, self.app)
params = self.tool.params_from_strings(param_dump)
json_internal = params_to_json_internal(self.tool.inputs, params, self.app)
job_to_copy = self.app.job_search.by_tool_input(
job.user,
@@ -2700,7 +2702,7 @@ class MinimalJobWrapper(HasResourceParameters):
"""
if self.tool and self.tool.id == "upload1":
job = self.get_job()
param_dict = job.get_param_values(self.app)
param_dict = self.tool.get_param_values(job)
return param_dict.get("link_data_only") == "link_to_files"
else:
# The tool is unavailable, we try to move the outputs.
lib/galaxy/jobs/dynamic_tool_destination.py
+32 -38
View File
@@ -8,13 +8,28 @@ import os
import re
import sys
from functools import reduce
from typing import Set
from typing import (
Dict,
List,
Optional,
Set,
TYPE_CHECKING,
Union,
)
import numpy as np
import yaml
from galaxy.util import parse_xml
if TYPE_CHECKING:
from galaxy.model import (
DatasetInstance,
Job,
)
from galaxy.structured_app import MinimalManagerApp
from galaxy.tools import Tool
__version__ = "1.1.0"
# log to galaxy's logger
@@ -66,7 +81,7 @@ class ScannerError(Exception):
pass
def get_keys_from_dict(dl, keys_list):
def get_keys_from_dict(dl: Union[Dict, List], keys_list: List) -> None:
"""
This function builds a list using the keys from nest dictionaries
"""
@@ -1251,30 +1266,15 @@ def str_to_bytes(size):
return curr_size
def importer(test):
"""
Uses Mock galaxy for testing or real galaxy for production
@type test: bool
@param test: True when being run from a test
"""
global JobDestination
global JobMappingException
if test:
class JobDestination:
def __init__(self, *kwd):
self.id = kwd.get("id")
self.nativeSpec = kwd.get("params")["nativeSpecification"]
self.runner = kwd.get("runner")
from galaxy.jobs.mapper import JobMappingException
else:
from galaxy.jobs import JobDestination
from galaxy.jobs.mapper import JobMappingException
def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, job_conf_path=None):
def map_tool_to_destination(
job: "Job",
app: "MinimalManagerApp",
tool: "Tool",
user_email: Optional[str],
test: bool = False,
path: Optional[str] = None,
job_conf_path: Optional[str] = None,
):
"""
Dynamically allocate resources
@@ -1283,7 +1283,7 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
@param tool: current tool
@type test: bool
@param test: True when running in test mode
@param test: unused and ignored
@type path: str
@param path: path to tool_destinations.yml
@@ -1291,7 +1291,7 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
@type job_conf_path: str
@param job_conf_path: path to job_conf.xml
"""
importer(test)
from galaxy.jobs.mapper import JobMappingException
# set verbose to True by default, just in case (some tests fail without
# this due to how the tests apparently work)
@@ -1313,7 +1313,7 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
raise JobMappingException(e)
# Get all inputs from tool and databases
inp_data = {da.name: da.dataset for da in job.input_datasets}
inp_data: Dict[str, DatasetInstance] = {da.name: da.dataset for da in job.input_datasets}
inp_data.update([(da.name, da.dataset) for da in job.input_library_datasets])
if config is not None and str(tool.old_id) in config["tools"]:
@@ -1333,13 +1333,6 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
num_input_datasets = 0
if filesize_rule_present or records_rule_present or num_input_datasets_rule_present:
# Loop through the database and look for amount of records
try:
for line in inp_db:
if line[0] == ">":
records += 1
except NameError:
pass
# Loops through each input file and adds the size to the total
# or looks through db for records
for da in inp_data:
@@ -1505,12 +1498,12 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
matched = True
elif rule["rule_type"] == "arguments":
options = job.get_param_values(app)
options = tool.get_param_values(job)
matched = True
# check if the args in the config file are available
for arg in rule["arguments"]:
arg_dict = {arg: rule["arguments"][arg]}
arg_keys_list = []
arg_keys_list: List = []
get_keys_from_dict(arg_dict, arg_keys_list)
try:
options_value = reduce(dict.__getitem__, arg_keys_list, options)
@@ -1581,6 +1574,7 @@ def map_tool_to_destination(job, app, tool, user_email, test=False, path=None, j
if fail_message:
raise JobMappingException(fail_message)
else:
assert matched_rule
raise JobMappingException(matched_rule["fail_message"])
if config is not None:
lib/galaxy/jobs/mapper.py
-5
View File
@@ -140,11 +140,6 @@ class JobRunnerMapper:
return expand_function(**actual_args)
def __job_params(self, job):
app = self.job_wrapper.app
param_values = job.get_param_values(app, ignore_errors=True)
return param_values
def __convert_url_to_destination(self, url: str):
"""
Job runner URLs are deprecated, but dynamic mapper functions may still
lib/galaxy/managers/jobs.py
+4 -4
View File
@@ -1097,7 +1097,7 @@ def summarize_destination_params(trans, job):
return destination_params
def summarize_job_parameters(trans: ProvidesUserContext, job: Job):
def summarize_job_parameters(trans: ProvidesUserContext, job: Job) -> Dict[str, Any]:
"""Produce a dict-ified version of job parameters ready for tabular rendering.
Precondition: the caller has verified the job is accessible to the user
@@ -1233,12 +1233,12 @@ def summarize_job_parameters(trans: ProvidesUserContext, job: Job):
# Load parameter objects, if a parameter type has changed, it's possible for the value to no longer be valid
if tool:
try:
params_objects = job.get_param_values(app, ignore_errors=False)
params_objects = tool.get_param_values(job, ignore_errors=False)
except Exception:
params_objects = job.get_param_values(app, ignore_errors=True)
params_objects = tool.get_param_values(job, ignore_errors=True)
# use different param_objects in the following line, since we want to display original values as much as possible
upgrade_messages = tool.check_and_update_param_values(
job.get_param_values(app, ignore_errors=True), trans, update_values=False
tool.get_param_values(job, ignore_errors=True), trans, update_values=False
)
has_parameter_errors = True
parameters = inputs_recursive(tool.inputs, params_objects, depth=1, upgrade_messages=upgrade_messages)
lib/galaxy/model/__init__.py
+3 -18
View File
@@ -2008,21 +2008,6 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable):
self.state_history.append(JobStateHistory(self))
return True
def get_param_values(self, app, ignore_errors=False):
"""
Read encoded parameter values from the database and turn back into a
dict of tool parameter values.
"""
param_dict = self.raw_param_dict()
tool = app.toolbox.get_tool(
self.tool_id,
tool_version=self.tool_version,
tool_uuid=self.dynamic_tool and self.dynamic_tool.uuid,
user=self.user,
)
param_dict = tool.params_from_strings(param_dict, app, ignore_errors=ignore_errors)
return param_dict
def raw_param_dict(self):
param_dict = {p.name: p.value for p in self.parameters}
return param_dict
@@ -2436,7 +2421,7 @@ class Task(Base, JobLike, RepresentById):
"""
param_dict = {p.name: p.value for p in self.job.parameters}
tool = app.toolbox.get_tool(self.job.tool_id, tool_version=self.job.tool_version)
param_dict = tool.params_from_strings(param_dict, app)
param_dict = tool.params_from_strings(param_dict)
return param_dict
def get_id_tag(self):
@@ -5471,9 +5456,9 @@ class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnot
"""
Create a a new HDA and associate it with the given history.
"""
# FIXME: sa_session is must be passed to DataSetInstance if the create_dataset
# FIXME: sa_session must be passed to DataSetInstance if the create_dataset
# parameter is True so that the new object can be flushed. Is there a better way?
DatasetInstance.__init__(self, sa_session=sa_session, **kwd)
super().__init__(sa_session=sa_session, **kwd)
self.hid = hid
# Relationships
self.history = history
lib/galaxy/tools/__init__.py
+40 -20
View File
@@ -53,6 +53,8 @@ from galaxy.job_execution.output_collect import (
)
from galaxy.metadata import get_metadata_compute_strategy
from galaxy.model import (
History,
HistoryDatasetAssociation,
HistoryDatasetCollectionAssociation,
Job,
JobToOutputDatasetAssociation,
@@ -2213,7 +2215,7 @@ class Tool(UsesDictVisibleKeys):
self,
trans,
incoming: ToolRequestT,
history: Optional[model.History] = None,
history: Optional[History] = None,
use_cached_job: bool = DEFAULT_USE_CACHED_JOB,
preferred_object_store_id: Optional[str] = DEFAULT_PREFERRED_OBJECT_STORE_ID,
input_format: InputFormatT = "legacy",
@@ -2293,7 +2295,7 @@ class Tool(UsesDictVisibleKeys):
trans,
rerun_remap_job_id: Optional[int],
execution_slice: ExecutionSlice,
history: model.History,
history: History,
execution_cache: ToolExecutionCache,
completed_job: Optional[Job],
collection_info: Optional[MatchingCollections],
@@ -2398,7 +2400,7 @@ class Tool(UsesDictVisibleKeys):
self,
trans,
incoming: Optional[ToolStateJobInstancePopulatedT] = None,
history: Optional[model.History] = None,
history: Optional[History] = None,
set_output_hid: bool = DEFAULT_SET_OUTPUT_HID,
flush_job: bool = True,
):
@@ -2424,7 +2426,7 @@ class Tool(UsesDictVisibleKeys):
self,
trans,
incoming: Optional[ToolStateJobInstancePopulatedT] = None,
history: Optional[model.History] = None,
history: Optional[History] = None,
rerun_remap_job_id: Optional[int] = DEFAULT_RERUN_REMAP_JOB_ID,
execution_cache: Optional[ToolExecutionCache] = None,
dataset_collection_elements: Optional[DatasetCollectionElementsSliceT] = None,
@@ -2468,8 +2470,16 @@ class Tool(UsesDictVisibleKeys):
def params_to_strings(self, params: ToolStateJobInstancePopulatedT, app, nested=False):
return params_to_strings(self.inputs, params, app, nested)
def params_from_strings(self, params, app, ignore_errors=False):
return params_from_strings(self.inputs, params, app, ignore_errors)
def params_from_strings(self, params: Dict, ignore_errors: bool = False) -> Dict:
return params_from_strings(self.inputs, params, self.app, ignore_errors)
def get_param_values(self, job: Job, ignore_errors: bool = False) -> Dict:
"""
Read encoded parameter values from the database and turn back into a
dict of tool parameter values.
"""
param_dict = job.raw_param_dict()
return self.params_from_strings(param_dict, ignore_errors=ignore_errors)
def check_and_update_param_values(self, values, trans, update_values=True, workflow_building_mode=False):
"""
@@ -2862,7 +2872,14 @@ class Tool(UsesDictVisibleKeys):
return tool_dict
def to_json(self, trans, kwd=None, job=None, workflow_building_mode=False, history=None):
def to_json(
self,
trans,
kwd=None,
job: Optional[Job] = None,
workflow_building_mode=False,
history: Optional[History] = None,
):
"""
Recursively creates a tool dictionary containing repeats, dynamic options and updated states.
"""
@@ -2875,6 +2892,7 @@ class Tool(UsesDictVisibleKeys):
# We don't need a history when exporting a workflow for the workflow editor or when downloading a workflow
history = history or trans.get_history()
if history is None and job is not None:
assert job.history
history = self.history_manager.get_owned(job.history.id, trans.user, current_history=trans.history)
# We can show the tool form if the current user is anonymous and doesn't have a history
user = trans.get_user()
@@ -2889,7 +2907,7 @@ class Tool(UsesDictVisibleKeys):
tool_warnings = None
if job:
try:
job_params = job.get_param_values(self.app, ignore_errors=True)
job_params = self.get_param_values(job, ignore_errors=True)
tool_warnings = self.check_and_update_param_values(job_params, request_context, update_values=True)
self._map_source_to_history(request_context, self.inputs, job_params)
tool_message = self._compare_tool_version(job)
@@ -2975,31 +2993,33 @@ class Tool(UsesDictVisibleKeys):
other_values=other_values,
)
def _map_source_to_history(self, trans, tool_inputs, params):
def _map_source_to_history(self, trans: WorkRequestContext, tool_inputs: "ToolInputsT", params: Dict) -> None:
# Need to remap dataset parameters. Job parameters point to original
# dataset used; parameter should be the analygous dataset in the
# current history.
history = trans.history
assert history
# Create index for hdas.
hda_source_dict = {}
hda_source_dict: Dict[Union[int, str], HistoryDatasetAssociation] = {}
for hda in history.datasets:
key = f"{hda.hid}_{hda.dataset.id}"
hda_source_dict[hda.dataset.id] = hda_source_dict[key] = hda
# Ditto for dataset collections.
hdca_source_dict = {}
hdca_source_dict: Dict[Union[int, str], HistoryDatasetCollectionAssociation] = {}
for hdca in history.dataset_collections:
key = f"{hdca.hid}_{hdca.collection.id}"
hdca_source_dict[hdca.collection.id] = hdca_source_dict[key] = hdca
# Map dataset or collection to current history
def map_to_history(value):
id = None
source = None
if isinstance(value, model.HistoryDatasetAssociation):
id = value.dataset.id
source = hda_source_dict
if isinstance(value, HistoryDatasetAssociation):
id: int = value.dataset.id
source: Union[
Dict[Union[int, str], HistoryDatasetAssociation],
Dict[Union[int, str], HistoryDatasetCollectionAssociation],
] = hda_source_dict
elif isinstance(value, HistoryDatasetCollectionAssociation):
id = value.collection.id
source = hdca_source_dict
@@ -3264,7 +3284,7 @@ class ExpressionTool(Tool):
output.set_meta()
output.set_metadata_success_state()
except Exception:
output.state = model.HistoryDatasetAssociation.states.FAILED_METADATA
output.state = HistoryDatasetAssociation.states.FAILED_METADATA
log.exception("Exception occured while setting metdata")
def parse_environment_variables(self, tool_source):
@@ -3386,7 +3406,7 @@ class SetMetadataTool(Tool):
tool_action: "SetMetadataToolAction"
def regenerate_imported_metadata_if_needed(
self, hda: model.HistoryDatasetAssociation, history: model.History, user: model.User, session_id: int
self, hda: HistoryDatasetAssociation, history: History, user: model.User, session_id: int
):
if hda.has_metadata_files:
job, *_ = self.tool_action.execute_via_app(
@@ -3520,7 +3540,7 @@ class DataManagerTool(OutputParameterJSONTool):
pass
elif data_manager_mode == "bundle":
for bundle_path, dataset in data_manager.write_bundle(out_data).items():
hda = cast(model.HistoryDatasetAssociation, dataset)
hda = cast(HistoryDatasetAssociation, dataset)
hda.dataset.object_store.update_from_file(
hda.dataset,
extra_dir=hda.dataset.extra_files_path_name,
@@ -3536,7 +3556,7 @@ class DataManagerTool(OutputParameterJSONTool):
def get_default_history_by_trans(self, trans, create=False):
def _create_data_manager_history(user):
history = model.History(name="Data Manager History (automatically created)", user=user)
history = History(name="Data Manager History (automatically created)", user=user)
data_manager_association = model.DataManagerHistoryAssociation(user=user, history=history)
trans.sa_session.add_all((history, data_manager_association))
trans.sa_session.commit()
lib/galaxy/tools/evaluation.py
+1 -1
View File
@@ -154,7 +154,7 @@ class ToolEvaluator:
job = self.job
incoming = {p.name: p.value for p in job.parameters}
incoming = self.tool.params_from_strings(incoming, self.app)
incoming = self.tool.params_from_strings(incoming)
if "__use_cached_job__" in incoming:
self.use_cached_job = bool(incoming["__use_cached_job__"])
lib/galaxy/webapps/base/controller.py
+1 -1
View File
@@ -634,7 +634,7 @@ class UsesVisualizationMixin(UsesLibraryMixinItems):
# -- Get tool definition and add input values from job. --
tool_dict = tool.to_dict(trans, io_details=True)
tool_param_values = {p.name: p.value for p in job.parameters}
tool_param_values = tool.params_from_strings(tool_param_values, trans.app, ignore_errors=True)
tool_param_values = tool.params_from_strings(tool_param_values, ignore_errors=True)
# Only get values for simple inputs for now.
inputs_dict = [i for i in tool_dict["inputs"] if i["type"] not in ["data", "hidden_data", "conditional"]]
lib/galaxy/webapps/galaxy/api/jobs.py
+2 -2
View File
@@ -472,7 +472,7 @@ class FastAPIJobs:
"""
hda_ldda_str = hda_ldda or "hda"
job = self.service.get_job(trans, job_id=job_id, hda_ldda=hda_ldda_str)
return summarize_job_parameters(trans, job)
return JobDisplayParametersSummary(**summarize_job_parameters(trans, job))
@router.get(
"/api/datasets/{dataset_id}/parameters_display",
@@ -492,7 +492,7 @@ class FastAPIJobs:
this endpoint will change frequently.
"""
job = self.service.get_job(trans, dataset_id=dataset_id, hda_ldda=hda_ldda)
return summarize_job_parameters(trans, job)
return JobDisplayParametersSummary(**summarize_job_parameters(trans, job))
@router.get(
"/api/jobs/{job_id}/metrics",
lib/galaxy/work/context.py
+11 -8
View File
@@ -5,16 +5,19 @@ from typing import (
List,
Optional,
Tuple,
TYPE_CHECKING,
)
from typing_extensions import Literal
from galaxy.managers.context import ProvidesHistoryContext
from galaxy.model import (
GalaxySession,
History,
Role,
)
if TYPE_CHECKING:
from galaxy.model import (
GalaxySession,
History,
Role,
)
class WorkRequestContext(ProvidesHistoryContext):
@@ -34,10 +37,10 @@ class WorkRequestContext(ProvidesHistoryContext):
self,
app,
user=None,
history=None,
history: Optional["History"] = None,
workflow_building_mode=False,
url_builder=None,
galaxy_session: Optional[GalaxySession] = None,
galaxy_session: Optional["GalaxySession"] = None,
):
self._app = app
self.__user = user
@@ -171,7 +174,7 @@ class SessionRequestContext(WorkRequestContext):
def proxy_work_context_for_history(
trans: ProvidesHistoryContext, history: Optional[History] = None, workflow_building_mode=False
trans: ProvidesHistoryContext, history: Optional["History"] = None, workflow_building_mode=False
) -> WorkRequestContext:
"""Create a WorkContext for supplied context with potentially different history.
lib/galaxy/workflow/extract.py
+2 -2
View File
@@ -391,8 +391,8 @@ class WorkflowSummary:
def step_inputs(trans, job):
tool = trans.app.toolbox.get_tool(job.tool_id, tool_version=job.tool_version)
param_values = job.get_param_values(
trans.app, ignore_errors=True
param_values = tool.get_param_values(
job, ignore_errors=True
) # If a tool was updated and e.g. had a text value changed to an integer, we don't want a traceback here
associations = __cleanup_param_values(tool.inputs, param_values)
tool_inputs = tool.params_to_strings(param_values, trans.app)
test/unit/app/jobs/dynamic_tool_destination/mockGalaxy.py
+3 -3
View File
@@ -9,9 +9,6 @@ class Job:
self.param_values = {}
self.parameters = []
def get_param_values(self, app, ignore_errors=False):
return self.param_values
def set_arg_value(self, key, value):
self.param_values[key] = value
@@ -61,6 +58,9 @@ class Tool:
def add_tool_dependency(self, dependency):
self.installed_tool_dependencies.append(dependency)
def get_param_values(self, job: Job, ignore_errors=False):
return job.param_values
class ToolDependency:
def __init__(self, name, dir_name):
test/unit/app/jobs/dynamic_tool_destination/test_dynamic_tool_destination.py
+29 -29
View File
@@ -1,5 +1,9 @@
import logging
import os
from typing import (
cast,
TYPE_CHECKING,
)
from testfixtures.logcapture import log_capture
@@ -12,60 +16,56 @@ from . import (
ymltests as yt,
)
theApp = mg.App("cluster_default", "test_spec")
if TYPE_CHECKING:
from galaxy.model import Job
from galaxy.structured_app import MinimalManagerApp
from galaxy.tools import Tool
theApp = cast("MinimalManagerApp", mg.App("cluster_default", "test_spec"))
script_dir = os.path.dirname(__file__)
# ======================Jobs====================================
zeroJob = mg.Job()
zeroJob = cast("Job", mg.Job())
emptyJob = mg.Job()
emptyJob = cast("Job", mg.Job())
emptyJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test.empty"), "txt", 14)))
failJob = mg.Job()
failJob = cast("Job", mg.Job())
failJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test1.full"), "txt", 15)))
msfileJob = mg.Job()
msfileJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/not_here.full"), "txt", 15)))
notfileinpJob = mg.Job()
msfileJob.add_input_dataset(mg.InputDataset("input1", mg.NotAFile()))
runJob = mg.Job()
runJob = cast("Job", mg.Job())
runJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test3.full"), "txt", 15)))
argJob = mg.Job()
mockJob = mg.Job()
mockJob.set_arg_value("careful", True)
argJob = cast("Job", mockJob)
argJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test3.full"), "txt", 15)))
argJob.set_arg_value("careful", True)
argNotFoundJob = mg.Job()
mockJob = mg.Job()
mockJob.set_arg_value("careful", False)
argNotFoundJob = cast("Job", mockJob)
argNotFoundJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test3.full"), "txt", 15)))
argNotFoundJob.set_arg_value("careful", False)
dbJob = mg.Job()
dbJob = cast("Job", mg.Job())
dbJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test.fasta"), "fasta", 10)))
dbcountJob = mg.Job()
dbcountJob = cast("Job", mg.Job())
dbcountJob.add_input_dataset(mg.InputDataset("input1", mg.Dataset((script_dir + "/data/test.fasta"), "fasta", None)))
# ======================Tools===================================
vanillaTool = mg.Tool("test")
vanillaTool = cast("Tool", mg.Tool("test"))
unTool = mg.Tool("unregistered")
unTool = cast("Tool", mg.Tool("unregistered"))
overlapTool = mg.Tool("test_overlap")
defaultTool = cast("Tool", mg.Tool("test_tooldefault"))
defaultTool = mg.Tool("test_tooldefault")
dbTool = cast("Tool", mg.Tool("test_db"))
dbTool = mg.Tool("test_db")
dbinfTool = mg.Tool("test_db_high")
argTool = cast("Tool", mg.Tool("test_arguments"))
argTool = mg.Tool("test_arguments")
noVBTool = cast("Tool", mg.Tool("test_no_verbose"))
noVBTool = mg.Tool("test_no_verbose")
usersTool = mg.Tool("test_users")
numinputsTool = mg.Tool("test_num_input_datasets")
usersTool = cast("Tool", mg.Tool("test_users"))
# =======================Configuration files================================
path = script_dir + "/data/tool_destination.yml"
test/unit/app/jobs/test_mapper.py
+20 -13
View File
@@ -1,11 +1,12 @@
import uuid
from typing import cast
from typing import (
cast,
TYPE_CHECKING,
)
from galaxy.jobs import (
HasResourceParameters,
JobConfiguration,
JobDestination,
JobWrapper,
)
from galaxy.jobs.mapper import (
ERROR_MESSAGE_NO_RULE_FUNCTION,
@@ -18,6 +19,13 @@ from . import (
test_rules_override,
)
if TYPE_CHECKING:
from galaxy.jobs import (
JobConfiguration,
JobWrapper,
)
from galaxy.tools import Tool
WORKFLOW_UUID = uuid.uuid1().hex
TOOL_JOB_DESTINATION = JobDestination()
DYNAMICALLY_GENERATED_DESTINATION = JobDestination()
@@ -137,8 +145,8 @@ def __assert_mapper_errors_with_message(mapper, message):
def __mapper(tool_job_destination=TOOL_JOB_DESTINATION):
job_wrapper = cast(JobWrapper, MockJobWrapper(tool_job_destination))
job_config = cast(JobConfiguration, MockJobConfig())
job_wrapper = cast("JobWrapper", MockJobWrapper(tool_job_destination))
job_config = cast("JobConfiguration", MockJobConfig())
mapper = JobRunnerMapper(job_wrapper, lambda url: JobDestination(), job_config)
mapper.rules_module = test_rules
@@ -164,7 +172,7 @@ class MockJobConfig:
class MockJobWrapper(HasResourceParameters):
def __init__(self, tool_job_destination):
self.tool = MockTool(tool_job_destination)
self.tool = cast("Tool", MockTool(tool_job_destination))
self.job_id = 12345
self.app = object()
@@ -177,16 +185,10 @@ class MockJobWrapper(HasResourceParameters):
"__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,
raw_params=raw_params,
)
@@ -203,3 +205,8 @@ class MockTool:
def is_mock_tool(self):
return True
def get_param_values(self, job, ignore_errors: bool = False):
params = job.raw_params.copy()
params["__job_resource"] = {"__job_resource__select": "True", "memory": "8gb"}
return params
test/unit/app/tools/test_evaluation.py
+2 -2
View File
@@ -319,8 +319,8 @@ class MockTool:
elem = XML('<param name="thresh" type="integer" value="5" />')
return IntegerToolParameter(cast("Tool", self), elem)
def params_from_strings(self, params, app, ignore_errors=False):
return params_from_strings(self.inputs, params, app, ignore_errors)
def params_from_strings(self, params, ignore_errors=False):
return params_from_strings(self.inputs, params, self.app, ignore_errors)
@property
def config_file(self):