4 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/e3d9d40cd05a/ Changeset: e3d9d40cd05a User: jmchilton Date: 2014-05-06 20:52:49 Summary: More refactoring away from mixins and toward 'data populators' in API tests. Affected #: 3 files diff -r 6d6449af7d623b9b4a5bb0118750ad49f725b49a -r e3d9d40cd05a78d0a2202e68daab0c5c0f3b345f test/api/helpers.py --- a/test/api/helpers.py +++ b/test/api/helpers.py @@ -84,8 +84,8 @@ class WorkflowPopulator( object ): # Impulse is to make this a Mixin, but probably better as an object. - def __init__( self, api_test_case ): - self.api_test_case = api_test_case + def __init__( self, galaxy_interactor ): + self.galaxy_interactor = galaxy_interactor def load_workflow( self, name, content=workflow_str, add_pja=False ): workflow = json.loads( content ) @@ -111,7 +111,7 @@ workflow=json.dumps( workflow ), **create_kwds ) - upload_response = self.api_test_case._post( "workflows/upload", data=data ) + upload_response = self.galaxy_interactor.post( "workflows/upload", data=data ) uploaded_workflow_id = upload_response.json()[ "id" ] return uploaded_workflow_id diff -r 6d6449af7d623b9b4a5bb0118750ad49f725b49a -r e3d9d40cd05a78d0a2202e68daab0c5c0f3b345f test/api/test_search.py --- a/test/api/test_search.py +++ b/test/api/test_search.py @@ -7,7 +7,7 @@ class SearchApiTestCase( api.ApiTestCase ): def test_search_workflows( self ): - workflow_populator = WorkflowPopulator( self ) + workflow_populator = WorkflowPopulator( self.galaxy_interactor ) workflow_id = workflow_populator.simple_workflow( "test_for_search" ) search_response = self.__search( "select * from workflow" ) assert self.__has_result_with_name( search_response, "test_for_search (imported from API)" ), search_response.json() diff -r 6d6449af7d623b9b4a5bb0118750ad49f725b49a -r e3d9d40cd05a78d0a2202e68daab0c5c0f3b345f test/api/test_workflows.py --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -1,8 +1,8 @@ from base import api from json import dumps import time -from .helpers import TestsDatasets from .helpers import WorkflowPopulator +from .helpers import DatasetPopulator from base.interactor import delete_request # requests like delete @@ -12,11 +12,12 @@ # - Allow post to workflows/<workflow_id>/run in addition to posting to # /workflows with id in payload. # - Much more testing obviously, always more testing. -class WorkflowsApiTestCase( api.ApiTestCase, TestsDatasets ): +class WorkflowsApiTestCase( api.ApiTestCase ): def setUp( self ): super( WorkflowsApiTestCase, self ).setUp() - self.workflow_populator = WorkflowPopulator( self ) + self.workflow_populator = WorkflowPopulator( self.galaxy_interactor ) + self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) def test_delete( self ): workflow_id = self.workflow_populator.simple_workflow( "test_delete" ) @@ -65,7 +66,7 @@ # something like that. run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) def test_extract_from_history( self ): workflow = self.workflow_populator.load_workflow( name="test_for_extract" ) @@ -77,7 +78,7 @@ run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) data = dict( history_id=history_id, tool_id="cat1" ) jobs_response = self._get( "jobs", data=data ) self._assert_status_code_is( jobs_response, 200 ) @@ -105,7 +106,7 @@ workflow_request[ "parameters" ] = dumps( dict( random_lines1=dict( num_lines=5 ) ) ) run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) # Would be 8 and 6 without modification self.__assert_lines_hid_line_count_is( history_id, 2, 5 ) self.__assert_lines_hid_line_count_is( history_id, 3, 5 ) @@ -120,7 +121,7 @@ workflow_request[ "parameters" ] = params run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) # Would be 8 and 6 without modification self.__assert_lines_hid_line_count_is( history_id, 2, 8 ) self.__assert_lines_hid_line_count_is( history_id, 3, 5 ) @@ -166,7 +167,7 @@ workflow_request, history_id = self._setup_workflow_run( workflow ) run_workflow_response = self._post( "workflows", data=workflow_request ) self._assert_status_code_is( run_workflow_response, 200 ) - self._wait_for_history( history_id, assert_ok=True ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) time.sleep(.1) # Give another little bit of time for rename (needed?) contents = self._get( "histories/%s/contents" % history_id ).json() # loading workflow with add_pja=True causes workflow output to be @@ -183,9 +184,9 @@ step_1 = key if label == "WorkflowInput2": step_2 = key - history_id = self._new_history() - hda1 = self._new_dataset( history_id, content="1 2 3" ) - hda2 = self._new_dataset( history_id, content="4 5 6" ) + history_id = self.dataset_populator.new_history() + hda1 = self.dataset_populator.new_dataset( history_id, content="1 2 3" ) + hda2 = self.dataset_populator.new_dataset( history_id, content="4 5 6" ) workflow_request = dict( history="hist_id=%s" % history_id, workflow_id=uploaded_workflow_id, @@ -201,9 +202,9 @@ uploaded_workflow_id = self.workflow_populator.create_workflow( workflow ) workflow_inputs = self._workflow_inputs( uploaded_workflow_id ) key = workflow_inputs.keys()[ 0 ] - history_id = self._new_history() + history_id = self.dataset_populator.new_history() ten_lines = "\n".join( map( str, range( 10 ) ) ) - hda1 = self._new_dataset( history_id, content=ten_lines ) + hda1 = self.dataset_populator.new_dataset( history_id, content=ten_lines ) workflow_request = dict( history="hist_id=%s" % history_id, workflow_id=uploaded_workflow_id, https://bitbucket.org/galaxy/galaxy-central/commits/d82e6db63d3b/ Changeset: d82e6db63d3b User: jmchilton Date: 2014-05-06 20:52:49 Summary: Fix location of test_dataset_collections.py. Didn't rebase properly. Affected #: 2 files diff -r e3d9d40cd05a78d0a2202e68daab0c5c0f3b345f -r d82e6db63d3b581dadedcb4cc38e150445b9637c test/api/test_dataset_collections.py --- /dev/null +++ b/test/api/test_dataset_collections.py @@ -0,0 +1,206 @@ +from base import api +import json +from .helpers import DatasetPopulator + + +# TODO: Move into helpers with rest of populators +class DatasetCollectionPopulator( object ): + + def __init__( self, galaxy_interactor ): + self.galaxy_interactor = galaxy_interactor + self.dataset_populator = DatasetPopulator( galaxy_interactor ) + + def create_list_from_pairs( self, history_id, pairs ): + element_identifiers = [] + for i, pair in enumerate( pairs ): + element_identifiers.append( dict( + name="test%d" % i, + src="hdca", + id=pair + ) ) + + payload = dict( + instance_type="history", + history_id=history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list:paired", + ) + return self.__create( payload ) + + def create_pair_in_history( self, history_id, **kwds ): + payload = self.create_pair_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_in_history( self, history_id, **kwds ): + payload = self.create_list_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) + + def create_pair_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) + + def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): + contents = None + if "contents" in kwds: + contents = kwds[ "contents" ] + del kwds[ "contents" ] + + if "element_identifiers" not in kwds: + kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) + + payload = dict( + history_id=history_id, + collection_type=collection_type, + **kwds + ) + return payload + + def pair_identifiers( self, history_id, contents=None ): + hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) + + element_identifiers = [ + dict( name="left", src="hda", id=hda1[ "id" ] ), + dict( name="right", src="hda", id=hda2[ "id" ] ), + ] + return element_identifiers + + def list_identifiers( self, history_id, contents=None ): + hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) + element_identifiers = [ + dict( name="data1", src="hda", id=hda1[ "id" ] ), + dict( name="data2", src="hda", id=hda2[ "id" ] ), + dict( name="data3", src="hda", id=hda3[ "id" ] ), + ] + return element_identifiers + + def __create( self, payload ): + create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) + return create_response + + def __datasets( self, history_id, count, contents=None ): + datasets = [] + for i in xrange( count ): + new_kwds = {} + if contents: + new_kwds[ "content" ] = contents[ i ] + datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) + return datasets + + +class DatasetCollectionApiTestCase( api.ApiTestCase ): + + def setUp( self ): + super( DatasetCollectionApiTestCase, self ).setUp() + self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) + self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor ) + self.history_id = self.dataset_populator.new_history() + + def test_create_pair_from_history( self ): + payload = self.dataset_collection_populator.create_pair_payload( + self.history_id, + instance_type="history", + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_datasets = dataset_collection[ "elements" ] + assert len( returned_datasets ) == 2, dataset_collection + + def test_create_list_from_history( self ): + element_identifiers = self.dataset_collection_populator.list_identifiers( self.history_id ) + + payload = dict( + instance_type="history", + history_id=self.history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list", + ) + + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_datasets = dataset_collection[ "elements" ] + assert len( returned_datasets ) == 3, dataset_collection + + def test_create_list_of_existing_pairs( self ): + pair_payload = self.dataset_collection_populator.create_pair_payload( + self.history_id, + instance_type="history", + ) + pair_create_response = self._post( "dataset_collections", pair_payload ) + dataset_collection = self._check_create_response( pair_create_response ) + hdca_id = dataset_collection[ "id" ] + + element_identifiers = [ + dict( name="test1", src="hdca", id=hdca_id ) + ] + + payload = dict( + instance_type="history", + history_id=self.history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list", + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + returned_collections = dataset_collection[ "elements" ] + assert len( returned_collections ) == 1, dataset_collection + + def test_create_list_of_new_pairs( self ): + pair_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) + element_identifiers = [ dict( + src="new_collection", + name="test_pair", + collection_type="paired", + element_identifiers=pair_identifiers, + ) ] + payload = dict( + collection_type="list:paired", + instance_type="history", + history_id=self.history_id, + name="nested_collecion", + element_identifiers=json.dumps( element_identifiers ), + ) + create_response = self._post( "dataset_collections", payload ) + dataset_collection = self._check_create_response( create_response ) + assert dataset_collection[ "collection_type" ] == "list:paired" + returned_collections = dataset_collection[ "elements" ] + assert len( returned_collections ) == 1, dataset_collection + pair_1_element = returned_collections[ 0 ] + self._assert_has_keys( pair_1_element, "element_index" ) + pair_1_object = pair_1_element[ "object" ] + self._assert_has_keys( pair_1_object, "collection_type", "elements" ) + self.assertEquals( pair_1_object[ "collection_type" ], "paired" ) + pair_elements = pair_1_object[ "elements" ] + assert len( pair_elements ) == 2 + pair_1_element_1 = pair_elements[ 0 ] + assert pair_1_element_1[ "element_index" ] == 0 + + def test_hda_security( self ): + element_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) + + with self._different_user( ): + history_id = self.dataset_populator.new_history() + payload = dict( + instance_type="history", + history_id=history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="paired", + ) + + create_response = self._post( "dataset_collections", payload ) + self._assert_status_code_is( create_response, 403 ) + + def _check_create_response( self, create_response ): + self._assert_status_code_is( create_response, 200 ) + dataset_collection = create_response.json() + self._assert_has_keys( dataset_collection, "elements", "url", "name", "collection_type" ) + return dataset_collection diff -r e3d9d40cd05a78d0a2202e68daab0c5c0f3b345f -r d82e6db63d3b581dadedcb4cc38e150445b9637c test/functional/api/test_dataset_collections.py --- a/test/functional/api/test_dataset_collections.py +++ /dev/null @@ -1,206 +0,0 @@ -from base import api -import json -from .helpers import DatasetPopulator - - -# TODO: Move into helpers with rest of populators -class DatasetCollectionPopulator( object ): - - def __init__( self, galaxy_interactor ): - self.galaxy_interactor = galaxy_interactor - self.dataset_populator = DatasetPopulator( galaxy_interactor ) - - def create_list_from_pairs( self, history_id, pairs ): - element_identifiers = [] - for i, pair in enumerate( pairs ): - element_identifiers.append( dict( - name="test%d" % i, - src="hdca", - id=pair - ) ) - - payload = dict( - instance_type="history", - history_id=history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list:paired", - ) - return self.__create( payload ) - - def create_pair_in_history( self, history_id, **kwds ): - payload = self.create_pair_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_in_history( self, history_id, **kwds ): - payload = self.create_list_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) - - def create_pair_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) - - def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): - contents = None - if "contents" in kwds: - contents = kwds[ "contents" ] - del kwds[ "contents" ] - - if "element_identifiers" not in kwds: - kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) - - payload = dict( - history_id=history_id, - collection_type=collection_type, - **kwds - ) - return payload - - def pair_identifiers( self, history_id, contents=None ): - hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) - - element_identifiers = [ - dict( name="left", src="hda", id=hda1[ "id" ] ), - dict( name="right", src="hda", id=hda2[ "id" ] ), - ] - return element_identifiers - - def list_identifiers( self, history_id, contents=None ): - hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) - element_identifiers = [ - dict( name="data1", src="hda", id=hda1[ "id" ] ), - dict( name="data2", src="hda", id=hda2[ "id" ] ), - dict( name="data3", src="hda", id=hda3[ "id" ] ), - ] - return element_identifiers - - def __create( self, payload ): - create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) - return create_response - - def __datasets( self, history_id, count, contents=None ): - datasets = [] - for i in xrange( count ): - new_kwds = {} - if contents: - new_kwds[ "content" ] = contents[ i ] - datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) - return datasets - - -class DatasetCollectionApiTestCase( api.ApiTestCase ): - - def setUp( self ): - super( DatasetCollectionApiTestCase, self ).setUp() - self.dataset_populator = DatasetPopulator( self.galaxy_interactor ) - self.dataset_collection_populator = DatasetCollectionPopulator( self.galaxy_interactor ) - self.history_id = self.dataset_populator.new_history() - - def test_create_pair_from_history( self ): - payload = self.dataset_collection_populator.create_pair_payload( - self.history_id, - instance_type="history", - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_datasets = dataset_collection[ "elements" ] - assert len( returned_datasets ) == 2, dataset_collection - - def test_create_list_from_history( self ): - element_identifiers = self.dataset_collection_populator.list_identifiers( self.history_id ) - - payload = dict( - instance_type="history", - history_id=self.history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list", - ) - - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_datasets = dataset_collection[ "elements" ] - assert len( returned_datasets ) == 3, dataset_collection - - def test_create_list_of_existing_pairs( self ): - pair_payload = self.dataset_collection_populator.create_pair_payload( - self.history_id, - instance_type="history", - ) - pair_create_response = self._post( "dataset_collections", pair_payload ) - dataset_collection = self._check_create_response( pair_create_response ) - hdca_id = dataset_collection[ "id" ] - - element_identifiers = [ - dict( name="test1", src="hdca", id=hdca_id ) - ] - - payload = dict( - instance_type="history", - history_id=self.history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list", - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - returned_collections = dataset_collection[ "elements" ] - assert len( returned_collections ) == 1, dataset_collection - - def test_create_list_of_new_pairs( self ): - pair_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) - element_identifiers = [ dict( - src="new_collection", - name="test_pair", - collection_type="paired", - element_identifiers=pair_identifiers, - ) ] - payload = dict( - collection_type="list:paired", - instance_type="history", - history_id=self.history_id, - name="nested_collecion", - element_identifiers=json.dumps( element_identifiers ), - ) - create_response = self._post( "dataset_collections", payload ) - dataset_collection = self._check_create_response( create_response ) - assert dataset_collection[ "collection_type" ] == "list:paired" - returned_collections = dataset_collection[ "elements" ] - assert len( returned_collections ) == 1, dataset_collection - pair_1_element = returned_collections[ 0 ] - self._assert_has_keys( pair_1_element, "element_index" ) - pair_1_object = pair_1_element[ "object" ] - self._assert_has_keys( pair_1_object, "collection_type", "elements" ) - self.assertEquals( pair_1_object[ "collection_type" ], "paired" ) - pair_elements = pair_1_object[ "elements" ] - assert len( pair_elements ) == 2 - pair_1_element_1 = pair_elements[ 0 ] - assert pair_1_element_1[ "element_index" ] == 0 - - def test_hda_security( self ): - element_identifiers = self.dataset_collection_populator.pair_identifiers( self.history_id ) - - with self._different_user( ): - history_id = self.dataset_populator.new_history() - payload = dict( - instance_type="history", - history_id=history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="paired", - ) - - create_response = self._post( "dataset_collections", payload ) - self._assert_status_code_is( create_response, 403 ) - - def _check_create_response( self, create_response ): - self._assert_status_code_is( create_response, 200 ) - dataset_collection = create_response.json() - self._assert_has_keys( dataset_collection, "elements", "url", "name", "collection_type" ) - return dataset_collection https://bitbucket.org/galaxy/galaxy-central/commits/4cf29a388e80/ Changeset: 4cf29a388e80 User: jmchilton Date: 2014-05-06 20:52:49 Summary: Move dataset colleciton test populator. Affected #: 4 files diff -r d82e6db63d3b581dadedcb4cc38e150445b9637c -r 4cf29a388e802d073d28deb7c365c6ec5f555122 test/api/helpers.py --- a/test/api/helpers.py +++ b/test/api/helpers.py @@ -189,6 +189,99 @@ return show().json() +class DatasetCollectionPopulator( object ): + + def __init__( self, galaxy_interactor ): + self.galaxy_interactor = galaxy_interactor + self.dataset_populator = DatasetPopulator( galaxy_interactor ) + + def create_list_from_pairs( self, history_id, pairs ): + element_identifiers = [] + for i, pair in enumerate( pairs ): + element_identifiers.append( dict( + name="test%d" % i, + src="hdca", + id=pair + ) ) + + payload = dict( + instance_type="history", + history_id=history_id, + element_identifiers=json.dumps(element_identifiers), + collection_type="list:paired", + ) + return self.__create( payload ) + + def create_pair_in_history( self, history_id, **kwds ): + payload = self.create_pair_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_in_history( self, history_id, **kwds ): + payload = self.create_list_payload( + history_id, + instance_type="history", + **kwds + ) + return self.__create( payload ) + + def create_list_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) + + def create_pair_payload( self, history_id, **kwds ): + return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) + + def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): + contents = None + if "contents" in kwds: + contents = kwds[ "contents" ] + del kwds[ "contents" ] + + if "element_identifiers" not in kwds: + kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) + + payload = dict( + history_id=history_id, + collection_type=collection_type, + **kwds + ) + return payload + + def pair_identifiers( self, history_id, contents=None ): + hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) + + element_identifiers = [ + dict( name="left", src="hda", id=hda1[ "id" ] ), + dict( name="right", src="hda", id=hda2[ "id" ] ), + ] + return element_identifiers + + def list_identifiers( self, history_id, contents=None ): + hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) + element_identifiers = [ + dict( name="data1", src="hda", id=hda1[ "id" ] ), + dict( name="data2", src="hda", id=hda2[ "id" ] ), + dict( name="data3", src="hda", id=hda3[ "id" ] ), + ] + return element_identifiers + + def __create( self, payload ): + create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) + return create_response + + def __datasets( self, history_id, count, contents=None ): + datasets = [] + for i in xrange( count ): + new_kwds = {} + if contents: + new_kwds[ "content" ] = contents[ i ] + datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) + return datasets + + def wait_on_state( state_func, assert_ok=False, timeout=5 ): delta = .1 iteration = 0 diff -r d82e6db63d3b581dadedcb4cc38e150445b9637c -r 4cf29a388e802d073d28deb7c365c6ec5f555122 test/api/test_dataset_collections.py --- a/test/api/test_dataset_collections.py +++ b/test/api/test_dataset_collections.py @@ -1,100 +1,7 @@ from base import api import json from .helpers import DatasetPopulator - - -# TODO: Move into helpers with rest of populators -class DatasetCollectionPopulator( object ): - - def __init__( self, galaxy_interactor ): - self.galaxy_interactor = galaxy_interactor - self.dataset_populator = DatasetPopulator( galaxy_interactor ) - - def create_list_from_pairs( self, history_id, pairs ): - element_identifiers = [] - for i, pair in enumerate( pairs ): - element_identifiers.append( dict( - name="test%d" % i, - src="hdca", - id=pair - ) ) - - payload = dict( - instance_type="history", - history_id=history_id, - element_identifiers=json.dumps(element_identifiers), - collection_type="list:paired", - ) - return self.__create( payload ) - - def create_pair_in_history( self, history_id, **kwds ): - payload = self.create_pair_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_in_history( self, history_id, **kwds ): - payload = self.create_list_payload( - history_id, - instance_type="history", - **kwds - ) - return self.__create( payload ) - - def create_list_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.list_identifiers, collection_type="list", **kwds ) - - def create_pair_payload( self, history_id, **kwds ): - return self.__create_payload( history_id, identifiers_func=self.pair_identifiers, collection_type="paired", **kwds ) - - def __create_payload( self, history_id, identifiers_func, collection_type, **kwds ): - contents = None - if "contents" in kwds: - contents = kwds[ "contents" ] - del kwds[ "contents" ] - - if "element_identifiers" not in kwds: - kwds[ "element_identifiers" ] = json.dumps( identifiers_func( history_id, contents=contents ) ) - - payload = dict( - history_id=history_id, - collection_type=collection_type, - **kwds - ) - return payload - - def pair_identifiers( self, history_id, contents=None ): - hda1, hda2 = self.__datasets( history_id, count=2, contents=contents ) - - element_identifiers = [ - dict( name="left", src="hda", id=hda1[ "id" ] ), - dict( name="right", src="hda", id=hda2[ "id" ] ), - ] - return element_identifiers - - def list_identifiers( self, history_id, contents=None ): - hda1, hda2, hda3 = self.__datasets( history_id, count=3, contents=contents ) - element_identifiers = [ - dict( name="data1", src="hda", id=hda1[ "id" ] ), - dict( name="data2", src="hda", id=hda2[ "id" ] ), - dict( name="data3", src="hda", id=hda3[ "id" ] ), - ] - return element_identifiers - - def __create( self, payload ): - create_response = self.galaxy_interactor.post( "dataset_collections", data=payload ) - return create_response - - def __datasets( self, history_id, count, contents=None ): - datasets = [] - for i in xrange( count ): - new_kwds = {} - if contents: - new_kwds[ "content" ] = contents[ i ] - datasets.append( self.dataset_populator.new_dataset( history_id, **new_kwds ) ) - return datasets +from .helpers import DatasetCollectionPopulator class DatasetCollectionApiTestCase( api.ApiTestCase ): diff -r d82e6db63d3b581dadedcb4cc38e150445b9637c -r 4cf29a388e802d073d28deb7c365c6ec5f555122 test/api/test_history_contents.py --- a/test/api/test_history_contents.py +++ b/test/api/test_history_contents.py @@ -3,7 +3,7 @@ from .helpers import TestsDatasets from .helpers import LibraryPopulator -from .test_dataset_collections import DatasetCollectionPopulator +from .helpers import DatasetCollectionPopulator from base.interactor import ( put_request, delete_request, diff -r d82e6db63d3b581dadedcb4cc38e150445b9637c -r 4cf29a388e802d073d28deb7c365c6ec5f555122 test/api/test_tools.py --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -2,7 +2,7 @@ from base import api from operator import itemgetter from .helpers import DatasetPopulator -from .test_dataset_collections import DatasetCollectionPopulator +from .helpers import DatasetCollectionPopulator class ToolsTestCase( api.ApiTestCase ): https://bitbucket.org/galaxy/galaxy-central/commits/c3370be77427/ Changeset: c3370be77427 User: jmchilton Date: 2014-05-06 20:52:49 Summary: Fix bug in test/api/test_tools.py. Affected #: 1 file diff -r 4cf29a388e802d073d28deb7c365c6ec5f555122 -r c3370be77427ba8cdd575306d804b27facb9f49f test/api/test_tools.py --- a/test/api/test_tools.py +++ b/test/api/test_tools.py @@ -167,7 +167,6 @@ self.assertEquals( len( jobs ), 2 ) self.assertEquals( len( outputs ), 2 ) self.assertEquals( len( implicit_collections ), 1 ) - self.dataset_populator.wait_for_history( history_id, assert_ok=True ) output1 = outputs[ 0 ] output2 = outputs[ 1 ] @@ -285,7 +284,8 @@ return self._run_outputs( self._run( tool_id, history_id, inputs ) ) def _run_outputs( self, create_response ): - self._assert_status_code_is( create_response, 200, assert_ok=True )[ 'outputs' ] + self._assert_status_code_is( create_response, 200 ) + return create_response.json()[ 'outputs' ] def _run_cat1( self, history_id, inputs, assert_ok=False ): return self._run( 'cat1', history_id, inputs, assert_ok=assert_ok ) Repository URL: https://bitbucket.org/galaxy/galaxy-central/ -- This is a commit notification from bitbucket.org. You are receiving this because you have the service enabled, addressing the recipient of this email.