sdk/python/tests/test_api.py

   1 #!/usr/bin/env python
   2
   3 import arvados
   4 import collections
   5 import httplib2
   6 import json
   7 import mimetypes
   8 import os
   9 import socket
  10 import string
  11 import unittest
  12
  13 import mock
  14 import run_test_server
  15
  16 from apiclient import errors as apiclient_errors
  17 from apiclient import http as apiclient_http
  18 from arvados.api import OrderedJsonModel
  19 from arvados_testutil import fake_httplib2_response
  20
  21 if not mimetypes.inited:
  22     mimetypes.init()
  23
  24 class ArvadosApiTest(run_test_server.TestCaseWithServers):
  25     MAIN_SERVER = {}
  26     ERROR_HEADERS = {'Content-Type': mimetypes.types_map['.json']}
  27
  28     def api_error_response(self, code, *errors):
  29         return (fake_httplib2_response(code, **self.ERROR_HEADERS),
  30                 json.dumps({'errors': errors,
  31                             'error_token': '1234567890+12345678'}))
  32
  33     def test_new_api_objects_with_cache(self):
  34         clients = [arvados.api('v1', cache=True) for index in [0, 1]]
  35         self.assertIsNot(*clients)
  36
  37     def test_empty_list(self):
  38         answer = arvados.api('v1').humans().list(
  39             filters=[['uuid', 'is', None]]).execute()
  40         self.assertEqual(answer['items_available'], len(answer['items']))
  41
  42     def test_nonempty_list(self):
  43         answer = arvados.api('v1').collections().list().execute()
  44         self.assertNotEqual(0, answer['items_available'])
  45         self.assertNotEqual(0, len(answer['items']))
  46
  47     def test_timestamp_inequality_filter(self):
  48         api = arvados.api('v1')
  49         new_item = api.specimens().create(body={}).execute()
  50         for operator, should_include in [
  51                 ['<', False], ['>', False],
  52                 ['<=', True], ['>=', True], ['=', True]]:
  53             response = api.specimens().list(filters=[
  54                 ['created_at', operator, new_item['created_at']],
  55                 # Also filter by uuid to ensure (if it matches) it's on page 0
  56                 ['uuid', '=', new_item['uuid']]]).execute()
  57             uuids = [item['uuid'] for item in response['items']]
  58             did_include = new_item['uuid'] in uuids
  59             self.assertEqual(
  60                 did_include, should_include,
  61                 "'%s %s' filter should%s have matched '%s'" % (
  62                     operator, new_item['created_at'],
  63                     ('' if should_include else ' not'),
  64                     new_item['created_at']))
  65
  66     def test_exceptions_include_errors(self):
  67         mock_responses = {
  68             'arvados.humans.get': self.api_error_response(
  69                 422, "Bad UUID format", "Bad output format"),
  70             }
  71         req_builder = apiclient_http.RequestMockBuilder(mock_responses)
  72         api = arvados.api('v1', requestBuilder=req_builder)
  73         with self.assertRaises(apiclient_errors.HttpError) as err_ctx:
  74             api.humans().get(uuid='xyz-xyz-abcdef').execute()
  75         err_s = str(err_ctx.exception)
  76         for msg in ["Bad UUID format", "Bad output format"]:
  77             self.assertIn(msg, err_s)
  78
  79     def test_exceptions_without_errors_have_basic_info(self):
  80         mock_responses = {
  81             'arvados.humans.delete': (
  82                 fake_httplib2_response(500, **self.ERROR_HEADERS),
  83                 "")
  84             }
  85         req_builder = apiclient_http.RequestMockBuilder(mock_responses)
  86         api = arvados.api('v1', requestBuilder=req_builder)
  87         with self.assertRaises(apiclient_errors.HttpError) as err_ctx:
  88             api.humans().delete(uuid='xyz-xyz-abcdef').execute()
  89         self.assertIn("500", str(err_ctx.exception))
  90
  91     def test_request_too_large(self):
  92         api = arvados.api('v1')
  93         maxsize = api._rootDesc.get('maxRequestSize', 0)
  94         with self.assertRaises(apiclient_errors.MediaUploadSizeError):
  95             text = "X" * maxsize
  96             arvados.api('v1').collections().create(body={"manifest_text": text}).execute()
  97
  98     def test_ordered_json_model(self):
  99         mock_responses = {
 100             'arvados.humans.get': (None, json.dumps(collections.OrderedDict(
 101                         (c, int(c, 16)) for c in string.hexdigits))),
 102             }
 103         req_builder = apiclient_http.RequestMockBuilder(mock_responses)
 104         api = arvados.api('v1',
 105                           requestBuilder=req_builder, model=OrderedJsonModel())
 106         result = api.humans().get(uuid='test').execute()
 107         self.assertEqual(string.hexdigits, ''.join(result.keys()))
 108
 109     def test_socket_errors_retried(self):
 110         api = arvados.api('v1')
 111         self.assertTrue(hasattr(api._http, 'orig_http_request'),
 112                         "test doesn't know how to intercept HTTP requests")
 113         api._http.orig_http_request = mock.MagicMock()
 114         mock_response = {'user': 'person'}
 115         api._http.orig_http_request.side_effect = [
 116             socket.error("mock error"),
 117             (fake_httplib2_response(200), json.dumps(mock_response))
 118             ]
 119         actual_response = api.users().current().execute()
 120         self.assertEqual(mock_response, actual_response)
 121         self.assertGreater(api._http.orig_http_request.call_count, 1,
 122                            "client got the right response without retrying")
 123
 124
 125 if __name__ == '__main__':
 126     unittest.main()