Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@test_util.skip_unless_galaxy('release_19.05')
def test_show_nonexistent_dataset(self):
with self.assertRaises(Exception):
self.gi.datasets.show_dataset('nonexistent_id')
@test_util.skip_unless_galaxy('release_19.09')
def test_invocation(self):
path = test_util.get_abspath(os.path.join('data', 'paste_columns.ga'))
workflow = self.gi.workflows.import_workflow_from_local_path(path)
history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
dataset1_id = self._test_dataset(history_id)
dataset = {'src': 'hda', 'id': dataset1_id}
invocation = self.gi.workflows.invoke_workflow(
workflow['id'],
inputs={'Input 1': dataset, 'Input 2': dataset},
history_id=history_id,
inputs_by='name',
)
invocation_id = invocation["id"]
invocations = self.gi.invocations.get_invocations()
@test_util.skip_unless_galaxy('release_15.03')
@test_util.skip_unless_tool("cat1")
@test_util.skip_unless_tool("cat")
def test_workflow_scheduling(self):
path = test_util.get_abspath(os.path.join('data', 'test_workflow_pause.ga'))
workflow = self.gi.workflows.import_workflow_from_local_path(path)
workflow_id = workflow["id"]
history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
dataset1_id = self._test_dataset(history_id)
invocations = self.gi.workflows.get_invocations(workflow_id)
self.assertEqual(len(invocations), 0)
invocation = self.gi.workflows.invoke_workflow(
workflow["id"],
inputs={"0": {"src": "hda", "id": dataset1_id}},
)
from bioblend.galaxy import dataset_collections as collections
from . import GalaxyTestBase, test_util
@test_util.skip_unless_galaxy()
class TestGalaxyDatasetCollections(GalaxyTestBase.GalaxyTestBase):
def test_create_list_in_history(self):
history_id = self.gi.histories.create_history(name="TestDSListCreate")["id"]
dataset1_id = self._test_dataset(history_id)
dataset2_id = self._test_dataset(history_id)
dataset3_id = self._test_dataset(history_id)
collection_response = self.gi.histories.create_dataset_collection(
history_id=history_id,
collection_description=collections.CollectionDescription(
name="MyDatasetList",
elements=[
collections.HistoryDatasetElement(name="sample1", id=dataset1_id),
collections.HistoryDatasetElement(name="sample2", id=dataset2_id),
collections.HistoryDatasetElement(name="sample3", id=dataset3_id),
]
@test_util.skip_unless_galaxy('release_19.09')
@test_util.skip_unless_tool("cat1")
@test_util.skip_unless_tool("cat")
def test_workflow_scheduling(self):
path = test_util.get_abspath(os.path.join('data', 'test_workflow_pause.ga'))
workflow = self.gi.workflows.import_workflow_from_local_path(path)
history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
dataset1_id = self._test_dataset(history_id)
invocation = self.gi.workflows.invoke_workflow(
workflow["id"],
inputs={"0": {"src": "hda", "id": dataset1_id}},
)
invocation_id = invocation["id"]
def invocation_steps_by_order_index():
invocation = self.gi.invocations.show_invocation(invocation_id)
@test_util.skip_unless_galaxy('release_19.05')
def test_show_nonexistent_dataset(self):
with self.assertRaises(Exception):
self.gi.datasets.show_dataset('nonexistent_id')
@test_util.skip_unless_galaxy('release_19.09')
def test_invocation_report(self):
path = test_util.get_abspath(os.path.join('data', 'paste_columns.ga'))
wf = self.gi.workflows.import_workflow_from_local_path(path)
history_id = self.gi.histories.create_history(name="test_wf_invocation")['id']
dataset1_id = self._test_dataset(history_id)
dataset = {'src': 'hda', 'id': dataset1_id}
invoke_response = self.gi.workflows.invoke_workflow(
wf['id'],
inputs={'Input 1': dataset, 'Input 2': dataset},
history_id=history_id,
inputs_by='name',
)
summary = self.gi.workflows.get_invocation_summary(wf['id'], invoke_response['id'])
report = self.gi.workflows.get_invocation_report(wf['id'], invoke_response['id'])
assert summary['states'] == {}