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