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'], [])