#
# SPDX-License-Identifier: Apache-2.0
-from __future__ import absolute_import
-from builtins import str
-from builtins import range
import arvados
import collections
import contextlib
from apiclient import errors as apiclient_errors
from apiclient import http as apiclient_http
from arvados.api import (
+ ThreadSafeAPIClient,
api_client,
normalize_api_kwargs,
api_kwargs_from_config,
- OrderedJsonModel,
_googleapiclient_log_lock,
)
from .arvados_testutil import fake_httplib2_response, mock_api_responses, queue_with
+
+import googleapiclient
import httplib2.error
if not mimetypes.inited:
self.assertIsNot(*clients)
def test_empty_list(self):
- answer = arvados.api('v1').humans().list(
- filters=[['uuid', '=', None]]).execute()
+ answer = arvados.api('v1').collections().list(
+ filters=[['uuid', '=', 'abcdef']]).execute()
self.assertEqual(answer['items_available'], len(answer['items']))
def test_nonempty_list(self):
def test_timestamp_inequality_filter(self):
api = arvados.api('v1')
- new_item = api.specimens().create(body={}).execute()
+ new_item = api.collections().create(body={}).execute()
for operator, should_include in [
['<', False], ['>', False],
['<=', True], ['>=', True], ['=', True]]:
- response = api.specimens().list(filters=[
+ response = api.collections().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()
def test_exceptions_include_errors(self):
mock_responses = {
- 'arvados.humans.get': self.api_error_response(
+ 'arvados.collections.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()
+ api.collections().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):
mock_responses = {
- 'arvados.humans.delete': (
+ 'arvados.collections.delete': (
fake_httplib2_response(500, **self.ERROR_HEADERS),
b"")
}
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()
+ api.collections().delete(uuid='xyz-xyz-abcdef').execute()
self.assertIn("500", str(err_ctx.exception))
def test_request_too_large(self):
self.assertEqual(response.status, code)
self.assertEqual(response.get('status'), str(code))
- 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
- )).encode(),
- ),
- }
- 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(list(result.keys())))
-
def test_api_is_threadsafe(self):
api_kwargs = {
'host': os.environ['ARVADOS_API_HOST'],
self.assertEqual(c.close.call_count, expect)
+class ThreadSafeAPIClientTestCase(run_test_server.TestCaseWithServers):
+ MAIN_SERVER = {}
+
+ def test_constructor(self):
+ env_mapping = {
+ key: value
+ for key, value in os.environ.items()
+ if key.startswith('ARVADOS_API_')
+ }
+ extra_params = {
+ 'timeout': 299,
+ }
+ base_params = {
+ key[12:].lower(): value
+ for key, value in env_mapping.items()
+ }
+ try:
+ base_params['insecure'] = base_params.pop('host_insecure')
+ except KeyError:
+ pass
+ expected_keep_params = {}
+ for config, params, subtest in [
+ (None, {}, "default arguments"),
+ (None, extra_params, "extra params"),
+ (env_mapping, {}, "explicit config"),
+ (env_mapping, extra_params, "explicit config and params"),
+ ({}, base_params, "params only"),
+ ]:
+ with self.subTest(f"test constructor with {subtest}"):
+ expected_timeout = params.get('timeout', 300)
+ expected_params = dict(params)
+ keep_params = dict(expected_keep_params)
+ client = ThreadSafeAPIClient(config, keep_params, params, 'v1')
+ self.assertTrue(hasattr(client, 'localapi'), "client missing localapi method")
+ self.assertEqual(client.api_token, os.environ['ARVADOS_API_TOKEN'])
+ self.assertEqual(client._http.timeout, expected_timeout)
+ self.assertEqual(params, expected_params,
+ "api_params was modified in-place")
+ self.assertEqual(keep_params, expected_keep_params,
+ "keep_params was modified in-place")
+
+ def test_constructor_no_args(self):
+ client = ThreadSafeAPIClient()
+ self.assertTrue(hasattr(client, 'localapi'), "client missing localapi method")
+ self.assertEqual(client.api_token, os.environ['ARVADOS_API_TOKEN'])
+ self.assertTrue(client.insecure)
+
+ def test_constructor_bad_version(self):
+ with self.assertRaises(googleapiclient.errors.UnknownApiNameOrVersion):
+ ThreadSafeAPIClient(version='BadTestVersion')
+
+ def test_pre_v3_0_name(self):
+ from arvados.safeapi import ThreadSafeApiCache
+ self.assertIs(ThreadSafeApiCache, ThreadSafeAPIClient)
+
+
if __name__ == '__main__':
unittest.main()
projects / arvados.git / blobdiff