Refactoring out methods usable for downstream work on scheduling.

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -383,7 +383,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
         results = self.workflow_manager.build_invocations_query( trans, decoded_stored_workflow_invocation_id )
         out = []
         for r in results:
-            out.append( self.encode_all_ids( trans, r.to_dict(), True) )
+            out.append( self.__encode_invocation( trans, r ) )
         return out
 
     @expose_api
@@ -403,7 +403,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
         decoded_workflow_invocation_id = self.__decode_id( trans, usage_id )
         workflow_invocation = self.workflow_manager.get_invocation( trans, decoded_workflow_invocation_id )
         if workflow_invocation:
-            return self.encode_all_ids( trans, workflow_invocation.to_dict('element'), True)
+            return self.__encode_invocation( trans, workflow_invocation )
         return None
 
     def __get_stored_accessible_workflow( self, trans, workflow_id ):
@@ -435,6 +435,13 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis
             raise exceptions.ObjectNotFound( "No such workflow found." )
         return stored_workflow
 
+    def __encode_invocation( self, trans, invocation, view="element" ):
+        return self.encode_all_ids(
+            trans,
+            invocation.to_dict( view ),
+            True
+        )
+
     def __decode_id( self, trans, workflow_id, model_type="workflow" ):
         try:
             return trans.security.decode_id( workflow_id )

test/api/test_workflows.py

@@ -515,9 +515,7 @@ class WorkflowsApiTestCase( api.ApiTestCase ):
     @skip_without_tool( "cat1" )
     def test_invocation_usage( self ):
         workflow_id, usage = self._run_workflow_once_get_invocation( "test_usage")
-        usage_details_response = self._get( "workflows/%s/usage/%s" % ( workflow_id, usage[ "id" ] ) )
-        self._assert_status_code_is( usage_details_response, 200 )
-        usage_details = usage_details_response.json()
+        usage_details = self._invocation_details( workflow_id, usage[ "id" ] )
         # Assert some high-level things about the structure of data returned.
         self._assert_has_keys( usage_details, "inputs", "steps" )
         for step in usage_details[ "steps" ]:
@@ -538,6 +536,12 @@ class WorkflowsApiTestCase( api.ApiTestCase ):
         # renamed to 'the_new_name'.
         assert "the_new_name" in map( lambda hda: hda[ "name" ], contents )
 
+    def _invocation_details( self, workflow_id, invocation_id ):
+        invocation_details_response = self._get( "workflows/%s/usage/%s" % ( workflow_id, invocation_id ) )
+        self._assert_status_code_is( invocation_details_response, 200 )
+        invocation_details = invocation_details_response.json()
+        return invocation_details
+
     def _run_workflow_once_get_invocation( self, name ):
         workflow = self.workflow_populator.load_workflow( name=name )
         workflow_request, history_id = self._setup_workflow_run( workflow )