From 4cbc33af0979b9a197b3934ab2a23bab4534d7be Mon Sep 17 00:00:00 2001 From: John Chilton Date: Wed, 10 Sep 2014 11:51:10 -0400 Subject: [PATCH] API functional test for workflow 'replacement_params' and PJA renaming. --- test/api/helpers.py | 28 +++++++++++++++++++++------- test/api/test_workflows.py | 10 ++++++++++ 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/test/api/helpers.py b/test/api/helpers.py index 9df844561d5..768a2229273 100644 --- a/test/api/helpers.py +++ b/test/api/helpers.py @@ -120,24 +120,38 @@ class DatasetPopulator( object ): return tool_response.json() def get_history_dataset_content( self, history_id, wait=True, **kwds ): + dataset_id = self.__history_dataset_id( history_id, wait=wait, **kwds ) + display_response = self.__get_contents_request( history_id, "/%s/display" % dataset_id ) + assert display_response.status_code == 200, display_response.content + return display_response.content + + def get_history_dataset_details( self, history_id, **kwds ): + dataset_id = self.__history_dataset_id( history_id, **kwds ) + details_response = self.__get_contents_request( history_id, "/%s" % dataset_id ) + assert details_response.status_code == 200 + return details_response.json() + + def __history_dataset_id( self, history_id, wait=True, **kwds ): if wait: assert_ok = kwds.get( "assert_ok", True ) self.wait_for_history( history_id, assert_ok=assert_ok ) # kwds should contain a 'dataset' object response, a 'dataset_id' or # the last dataset in the history will be fetched. - contents_url = "histories/%s/contents" % history_id if "dataset_id" in kwds: dataset_id = kwds[ "dataset_id" ] elif "dataset" in kwds: dataset_id = kwds[ "dataset" ][ "id" ] else: hid = kwds.get( "hid", -1 ) # If not hid, just grab last dataset - dataset_contents = self.galaxy_interactor.get( contents_url ).json() - dataset_id = dataset_contents[ hid ][ "id" ] + dataset_contents = self.__get_contents_request( history_id ).json() + dataset_id = dataset_contents[ hid - 1 ][ "id" ] + return dataset_id - display_response = self.galaxy_interactor.get( "%s/%s/display" % ( contents_url, dataset_id ) ) - assert display_response.status_code == 200 - return display_response.content + def __get_contents_request( self, history_id, suffix=""): + url = "histories/%s/contents" % history_id + if suffix: + url = "%s%s" % ( url, suffix ) + return self.galaxy_interactor.get( url ) class WorkflowPopulator( object ): @@ -154,7 +168,7 @@ class WorkflowPopulator( object ): tool_step[ "post_job_actions" ][ "RenameDatasetActionout_file1" ] = dict( action_type="RenameDatasetAction", output_name="out_file1", - action_arguments=dict( newname="the_new_name" ), + action_arguments=dict( newname="foo ${replaceme}" ), ) return workflow diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index ec2a0ca7f42..75a7594586c 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -467,6 +467,16 @@ class WorkflowsApiTestCase( api.ApiTestCase ): assert n == expected_len, "Expected %d steps of type %s, found %d" % ( expected_len, type, n ) return sorted( steps, key=operator.itemgetter("id") ) + @skip_without_tool( "cat1" ) + def test_run_with_pja( self ): + workflow = self.workflow_populator.load_workflow( name="test_for_pja_run", add_pja=True ) + workflow_request, history_id = self._setup_workflow_run( workflow, inputs_by='step_index' ) + workflow_request[ "replacement_params" ] = dumps( dict( replaceme="was replaced" ) ) + run_workflow_response = self._post( "workflows", data=workflow_request ) + self._assert_status_code_is( run_workflow_response, 200 ) + content = self.dataset_populator.get_history_dataset_details( history_id, wait=True, assert_ok=True ) + assert content[ "name" ] == "foo was replaced" + @skip_without_tool( "random_lines1" ) def test_run_replace_params_by_tool( self ): workflow_request, history_id = self._setup_random_x2_workflow( "test_for_replace_tool_params" )