X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/728ee9775de422840b81827c71c7b515708620a3..2c149e2e0a4d00831f111aedaaf47320100c3a00:/sdk/python/tests/test_api.py diff --git a/sdk/python/tests/test_api.py b/sdk/python/tests/test_api.py index e9cb838cb3..6d1e9798cf 100644 --- a/sdk/python/tests/test_api.py +++ b/sdk/python/tests/test_api.py @@ -1,74 +1,126 @@ #!/usr/bin/env python -import apiclient.errors import arvados +import collections import httplib2 import json import mimetypes import os -import run_test_server +import socket +import string import unittest -from apiclient.http import RequestMockBuilder +import mock +import run_test_server + +from apiclient import errors as apiclient_errors +from apiclient import http as apiclient_http +from arvados.api import OrderedJsonModel from arvados_testutil import fake_httplib2_response if not mimetypes.inited: mimetypes.init() -class ArvadosApiClientTest(unittest.TestCase): +class ArvadosApiTest(run_test_server.TestCaseWithServers): + MAIN_SERVER = {} ERROR_HEADERS = {'Content-Type': mimetypes.types_map['.json']} - @classmethod - def api_error_response(cls, code, *errors): - return (fake_httplib2_response(code, **cls.ERROR_HEADERS), + def api_error_response(self, code, *errors): + return (fake_httplib2_response(code, **self.ERROR_HEADERS), json.dumps({'errors': errors, 'error_token': '1234567890+12345678'})) - @classmethod - def setUpClass(cls): - # The apiclient library has support for mocking requests for - # testing, but it doesn't extend to the discovery document - # itself. For now, bring up an API server that will serve - # a discovery document. - # FIXME: Figure out a better way to stub this out. - run_test_server.run() - mock_responses = { - 'arvados.humans.delete': ( - fake_httplib2_response(500, **cls.ERROR_HEADERS), - ""), - 'arvados.humans.get': cls.api_error_response( - 422, "Bad UUID format", "Bad output format"), - 'arvados.humans.list': (None, json.dumps( - {'items_available': 0, 'items': []})), - } - req_builder = RequestMockBuilder(mock_responses) - cls.api = arvados.api('v1', cache=False, - host=os.environ['ARVADOS_API_HOST'], - token='discovery-doc-only-no-token-needed', - insecure=True, - requestBuilder=req_builder) - - @classmethod - def tearDownClass(cls): - run_test_server.stop() - - def test_basic_list(self): - answer = self.api.humans().list( + def test_new_api_objects_with_cache(self): + clients = [arvados.api('v1', cache=True) for index in [0, 1]] + self.assertIsNot(*clients) + + def test_empty_list(self): + answer = arvados.api('v1').humans().list( filters=[['uuid', 'is', None]]).execute() self.assertEqual(answer['items_available'], len(answer['items'])) + def test_nonempty_list(self): + answer = arvados.api('v1').collections().list().execute() + self.assertNotEqual(0, answer['items_available']) + self.assertNotEqual(0, len(answer['items'])) + + def test_timestamp_inequality_filter(self): + api = arvados.api('v1') + new_item = api.specimens().create(body={}).execute() + for operator, should_include in [ + ['<', False], ['>', False], + ['<=', True], ['>=', True], ['=', True]]: + response = api.specimens().list(filters=[ + ['created_at', operator, new_item['created_at']], + # Also filter by uuid to ensure (if it matches) it's on page 0 + ['uuid', '=', new_item['uuid']]]).execute() + uuids = [item['uuid'] for item in response['items']] + did_include = new_item['uuid'] in uuids + self.assertEqual( + did_include, should_include, + "'%s %s' filter should%s have matched '%s'" % ( + operator, new_item['created_at'], + ('' if should_include else ' not'), + new_item['created_at'])) + def test_exceptions_include_errors(self): - with self.assertRaises(apiclient.errors.HttpError) as err_ctx: - self.api.humans().get(uuid='xyz-xyz-abcdef').execute() + mock_responses = { + 'arvados.humans.get': self.api_error_response( + 422, "Bad UUID format", "Bad output format"), + } + req_builder = apiclient_http.RequestMockBuilder(mock_responses) + api = arvados.api('v1', requestBuilder=req_builder) + with self.assertRaises(apiclient_errors.HttpError) as err_ctx: + api.humans().get(uuid='xyz-xyz-abcdef').execute() err_s = str(err_ctx.exception) for msg in ["Bad UUID format", "Bad output format"]: self.assertIn(msg, err_s) def test_exceptions_without_errors_have_basic_info(self): - with self.assertRaises(apiclient.errors.HttpError) as err_ctx: - self.api.humans().delete(uuid='xyz-xyz-abcdef').execute() + mock_responses = { + 'arvados.humans.delete': ( + fake_httplib2_response(500, **self.ERROR_HEADERS), + "") + } + req_builder = apiclient_http.RequestMockBuilder(mock_responses) + api = arvados.api('v1', requestBuilder=req_builder) + with self.assertRaises(apiclient_errors.HttpError) as err_ctx: + api.humans().delete(uuid='xyz-xyz-abcdef').execute() self.assertIn("500", str(err_ctx.exception)) + def test_request_too_large(self): + api = arvados.api('v1') + maxsize = api._rootDesc.get('maxRequestSize', 0) + with self.assertRaises(apiclient_errors.MediaUploadSizeError): + text = "X" * maxsize + arvados.api('v1').collections().create(body={"manifest_text": text}).execute() + + def test_ordered_json_model(self): + mock_responses = { + 'arvados.humans.get': (None, json.dumps(collections.OrderedDict( + (c, int(c, 16)) for c in string.hexdigits))), + } + req_builder = apiclient_http.RequestMockBuilder(mock_responses) + api = arvados.api('v1', + requestBuilder=req_builder, model=OrderedJsonModel()) + result = api.humans().get(uuid='test').execute() + self.assertEqual(string.hexdigits, ''.join(result.keys())) + + def test_socket_errors_retried(self): + api = arvados.api('v1') + self.assertTrue(hasattr(api._http, 'orig_http_request'), + "test doesn't know how to intercept HTTP requests") + api._http.orig_http_request = mock.MagicMock() + mock_response = {'user': 'person'} + api._http.orig_http_request.side_effect = [ + socket.error("mock error"), + (fake_httplib2_response(200), json.dumps(mock_response)) + ] + actual_response = api.users().current().execute() + self.assertEqual(mock_response, actual_response) + self.assertGreater(api._http.orig_http_request.call_count, 1, + "client got the right response without retrying") + if __name__ == '__main__': unittest.main()