From 0c3f7989bb913842e28a8fbb267d1280420caa92 Mon Sep 17 00:00:00 2001 From: anuprulez Date: Mon, 3 Apr 2017 16:41:16 +0200 Subject: [PATCH] Merging wtih dev --- .ci/flake8_lint_include_list.txt | 3 +- config/galaxy.ini.sample | 4 + lib/galaxy/config.py | 1 + lib/galaxy/datatypes/interval.py | 15 ++- lib/galaxy/datatypes/util/gff_util.py | 27 +++++- lib/galaxy/model/__init__.py | 5 + lib/galaxy/tools/deps/resolvers/conda.py | 21 +++++ lib/galaxy/tools/deps/views.py | 48 ++++++++++ lib/galaxy/web/base/controllers/admin.py | 27 ++++-- lib/galaxy/workflow/modules.py | 12 ++- lib/galaxy/workflow/run.py | 40 +++++--- .../galaxy/admin/manage_dependencies.mako | 92 ++++++++++++++++--- test/api/test_workflows.py | 22 +---- test/api/test_workflows_from_yaml.py | 1 + test/base/data/test_workflow_pause.ga | 2 +- test/base/populators.py | 22 ++++- ...st_maximum_worklfow_invocation_duration.py | 48 ++++++++++ 17 files changed, 325 insertions(+), 65 deletions(-) create mode 100644 test/integration/test_maximum_worklfow_invocation_duration.py diff --git a/.ci/flake8_lint_include_list.txt b/.ci/flake8_lint_include_list.txt index bdb890cf6c6..30e3db4d20e 100644 --- a/.ci/flake8_lint_include_list.txt +++ b/.ci/flake8_lint_include_list.txt @@ -66,8 +66,7 @@ lib/galaxy/datatypes/sequence.py lib/galaxy/datatypes/tabular.py lib/galaxy/datatypes/text.py lib/galaxy/datatypes/tracks.py -lib/galaxy/datatypes/util/generic_util.py -lib/galaxy/datatypes/util/__init__.py +lib/galaxy/datatypes/util/ lib/galaxy/eggs/ lib/galaxy/exceptions/__init__.py lib/galaxy/external_services/__init__.py diff --git a/config/galaxy.ini.sample b/config/galaxy.ini.sample index c201ffc8818..3159d7443fa 100644 --- a/config/galaxy.ini.sample +++ b/config/galaxy.ini.sample @@ -1089,6 +1089,10 @@ use_interactive = True # collections. #force_beta_workflow_scheduled_for_collections=False +# This is the maximum amount of time a workflow invocation may stay in an active +# scheduling state in seconds. Set to -1 to disable this maximum and allow any workflow +# invocation to schedule indefinitely. The default corresponds to 1 month. +#maximum_workflow_invocation_duration = 2678400 # Force serial scheduling of workflows within the context of a particular history #history_local_serial_workflow_scheduling=False diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 0c0fb50bdc8..2a4cfa227d0 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -327,6 +327,7 @@ class Configuration( object ): self.force_beta_workflow_scheduled_for_collections = string_as_bool( kwargs.get( 'force_beta_workflow_scheduled_for_collections', 'False' ) ) self.history_local_serial_workflow_scheduling = string_as_bool( kwargs.get( 'history_local_serial_workflow_scheduling', 'False' ) ) + self.maximum_workflow_invocation_duration = int( kwargs.get( "maximum_workflow_invocation_duration", 2678400 ) ) # Per-user Job concurrency limitations self.cache_user_job_count = string_as_bool( kwargs.get( 'cache_user_job_count', False ) ) diff --git a/lib/galaxy/datatypes/interval.py b/lib/galaxy/datatypes/interval.py index b12a6a8fe58..1992479e8bb 100644 --- a/lib/galaxy/datatypes/interval.py +++ b/lib/galaxy/datatypes/interval.py @@ -16,7 +16,7 @@ from galaxy.datatypes import metadata from galaxy.datatypes.metadata import MetadataElement from galaxy.datatypes.sniff import get_headers from galaxy.datatypes.tabular import Tabular -from galaxy.datatypes.util.gff_util import parse_gff_attributes +from galaxy.datatypes.util.gff_util import parse_gff3_attributes, parse_gff_attributes from galaxy.web import url_for from . import ( @@ -649,6 +649,7 @@ class Gff( Tabular, _RemoteCallMixin ): edam_data = "data_1255" edam_format = "format_2305" file_ext = "gff" + valid_gff_frame = ['.', '0', '1', '2'] column_names = [ 'Seqname', 'Source', 'Feature', 'Start', 'End', 'Score', 'Strand', 'Frame', 'Group' ] data_sources = { "data": "interval_index", "index": "bigwig", "feature_search": "fli" } track_type = Interval.track_type @@ -868,6 +869,8 @@ class Gff( Tabular, _RemoteCallMixin ): return False if hdr[6] not in data.valid_strand: return False + if hdr[7] not in self.valid_gff_frame: + return False return True except: return False @@ -902,7 +905,7 @@ class Gff3( Gff ): edam_format = "format_1975" file_ext = "gff3" valid_gff3_strand = ['+', '-', '.', '?'] - valid_gff3_phase = ['.', '0', '1', '2'] + valid_gff3_phase = Gff.valid_gff_frame column_names = [ 'Seqid', 'Source', 'Type', 'Start', 'End', 'Score', 'Strand', 'Phase', 'Attributes' ] track_type = Interval.track_type @@ -945,7 +948,7 @@ class Gff3( Gff ): def sniff( self, filename ): """ - Determines whether the file is in gff version 3 format + Determines whether the file is in GFF version 3 format GFF 3 format: @@ -969,6 +972,9 @@ class Gff3( Gff ): >>> fname = get_test_fname( 'test.gff' ) >>> Gff3().sniff( fname ) False + >>> fname = get_test_fname( 'test.gtf' ) + >>> Gff3().sniff( fname ) + False >>> fname = get_test_fname('gff_version_3.gff') >>> Gff3().sniff( fname ) True @@ -1005,6 +1011,7 @@ class Gff3( Gff ): return False if hdr[7] not in self.valid_gff3_phase: return False + parse_gff3_attributes(hdr[8]) return True except: return False @@ -1069,6 +1076,8 @@ class Gtf( Gff ): return False if hdr[6] not in data.valid_strand: return False + if hdr[7] not in self.valid_gff_frame: + return False # Check attributes for gene_id, transcript_id attributes = parse_gff_attributes( hdr[8] ) diff --git a/lib/galaxy/datatypes/util/gff_util.py b/lib/galaxy/datatypes/util/gff_util.py index 2a445e03765..502f405cc29 100644 --- a/lib/galaxy/datatypes/util/gff_util.py +++ b/lib/galaxy/datatypes/util/gff_util.py @@ -3,8 +3,8 @@ Provides utilities for working with GFF files. """ import copy -from bx.intervals.io import GenomicInterval, MissingFieldError, NiceReaderWrapper, ParseError, GenomicIntervalReader -from bx.tabular.io import Header, Comment +from bx.intervals.io import GenomicInterval, GenomicIntervalReader, MissingFieldError, NiceReaderWrapper, ParseError +from bx.tabular.io import Comment, Header from galaxy.util.odict import odict @@ -347,6 +347,29 @@ def parse_gff_attributes( attr_str ): return attributes +def parse_gff3_attributes( attr_str ): + """ + Parses a GFF3 attribute string and returns a dictionary of name-value + pairs. The general format for a GFF3 attributes string is + + name1=value1;name2=value2 + """ + attributes_list = attr_str.split(";") + attributes = {} + for tag_value_pair in attributes_list: + pair = tag_value_pair.strip().split("=") + if len(pair) == 1: + raise Exception("Attribute '%s' does not contain a '='" % tag_value_pair) + if pair == '': + continue + tag = pair[0].strip() + if tag == '': + raise Exception("Empty tag in attribute '%s'" % tag_value_pair) + value = pair[1].strip() + attributes[tag] = value + return attributes + + def gff_attributes_to_str( attrs, gff_format ): """ Convert GFF attributes to string. Supported formats are GFF3, GTF. diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 0345cce6b07..dad50d9613e 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -4090,6 +4090,11 @@ class WorkflowInvocation( object, Dictifiable ): return True return False + @property + def seconds_since_created( self ): + create_time = self.create_time or galaxy.model.orm.now.now() # In case not flushed yet + return (galaxy.model.orm.now.now() - create_time).total_seconds() + class WorkflowInvocationToSubworkflowInvocationAssociation( object, Dictifiable ): dict_collection_visible_keys = ( 'id', 'workflow_step_id', 'workflow_invocation_id', 'subworkflow_invocation_id' ) diff --git a/lib/galaxy/tools/deps/resolvers/conda.py b/lib/galaxy/tools/deps/resolvers/conda.py index a74b219541e..01a709210c5 100644 --- a/lib/galaxy/tools/deps/resolvers/conda.py +++ b/lib/galaxy/tools/deps/resolvers/conda.py @@ -131,6 +131,10 @@ class CondaDependencyResolver(DependencyResolver, MultipleDependencyResolver, Li if not all_resolved: return None environments = set([os.path.basename(dependency.environment_path) for dependency in all_resolved]) + return self.uninstall_environments(environments) + + def uninstall_environments(self, environments): + environments = [env if not env.startswith(self.conda_context.envs_path) else os.path.basename(env) for env in environments] return_codes = [self.conda_context.exec_remove([env]) for env in environments] final_return_code = 0 for env, return_code in zip(environments, return_codes): @@ -286,6 +290,23 @@ class CondaDependencyResolver(DependencyResolver, MultipleDependencyResolver, Li def _expand_requirement(self, requirement): return self._expand_specs(self._expand_mappings(requirement)) + def unused_dependency_paths(self, toolbox_requirements_status): + """ + Identify all local environments that are not needed to build requirements_status. + + We try to resolve the requirements, and we note every environment_path that has been taken. + """ + used_paths = set() + for dependencies in toolbox_requirements_status.values(): + for dependency in dependencies: + if dependency.get('dependency_type') == 'conda': + path = os.path.basename(dependency['environment_path']) + used_paths.add(path) + dir_contents = set(os.listdir(self.conda_context.envs_path) if os.path.exists(self.conda_context.envs_path) else []) + unused_paths = dir_contents.difference(used_paths) # New set with paths in dir_contents but not in used_paths + unused_paths = [os.path.join(self.conda_context.envs_path, p) for p in unused_paths] + return unused_paths + def list_dependencies(self): for install_target in installed_conda_targets(self.conda_context): name = install_target.package diff --git a/lib/galaxy/tools/deps/views.py b/lib/galaxy/tools/deps/views.py index 0b5c23e3d54..683f77c42f5 100644 --- a/lib/galaxy/tools/deps/views.py +++ b/lib/galaxy/tools/deps/views.py @@ -71,6 +71,35 @@ class DependencyResolversView(object): return return_code return None + @property + def unused_dependency_paths(self): + """List dependencies that are not currently installed.""" + unused_dependencies = [] + toolbox_requirements_status = self.toolbox_requirements_status + for resolver in self._dependency_resolvers: + if hasattr(resolver, 'unused_dependency_paths'): + unused_dependencies.extend(resolver.unused_dependency_paths(toolbox_requirements_status)) + return set(unused_dependencies) + + def remove_unused_dependency_paths(self, envs): + """ + Remove dependencies that are not currently used. + + Returns a list of all environments that have been successfully removed. + """ + envs_to_remove = set(envs) + toolbox_requirements_status = self.toolbox_requirements_status + removed_environments = set() + for resolver in self._dependency_resolvers: + if hasattr(resolver, 'unused_dependency_paths') and hasattr(resolver, 'uninstall_environments'): + unused_dependencies = resolver.unused_dependency_paths(toolbox_requirements_status) + can_remove = envs_to_remove & set(unused_dependencies) + exit_code = resolver.uninstall_environments(can_remove) + if exit_code == 0: + removed_environments = removed_environments.union(can_remove) + envs_to_remove = envs_to_remove.difference(can_remove) + return list(removed_environments) + def install_dependencies(self, requirements): return self._dependency_manager._requirements_to_dependencies_dict(requirements, **{'install': True}) @@ -161,6 +190,25 @@ class DependencyResolversView(object): """ return [index for index, resolver in enumerate(self._dependency_resolvers) if resolver.can_uninstall_dependencies and not resolver.disabled] + @property + def tool_ids_by_requirements(self): + """Dictionary with requirements as keys, and tool_ids as values.""" + tool_ids_by_requirements = {} + if not self._app.toolbox.tools_by_id: + return {} + for tid, tool in self._app.toolbox.tools_by_id.items(): + if tool.tool_requirements not in tool_ids_by_requirements: + tool_ids_by_requirements[tool.tool_requirements] = [tid] + else: + tool_ids_by_requirements[tool.tool_requirements].append(tid) + return tool_ids_by_requirements + + @property + def toolbox_requirements_status(self): + return {r: self.get_requirements_status(tool_requirements_d={tids[0]: r}, + installed_tool_dependencies=self._app.toolbox.tools_by_id[tids[0]].installed_tool_dependencies) + for r, tids in self.tool_ids_by_requirements.items()} + def get_requirements_status(self, tool_requirements_d, installed_tool_dependencies=None): dependencies = self.show_dependencies(tool_requirements_d, installed_tool_dependencies) # dependencies is a dict keyed on tool_ids, value is a ToolRequirements object for that tool. diff --git a/lib/galaxy/web/base/controllers/admin.py b/lib/galaxy/web/base/controllers/admin.py index 1490fdf667e..2b8f11e0386 100644 --- a/lib/galaxy/web/base/controllers/admin.py +++ b/lib/galaxy/web/base/controllers/admin.py @@ -1122,9 +1122,18 @@ class Admin( object ): @web.expose @web.require_admin - def manage_tool_dependencies( self, trans, install_dependencies=False, uninstall_dependencies=False, selected_tool_ids=None, viewkey='View tool-centric dependencies'): + def manage_tool_dependencies( self, + trans, + install_dependencies=False, + uninstall_dependencies=False, + remove_unused_dependencies=False, + selected_tool_ids=None, + selected_environments_to_uninstall=None, + viewkey='View tool-centric dependencies'): if not selected_tool_ids: selected_tool_ids = [] + if not selected_environments_to_uninstall: + selected_environments_to_uninstall = [] tools_by_id = trans.app.toolbox.tools_by_id view = six.next(six.itervalues(trans.app.toolbox.tools_by_id))._view if selected_tool_ids: @@ -1136,17 +1145,15 @@ class Admin( object ): [view.install_dependencies(r) for r in requirements] elif uninstall_dependencies: [view.uninstall_dependencies(index=None, requirements=r) for r in requirements] - tool_ids_by_requirements = {} - for tid, tool in trans.app.toolbox.tools_by_id.items(): - if tool.tool_requirements not in tool_ids_by_requirements: - tool_ids_by_requirements[tool.tool_requirements] = [tid] - else: - tool_ids_by_requirements[tool.tool_requirements].append(tid) - requirements_status = {r: view.get_requirements_status({tid: r}, tools_by_id[tids[0]].installed_tool_dependencies) for r, tids in tool_ids_by_requirements.items()} + if selected_environments_to_uninstall and remove_unused_dependencies: + if not isinstance(selected_environments_to_uninstall, list): + selected_environments_to_uninstall = [selected_environments_to_uninstall] + view.remove_unused_dependency_paths(selected_environments_to_uninstall) return trans.fill_template( '/webapps/galaxy/admin/manage_dependencies.mako', tools=tools_by_id, - requirements_status=requirements_status, - tool_ids_by_requirements=tool_ids_by_requirements, + requirements_status=view.toolbox_requirements_status, + tool_ids_by_requirements=view.tool_ids_by_requirements, + unused_environments=view.unused_dependency_paths, viewkey=viewkey ) @web.expose diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index 87fc5894eb5..fd6571660f7 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -508,7 +508,7 @@ class PauseModule( WorkflowModule ): return state def execute( self, trans, progress, invocation, step ): - progress.mark_step_outputs_delayed( step ) + progress.mark_step_outputs_delayed( step, why="executing pause step" ) return None def recover_mapping( self, step, step_invocations, progress ): @@ -522,7 +522,8 @@ class PauseModule( WorkflowModule ): return elif action is False: raise CancelWorkflowEvaluation() - raise DelayedWorkflowEvaluation() + delayed_why = "workflow paused at this step waiting for review" + raise DelayedWorkflowEvaluation(why=delayed_why) def do_invocation_step_action( self, step, action ): """ Update or set the workflow invocation state action - generic @@ -834,7 +835,8 @@ class ToolModule( WorkflowModule ): workflow_invocation_uuid=invocation.uuid.hex ) except ToolInputsNotReadyException: - raise DelayedWorkflowEvaluation() + delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id + raise DelayedWorkflowEvaluation(why=delayed_why) if collection_info: step_outputs = dict( execution_tracker.implicit_collections ) @@ -1020,7 +1022,9 @@ def load_module_sections( trans ): class DelayedWorkflowEvaluation(Exception): - pass + + def __init__(self, why=None): + self.why = why class CancelWorkflowEvaluation(Exception): diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 190c2d9d67e..e47694501c2 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -145,6 +145,17 @@ class WorkflowInvoker( object ): def invoke( self ): workflow_invocation = self.workflow_invocation + maximum_duration = getattr( self.trans.app.config, "maximum_workflow_invocation_duration", -1 ) + if maximum_duration > 0 and workflow_invocation.seconds_since_created > maximum_duration: + log.debug("Workflow invocation [%s] exceeded maximum number of seconds allowed for scheduling [%s], failing." % (workflow_invocation.id, maximum_duration)) + workflow_invocation.state = model.WorkflowInvocation.states.FAILED + # All jobs ran successfully, so we can save now + self.trans.sa_session.add( workflow_invocation ) + + # Not flushing in here, because web controller may create multiple + # invocations. + return self.progress.outputs + remaining_steps = self.progress.remaining_steps() delayed_steps = False for step in remaining_steps: @@ -165,9 +176,9 @@ class WorkflowInvoker( object ): # https://github.com/galaxyproject/galaxy/issues/2259 if job: workflow_invocation_step.job_id = job.id - except modules.DelayedWorkflowEvaluation: + except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True - self.progress.mark_step_outputs_delayed( step ) + self.progress.mark_step_outputs_delayed( step, why=de.why ) except Exception: log.exception( "Failed to schedule %s, problem occurred on %s.", @@ -176,8 +187,8 @@ class WorkflowInvoker( object ): ) raise - step_verb = "invoked" if not step_delayed else "delayed" - log.debug("Workflow step %s of invocation %s %s %s" % (step.id, workflow_invocation.id, step_verb, step_timer)) + if not step_delayed: + log.debug("Workflow step %s of invocation %s invoked %s" % (step.id, workflow_invocation.id, step_timer)) if delayed_steps: state = model.WorkflowInvocation.states.READY @@ -207,14 +218,16 @@ class WorkflowInvoker( object ): # No steps created yet - have to delay evaluation. if not step_invocations: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but that step has not been invoked yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) for step_invocation in step_invocations: job = step_invocation.job if job: # At least one job in incomplete. if not job.finished: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "depends on step [%s] but one or more jobs created from that step have not finished yet" % output_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) if job.state != job.states.OK: raise modules.CancelWorkflowEvaluation() @@ -292,7 +305,8 @@ class WorkflowProgress( object ): raise Exception(message) step_outputs = self.outputs[ output_step_id ] if step_outputs is STEP_OUTPUT_DELAYED: - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent step [%s] delayed, so this step must be delayed" % output_step_id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) output_name = connection.output_name try: replacement = step_outputs[ output_name ] @@ -313,7 +327,8 @@ class WorkflowProgress( object ): # TODO: consider distinguish between cancelled and failed? raise modules.CancelWorkflowEvaluation() - raise modules.DelayedWorkflowEvaluation() + delayed_why = "dependent collection [%s] not yet populated with datasets" % replacement.id + raise modules.DelayedWorkflowEvaluation(why=delayed_why) return replacement def get_replacement_workflow_output( self, workflow_output ): @@ -338,7 +353,10 @@ class WorkflowProgress( object ): def set_step_outputs(self, step, outputs): self.outputs[ step.id ] = outputs - def mark_step_outputs_delayed(self, step): + def mark_step_outputs_delayed(self, step, why=None): + if why: + message = "Marking step %s outputs of invocation %s delayed (%s)" % (step.id, self.workflow_invocation.id, why) + log.debug(message) self.outputs[ step.id ] = STEP_OUTPUT_DELAYED def _subworkflow_invocation(self, step): @@ -395,8 +413,8 @@ class WorkflowProgress( object ): def _recover_mapping( self, step, step_invocations ): try: step.module.recover_mapping( step, step_invocations, self ) - except modules.DelayedWorkflowEvaluation: - self.mark_step_outputs_delayed( step ) + except modules.DelayedWorkflowEvaluation as de: + self.mark_step_outputs_delayed( step, de.why ) __all__ = ( 'invoke', 'WorkflowRunConfig' ) diff --git a/templates/webapps/galaxy/admin/manage_dependencies.mako b/templates/webapps/galaxy/admin/manage_dependencies.mako index 86b1d906392..29ae1f3077f 100644 --- a/templates/webapps/galaxy/admin/manage_dependencies.mako +++ b/templates/webapps/galaxy/admin/manage_dependencies.mako @@ -23,10 +23,62 @@ <%def name="render_tool_centric_table( tools, requirements_status)"> + + Select + Name + ID + Requirement + Version + Resolver + Exact + + + <% ctr = 0 %> + %for tool in tools.values(): + %if tool.tool_requirements: + %if ctr % 2 == 1: + + %else: + + %endif + + + + ${ tool.name | h } + ${ tool.id | h } + ${render_tool_dependencies( requirements_status[tool.tool_requirements], ctr=ctr) } + + <% ctr += 1 %> + %endif + %endfor + + +<%def name="render_unused_dependencies(unused_environments)"> Select - Name - ID + Environment Path + + <% ctr = 0 %> + %for path in unused_environments: + %if ctr % 2 == 1: + + %else: + + %endif + + + + ${ path | h } + + <% ctr += 1 %> + %endfor + + +<%def name="render_dependencies_details(tools, requirements_status, tool_ids_by_requirements)"> + + Select + Used by + Environment Path Requirement Version Resolver @@ -34,26 +86,26 @@ <% ctr = 0 %> - %for tool in tools.values(): - %if tool.tool_requirements: + %for requirements, r_status in requirements_status.items(): + %if requirements: + <% tool_ids = tool_ids_by_requirements[requirements] %> %if ctr % 2 == 1: %else: %endif - + - ${ tool.name | h } - ${ tool.id | h } - ${render_tool_dependencies( requirements_status[tool.tool_requirements], ctr=ctr) } + ${ ", ".join([tools[tid].name for tid in tool_ids]) | h } + ${render_tool_dependencies( r_status, ctr=ctr, show_environment_path=True, ncols_extra=3) } - <% ctr += 1 %> %endif + <% ctr += 1 %> %endfor -<%def name="render_dependencies_details( tools, requirements_status, tool_ids_by_requirements)"> +<%def name="render_dependencies_details(tools, requirements_status, tool_ids_by_requirements)"> Select Used by @@ -94,22 +146,36 @@
%if viewkey == "Switch to tool-centric view": +
Tool-centric dependencies
${render_tool_centric_table(tools, requirements_status)} +%elif viewkey == "Switch to unused dependencies view": + + +
+
Unused dependency environments
+
+
+ ${render_unused_dependencies(unused_environments)} %else: +
Dependency details
- ${render_dependencies_details(tools, requirements_status, tool_ids_by_requirements)} + ${render_dependencies_details(tools, requirements_status, tool_ids_by_requirements)} %endif
- - + %if viewkey =="Switch to unused dependencies view": + + %else: + + + %endif
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index 38261d06812..67d80aa8bfa 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -18,10 +18,6 @@ from base.populators import ( from galaxy.exceptions import error_codes from galaxy.tools.verify.test_data import TestDataResolver -from base.workflows_format_2 import ( - convert_and_import_workflow, - ImporterGalaxyInterface, -) SIMPLE_NESTED_WORKFLOW_YAML = """ class: GalaxyWorkflow @@ -69,7 +65,7 @@ test_data: """ -class BaseWorkflowsApiTestCase( api.ApiTestCase, ImporterGalaxyInterface ): +class BaseWorkflowsApiTestCase( api.ApiTestCase ): # TODO: Find a new file for this class. def setUp( self ): @@ -88,20 +84,12 @@ class BaseWorkflowsApiTestCase( api.ApiTestCase, ImporterGalaxyInterface ): names = [w[ "name" ] for w in index_response.json()] return names - # Import importer interface... def import_workflow(self, workflow, **kwds): - workflow_str = dumps(workflow, indent=4) - data = { - 'workflow': workflow_str, - } - data.update(**kwds) - upload_response = self._post( "workflows", data=data ) - self._assert_status_code_is( upload_response, 200 ) - return upload_response.json() + upload_response = self.workflow_populator.import_workflow(workflow, **kwds) + return upload_response def _upload_yaml_workflow(self, has_yaml, **kwds): - workflow = convert_and_import_workflow(has_yaml, galaxy_interface=self, **kwds) - return workflow[ "id" ] + return self.workflow_populator.upload_yaml_workflow(has_yaml, **kwds) def _setup_workflow_run( self, workflow, inputs_by='step_id', history_id=None ): uploaded_workflow_id = self.workflow_populator.create_workflow( workflow ) @@ -869,8 +857,6 @@ test_data: @skip_without_tool( "cat1" ) @skip_without_tool( "collection_paired_test" ) def test_workflow_run_zip_collections( self ): - # A more advanced output collection workflow, testing regression of - # https://github.com/galaxyproject/galaxy/issues/776 history_id = self.dataset_populator.new_history() workflow_id = self._upload_yaml_workflow(""" class: GalaxyWorkflow diff --git a/test/api/test_workflows_from_yaml.py b/test/api/test_workflows_from_yaml.py index cc269146a54..d1a41027156 100644 --- a/test/api/test_workflows_from_yaml.py +++ b/test/api/test_workflows_from_yaml.py @@ -279,6 +279,7 @@ steps: def _steps_by_label(self, workflow_as_dict): by_label = {} + assert "steps" in workflow_as_dict, workflow_as_dict for step in workflow_as_dict["steps"].values(): by_label[step['label']] = step return by_label diff --git a/test/base/data/test_workflow_pause.ga b/test/base/data/test_workflow_pause.ga index 6f5b4cd4580..02b9797ddd8 100644 --- a/test/base/data/test_workflow_pause.ga +++ b/test/base/data/test_workflow_pause.ga @@ -107,7 +107,7 @@ }, "post_job_actions": {}, "tool_errors": null, - "tool_id": "cat1", + "tool_id": "cat", "tool_state": "{\"__page__\": 0, \"__rerun_remap_job_id__\": null, \"input1\": \"null\", \"queries\": \"[]\"}", "tool_version": "1.0.0", "type": "tool", diff --git a/test/base/populators.py b/test/base/populators.py index 7e059a94d2a..f83386ddadf 100644 --- a/test/base/populators.py +++ b/test/base/populators.py @@ -9,6 +9,10 @@ from pkg_resources import resource_string from six import StringIO from base import api_asserts +from base.workflows_format_2 import ( + convert_and_import_workflow, + ImporterGalaxyInterface, +) # Simple workflow that takes an input and call cat wrapper on it. workflow_str = resource_string( __name__, "data/test_workflow_1.ga" ) @@ -257,6 +261,10 @@ class BaseWorkflowPopulator( object ): upload_response = self._post( "workflows/upload", data=data ) return upload_response + def upload_yaml_workflow(self, has_yaml, **kwds): + workflow = convert_and_import_workflow(has_yaml, galaxy_interface=self, **kwds) + return workflow[ "id" ] + def wait_for_invocation( self, workflow_id, invocation_id, timeout=DEFAULT_TIMEOUT ): url = "workflows/%s/usage/%s" % ( workflow_id, invocation_id ) return wait_on_state( lambda: self._get( url ), timeout=timeout ) @@ -268,7 +276,7 @@ class BaseWorkflowPopulator( object ): self.dataset_populator.wait_for_history( history_id, assert_ok=assert_ok, timeout=timeout ) -class WorkflowPopulator( BaseWorkflowPopulator ): +class WorkflowPopulator( BaseWorkflowPopulator, ImporterGalaxyInterface ): def __init__( self, galaxy_interactor ): self.galaxy_interactor = galaxy_interactor @@ -280,6 +288,18 @@ class WorkflowPopulator( BaseWorkflowPopulator ): def _get( self, route ): return self.galaxy_interactor.get( route ) + # Required for ImporterGalaxyInterface interface - so we can recurisvely import + # nested workflows. + def import_workflow(self, workflow, **kwds): + workflow_str = json.dumps(workflow, indent=4) + data = { + 'workflow': workflow_str, + } + data.update(**kwds) + upload_response = self._post( "workflows", data=data ) + assert upload_response.status_code == 200, upload_response + return upload_response.json() + class LibraryPopulator( object ): diff --git a/test/integration/test_maximum_worklfow_invocation_duration.py b/test/integration/test_maximum_worklfow_invocation_duration.py new file mode 100644 index 00000000000..b3544504352 --- /dev/null +++ b/test/integration/test_maximum_worklfow_invocation_duration.py @@ -0,0 +1,48 @@ +"""Integration tests for maximum workflow invocation duration configuration option.""" + +import time + +from json import dumps + +from base import integration_util +from base.populators import ( + DatasetPopulator, + WorkflowPopulator, +) + + +class MaximumWorkflowInvocationDurationTestCase(integration_util.IntegrationTestCase): + """Start a Pulsar job.""" + + framework_tool_and_types = True + + def setUp( self ): + super( MaximumWorkflowInvocationDurationTestCase, self ).setUp() + self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) + self.workflow_populator = WorkflowPopulator( self.galaxy_interactor ) + + @classmethod + def handle_galaxy_config_kwds(cls, config): + config["maximum_workflow_invocation_duration"] = 20 + + def do_test(self): + workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause") + workflow_id = self.workflow_populator.create_workflow(workflow) + history_id = self.dataset_populator.new_history() + hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3") + index_map = { + '0': dict(src="hda", id=hda1["id"]) + } + request = {} + request["history"] = "hist_id=%s" % history_id + request[ "inputs" ] = dumps(index_map) + request[ "inputs_by" ] = 'step_index' + url = "workflows/%s/invocations" % (workflow_id) + invocation_response = self._post(url, data=request) + invocation_url = url + "/" + invocation_response.json()["id"] + time.sleep(5) + state = self._get(invocation_url).json()["state"] + assert state != "failed", state + time.sleep(35) + state = self._get(invocation_url).json()["state"] + assert state == "failed", state