Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy_test.api.test_api_batch
import json
from requests import post
from ._framework import ApiTestCase
[docs]class ApiBatchTestCase(ApiTestCase):
def _get_api_key(self, admin=False):
return self.galaxy_interactor.api_key if not admin else self.galaxy_interactor.master_api_key
def _with_key(self, url, admin=False):
sep = '&' if '?' in url else '?'
return url + sep + 'key=' + self._get_api_key(admin=admin)
def _post_batch(self, batch):
data = json.dumps({"batch" : batch})
return post("%s/batch" % (self.galaxy_interactor.api_url), data=data)
# ---- tests
[docs] def test_simple_array(self):
batch = [
dict(url=self._with_key('/api/histories')),
dict(url=self._with_key('/api/histories'),
method='POST', body=json.dumps(dict(name='Wat'))),
dict(url=self._with_key('/api/histories')),
]
response = self._post_batch(batch)
response = response.json()
self.assertIsInstance(response, list)
self.assertEqual(len(response), 3)
[docs] def test_unallowed_route(self):
batch = [
dict(url=self._with_key('/api/workflow'))
]
response = self._post_batch(batch)
response = response.json()
self.assertIsInstance(response, list)
self.assertEqual(response[0]['status'], 403)
[docs] def test_404_route(self):
# needs to be within the allowed routes
batch = [
dict(url=self._with_key('/api/histories_bler'))
]
response = self._post_batch(batch)
response = response.json()
self.assertIsInstance(response, list)
self.assertEqual(response[0]['status'], 404)
[docs] def test_errors(self):
batch = [
dict(url=self._with_key('/api/histories/abc123')),
dict(url=self._with_key('/api/jobs'), method='POST', body=json.dumps(dict(name='Wat'))),
]
response = self._post_batch(batch)
response = response.json()
self.assertIsInstance(response, list)
self.assertEqual(response[0]['status'], 400)
self.assertEqual(response[1]['status'], 501)
[docs] def test_querystring_params(self):
post_data = dict(name='test')
create_response = self._post('histories', data=post_data).json()
history_url = '/api/histories/' + create_response['id']
history_url_with_keys = history_url + '?v=dev&keys=size,non_ready_jobs'
contents_url_with_filters = history_url + '/contents?v=dev&q=deleted&qv=True'
batch = [
dict(url=self._with_key(history_url_with_keys)),
dict(url=self._with_key(contents_url_with_filters)),
]
response = self._post_batch(batch)
response = response.json()
self.assertEqual(len(response), 2)
self.assertEqual(len(response[0]['body'].keys()), 2)
self.assertEqual(response[1]['body'], [])