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 run_test_server
  10 import string
  11 import unittest
  12 from apiclient import errors as apiclient_errors
  13 from apiclient import http as apiclient_http
  14 from arvados.api import OrderedJsonModel
  15
  16 from arvados_testutil import fake_httplib2_response
  17
  18 if not mimetypes.inited:
  19     mimetypes.init()
  20
  21 class ArvadosApiClientTest(unittest.TestCase):
  22     ERROR_HEADERS = {'Content-Type': mimetypes.types_map['.json']}
  23
  24     @classmethod
  25     def api_error_response(cls, code, *errors):
  26         return (fake_httplib2_response(code, **cls.ERROR_HEADERS),
  27                 json.dumps({'errors': errors,
  28                             'error_token': '1234567890+12345678'}))
  29
  30     @classmethod
  31     def setUpClass(cls):
  32         # The apiclient library has support for mocking requests for
  33         # testing, but it doesn't extend to the discovery document
  34         # itself. For now, bring up an API server that will serve
  35         # a discovery document.
  36         # FIXME: Figure out a better way to stub this out.
  37         run_test_server.run()
  38         mock_responses = {
  39             'arvados.humans.delete': (
  40                 fake_httplib2_response(500, **cls.ERROR_HEADERS),
  41                 ""),
  42             'arvados.humans.get': cls.api_error_response(
  43                 422, "Bad UUID format", "Bad output format"),
  44             'arvados.humans.list': (None, json.dumps(
  45                     {'items_available': 0, 'items': []})),
  46             }
  47         req_builder = apiclient_http.RequestMockBuilder(mock_responses)
  48         cls.api = arvados.api('v1',
  49                               host=os.environ['ARVADOS_API_HOST'],
  50                               token='discovery-doc-only-no-token-needed',
  51                               insecure=True,
  52                               requestBuilder=req_builder)
  53
  54     def tearDown(cls):
  55         run_test_server.reset()
  56
  57     def test_new_api_objects_with_cache(self):
  58         clients = [arvados.api('v1', cache=True,
  59                                host=os.environ['ARVADOS_API_HOST'],
  60                                token='discovery-doc-only-no-token-needed',
  61                                insecure=True)
  62                    for index in [0, 1]]
  63         self.assertIsNot(*clients)
  64
  65     def test_empty_list(self):
  66         answer = arvados.api('v1').humans().list(
  67             filters=[['uuid', 'is', None]]).execute()
  68         self.assertEqual(answer['items_available'], len(answer['items']))
  69
  70     def test_nonempty_list(self):
  71         answer = arvados.api('v1').collections().list().execute()
  72         self.assertNotEqual(0, answer['items_available'])
  73         self.assertNotEqual(0, len(answer['items']))
  74
  75     def test_timestamp_inequality_filter(self):
  76         api = arvados.api('v1')
  77         new_item = api.specimens().create(body={}).execute()
  78         for operator, should_include in [
  79                 ['<', False], ['>', False],
  80                 ['<=', True], ['>=', True], ['=', True]]:
  81             response = api.specimens().list(filters=[
  82                 ['created_at', operator, new_item['created_at']],
  83                 # Also filter by uuid to ensure (if it matches) it's on page 0
  84                 ['uuid', '=', new_item['uuid']]]).execute()
  85             uuids = [item['uuid'] for item in response['items']]
  86             did_include = new_item['uuid'] in uuids
  87             self.assertEqual(
  88                 did_include, should_include,
  89                 "'%s %s' filter should%s have matched '%s'" % (
  90                     operator, new_item['created_at'],
  91                     ('' if should_include else ' not'),
  92                     new_item['created_at']))
  93
  94     def test_exceptions_include_errors(self):
  95         with self.assertRaises(apiclient_errors.HttpError) as err_ctx:
  96             self.api.humans().get(uuid='xyz-xyz-abcdef').execute()
  97         err_s = str(err_ctx.exception)
  98         for msg in ["Bad UUID format", "Bad output format"]:
  99             self.assertIn(msg, err_s)
 100
 101     def test_exceptions_without_errors_have_basic_info(self):
 102         with self.assertRaises(apiclient_errors.HttpError) as err_ctx:
 103             self.api.humans().delete(uuid='xyz-xyz-abcdef').execute()
 104         self.assertIn("500", str(err_ctx.exception))
 105
 106     def test_request_too_large(self):
 107         api = arvados.api('v1')
 108         maxsize = api._rootDesc.get('maxRequestSize', 0)
 109         with self.assertRaises(apiclient_errors.MediaUploadSizeError):
 110             text = "X" * maxsize
 111             arvados.api('v1').collections().create(body={"manifest_text": text}).execute()
 112
 113     def test_ordered_json_model(self):
 114         mock_responses = {
 115             'arvados.humans.get': (None, json.dumps(collections.OrderedDict(
 116                         (c, int(c, 16)) for c in string.hexdigits))),
 117             }
 118         req_builder = apiclient_http.RequestMockBuilder(mock_responses)
 119         api = arvados.api('v1',
 120                           host=os.environ['ARVADOS_API_HOST'],
 121                           token='discovery-doc-only-no-token-needed',
 122                           insecure=True,
 123                           requestBuilder=req_builder,
 124                           model=OrderedJsonModel())
 125         result = api.humans().get(uuid='test').execute()
 126         self.assertEqual(string.hexdigits, ''.join(result.keys()))
 127
 128
 129 if __name__ == '__main__':
 130     unittest.main()