Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/bioblend/_tests/TestGalaxyInvocations.py @ 0:d30785e31577 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 6eee67778febed82ddd413c3ca40b3183a3898f1"
author guerler
date Fri, 31 Jul 2020 00:18:57 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:d30785e31577
1 import os
2 import time
3
4 from . import GalaxyTestBase, test_util
5
6
7 class TestGalaxyInvocations(GalaxyTestBase.GalaxyTestBase):
8 @test_util.skip_unless_galaxy('release_19.09')
9 def test_invocation(self):
10 path = test_util.get_abspath(os.path.join('data', 'paste_columns.ga'))
11 workflow = self.gi.workflows.import_workflow_from_local_path(path)
12 history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
13 dataset1_id = self._test_dataset(history_id)
14 dataset = {'src': 'hda', 'id': dataset1_id}
15
16 invocation = self.gi.workflows.invoke_workflow(
17 workflow['id'],
18 inputs={'Input 1': dataset, 'Input 2': dataset},
19 history_id=history_id,
20 inputs_by='name',
21 )
22
23 invocation_id = invocation["id"]
24 invocations = self.gi.invocations.get_invocations()
25 self.assertEqual(len(invocations), 1)
26 self.assertEqual(invocations[0]["id"], invocation_id)
27 self.gi.invocations.cancel_invocation(invocation_id)
28 invocation = self.gi.invocations.show_invocation(invocation_id)
29 self.assertEqual(invocation['state'], 'cancelled')
30
31 summary = self.gi.invocations.get_invocation_summary(invocation_id)
32 report = self.gi.invocations.get_invocation_report(invocation_id)
33
34 assert summary['states'] == {}
35 assert report['workflows'] == {workflow['id']: {'name': 'paste_columns'}}
36
37 @test_util.skip_unless_galaxy('release_19.09')
38 @test_util.skip_unless_tool("cat1")
39 @test_util.skip_unless_tool("cat")
40 def test_workflow_scheduling(self):
41 path = test_util.get_abspath(os.path.join('data', 'test_workflow_pause.ga'))
42 workflow = self.gi.workflows.import_workflow_from_local_path(path)
43 history_id = self.gi.histories.create_history(name="TestWorkflowState")["id"]
44 dataset1_id = self._test_dataset(history_id)
45
46 invocation = self.gi.workflows.invoke_workflow(
47 workflow["id"],
48 inputs={"0": {"src": "hda", "id": dataset1_id}},
49 )
50 invocation_id = invocation["id"]
51
52 def invocation_steps_by_order_index():
53 invocation = self.gi.invocations.show_invocation(invocation_id)
54 return dict((s["order_index"], s) for s in invocation["steps"])
55
56 for _ in range(20):
57 if 2 in invocation_steps_by_order_index():
58 break
59 time.sleep(.5)
60
61 steps = invocation_steps_by_order_index()
62 pause_step = steps[2]
63 self.assertIsNone(
64 self.gi.invocations.show_invocation_step(invocation_id, pause_step["id"])["action"])
65 self.gi.invocations.run_invocation_step_action(invocation_id, pause_step["id"], action=True)
66 self.assertTrue(self.gi.invocations.show_invocation_step(invocation_id, pause_step["id"])["action"])
67 for _ in range(20):
68 invocation = self.gi.invocations.show_invocation(invocation_id)
69 if invocation["state"] == "scheduled":
70 break
71
72 time.sleep(.5)
73
74 invocation = self.gi.invocations.show_invocation(invocation_id)
75 self.assertEqual(invocation["state"], "scheduled")