Mercurial > repos > guerler > springsuite

comparison planemo/lib/python3.7/site-packages/bioblend/_tests/TestGalaxyHistories.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 """
2 """
3 import os
4 import shutil
5 import tarfile
6 import tempfile
7
8 from . import GalaxyTestBase, test_util
9
10
11 class TestGalaxyHistories(GalaxyTestBase.GalaxyTestBase):
12
13 def setUp(self):
14 super().setUp()
15 self.default_history_name = "buildbot - automated test"
16 self.history = self.gi.histories.create_history(name=self.default_history_name)
17
18 def test_create_history(self):
19 history_name = "another buildbot - automated test"
20 new_history = self.gi.histories.create_history(name=history_name)
21 self.assertIsNotNone(new_history['id'])
22 self.assertEqual(new_history['name'], history_name)
23 self.assertIsNotNone(new_history['url'])
24
25 def test_update_history(self):
26 new_name = 'buildbot - automated test renamed'
27 new_annotation = 'Annotation for %s' % new_name
28 new_tags = ['tag1', 'tag2']
29 updated_hist = self.gi.histories.update_history(self.history['id'], name=new_name, annotation=new_annotation, tags=new_tags)
30 if 'id' not in updated_hist:
31 updated_hist = self.gi.histories.show_history(self.history['id'])
32 self.assertEqual(self.history['id'], updated_hist['id'])
33 self.assertEqual(updated_hist['name'], new_name)
34 self.assertEqual(updated_hist['annotation'], new_annotation)
35 self.assertEqual(updated_hist['tags'], new_tags)
36 updated_hist = self.gi.histories.update_history(self.history['id'], published=True)
37 if 'id' not in updated_hist:
38 updated_hist = self.gi.histories.show_history(self.history['id'])
39 self.assertEqual(self.history['id'], updated_hist['id'])
40 self.assertTrue(updated_hist['published'])
41
42 def test_get_histories(self):
43 # Make sure there's at least one value - the one we created
44 all_histories = self.gi.histories.get_histories()
45 self.assertGreater(len(all_histories), 0)
46
47 # Check whether name is correct, when searched by id
48 new_history = self.gi.histories.get_histories(history_id=self.history['id'])[0]
49 self.assertEqual(new_history['name'], self.default_history_name)
50
51 # Check whether id is present, when searched by name
52 new_history = self.gi.histories.get_histories(name=self.default_history_name)
53 self.assertTrue(any(d['id'] == self.history['id'] for d in new_history))
54
55 # TODO: check whether deleted history is returned correctly
56 # At the moment, get_histories() returns only not-deleted histories
57 # and get_histories(deleted=True) returns only deleted histories,
58 # so they are not comparable.
59 # In the future, according to https://trello.com/c/MoilsmVv/1673-api-incoherent-and-buggy-indexing-of-deleted-entities ,
60 # get_histories() will return both not-deleted and deleted histories
61 # and we can uncomment the following test.
62 # deleted_history = self.gi.histories.get_histories(deleted=True)
63 # self.assertGreaterEqual(len(all_histories), len(deleted_history))
64
65 def test_show_history(self):
66 history_data = self.gi.histories.show_history(self.history['id'])
67 self.assertEqual(self.history['id'], history_data['id'])
68 self.assertEqual(self.history['name'], history_data['name'])
69 self.assertEqual('new', history_data['state'])
70
71 def test_show_history_with_contents(self):
72 history_id = self.history["id"]
73 contents = self.gi.histories.show_history(history_id, contents=True)
74 # Empty history has no datasets, content length should be 0
75 self.assertEqual(len(contents), 0)
76 self._test_dataset(history_id)
77 contents = self.gi.histories.show_history(history_id, contents=True)
78 # history has 1 dataset, content length should be 1
79 self.assertEqual(len(contents), 1)
80 contents = self.gi.histories.show_history(history_id, contents=True, types=['dataset'])
81 # filtering for dataset, content length should still be 1
82 self.assertEqual(len(contents), 1)
83 contents = self.gi.histories.show_history(history_id, contents=True, types=['dataset_collection'])
84 # filtering for dataset collection but there's no collection in the history
85 self.assertEqual(len(contents), 0)
86 contents = self.gi.histories.show_history(history_id, contents=True, types=['dataset', 'dataset_collection'])
87 self.assertEqual(len(contents), 1)
88
89 def test_create_history_tag(self):
90 new_tag = 'tag1'
91 self.gi.histories.create_history_tag(self.history['id'], new_tag)
92 updated_hist = self.gi.histories.show_history(self.history['id'])
93 self.assertEqual(self.history['id'], updated_hist['id'])
94 self.assertIn(new_tag, updated_hist['tags'])
95
96 def test_show_dataset(self):
97 history_id = self.history["id"]
98 dataset1_id = self._test_dataset(history_id)
99 dataset = self.gi.histories.show_dataset(history_id, dataset1_id)
100 for key in ["name", "hid", "id", "deleted", "history_id", "visible"]:
101 self.assertIn(key, dataset)
102 self.assertEqual(dataset["history_id"], history_id)
103 self.assertEqual(dataset["hid"], 1)
104 self.assertEqual(dataset["id"], dataset1_id)
105 self.assertEqual(dataset["deleted"], False)
106 self.assertEqual(dataset["visible"], True)
107
108 def test_show_dataset_provenance(self):
109 history_id = self.history["id"]
110 dataset1_id = self._test_dataset(history_id)
111 prov = self.gi.histories.show_dataset_provenance(history_id, dataset1_id)
112 for key in ["id", "job_id", "parameters", "stderr", "stdout", "tool_id"]:
113 self.assertIn(key, prov)
114
115 def test_delete_dataset(self):
116 history_id = self.history["id"]
117 dataset1_id = self._test_dataset(history_id)
118 self.gi.histories.delete_dataset(history_id, dataset1_id)
119 dataset = self.gi.histories.show_dataset(history_id, dataset1_id)
120 self.assertTrue(dataset["deleted"])
121 self.assertFalse(dataset['purged'])
122
123 def test_purge_dataset(self):
124 history_id = self.history["id"]
125 dataset1_id = self._test_dataset(history_id)
126 self.gi.histories.delete_dataset(history_id, dataset1_id, purge=True)
127 dataset = self.gi.histories.show_dataset(history_id, dataset1_id)
128 self.assertTrue(dataset["deleted"])
129 self.assertTrue(dataset['purged'])
130
131 def test_update_dataset(self):
132 history_id = self.history["id"]
133 dataset1_id = self._test_dataset(history_id)
134 updated_dataset = self.gi.histories.update_dataset(history_id, dataset1_id, visible=False)
135 if 'id' not in updated_dataset:
136 updated_dataset = self.gi.histories.show_dataset(history_id, dataset1_id)
137 self.assertFalse(updated_dataset["visible"])
138
139 def test_upload_dataset_from_library(self):
140 pass
141
142 # download_dataset() is already tested in TestGalaxyDatasets
143
144 def test_delete_history(self):
145 result = self.gi.histories.delete_history(self.history['id'])
146 self.assertTrue(result['deleted'])
147
148 all_histories = self.gi.histories.get_histories()
149 self.assertTrue(not any(d['id'] == self.history['id'] for d in all_histories))
150
151 def test_undelete_history(self):
152 self.gi.histories.delete_history(self.history['id'])
153 self.gi.histories.undelete_history(self.history['id'])
154 all_histories = self.gi.histories.get_histories()
155 self.assertTrue(any(d['id'] == self.history['id'] for d in all_histories))
156
157 def test_get_status(self):
158 state = self.gi.histories.get_status(self.history['id'])
159 self.assertEqual('new', state['state'])
160
161 def test_get_most_recently_used_history(self):
162 most_recently_used_history = self.gi.histories.get_most_recently_used_history()
163 # if the user has been created via the API, it does not have
164 # a session, therefore no history
165 if most_recently_used_history is not None:
166 self.assertIsNotNone(most_recently_used_history['id'])
167 self.assertIsNotNone(most_recently_used_history['name'])
168 self.assertIsNotNone(most_recently_used_history['state'])
169
170 def test_download_history(self):
171 jeha_id = self.gi.histories.export_history(
172 self.history['id'], wait=True, maxwait=60)
173 self.assertTrue(jeha_id)
174 tempdir = tempfile.mkdtemp(prefix='bioblend_test_')
175 temp_fn = os.path.join(tempdir, 'export.tar.gz')
176 try:
177 with open(temp_fn, 'wb') as fo:
178 self.gi.histories.download_history(self.history['id'], jeha_id,
179 fo)
180 self.assertTrue(tarfile.is_tarfile(temp_fn))
181 finally:
182 shutil.rmtree(tempdir)
183
184 def test_import_history(self):
185 path = test_util.get_abspath(os.path.join('data', 'Galaxy-History-test.tar.gz'))
186 self.gi.histories.import_history(file_path=path)
187
188 def test_copy_dataset(self):
189 history_id = self.history["id"]
190 contents = "1\t2\t3"
191 dataset1_id = self._test_dataset(history_id, contents=contents)
192 self.history_id2 = self.gi.histories.create_history('TestCopyDataset')['id']
193 copied_dataset = self.gi.histories.copy_dataset(self.history_id2, dataset1_id)
194 expected_contents = ("\n".join(contents.splitlines()) + "\n").encode()
195 self._wait_and_verify_dataset(copied_dataset['id'], expected_contents)
196 self.gi.histories.delete_history(self.history_id2, purge=True)
197
198 def tearDown(self):
199 self.gi.histories.delete_history(self.history['id'], purge=True)