sdk/python/test_keep_client.py

   1 # usage example:
   2 #
   3 # ARVADOS_API_TOKEN=abc ARVADOS_API_HOST=arvados.local python -m unittest discover
   4
   5 import unittest
   6 import arvados
   7 import os
   8 import run_test_server
   9
  10 class KeepTestCase(unittest.TestCase):
  11     @classmethod
  12     def setUpClass(cls):
  13         super(KeepTestCase, cls).setUpClass()
  14         try:
  15             del os.environ['KEEP_LOCAL_STORE']
  16         except KeyError:
  17             pass
  18
  19         run_test_server.run()
  20         run_test_server.run_keep()
  21         arvados.keep.global_client_object = None
  22         arvados.config._settings = None
  23         run_test_server.authorize_with("admin")
  24
  25     @classmethod
  26     def tearDownClass(cls):
  27         super(KeepTestCase, cls).tearDownClass()
  28         run_test_server.stop()
  29         run_test_server.stop_keep()
  30
  31     def test_KeepBasicRWTest(self):
  32         foo_locator = arvados.Keep.put('foo')
  33         self.assertEqual(foo_locator,
  34                          'acbd18db4cc2f85cedef654fccc4a4d8+3',
  35                          'wrong md5 hash from Keep.put("foo"): ' + foo_locator)
  36         self.assertEqual(arvados.Keep.get(foo_locator),
  37                          'foo',
  38                          'wrong content from Keep.get(md5("foo"))')
  39
  40     def test_KeepBinaryRWTest(self):
  41         blob_str = '\xff\xfe\xf7\x00\x01\x02'
  42         blob_locator = arvados.Keep.put(blob_str)
  43         self.assertEqual(blob_locator,
  44                          '7fc7c53b45e53926ba52821140fef396+6',
  45                          ('wrong locator from Keep.put(<binarydata>):' +
  46                           blob_locator))
  47         self.assertEqual(arvados.Keep.get(blob_locator),
  48                          blob_str,
  49                          'wrong content from Keep.get(md5(<binarydata>))')
  50
  51     def test_KeepLongBinaryRWTest(self):
  52         blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
  53         for i in range(0,23):
  54             blob_str = blob_str + blob_str
  55         blob_locator = arvados.Keep.put(blob_str)
  56         self.assertEqual(blob_locator,
  57                          '84d90fc0d8175dd5dcfab04b999bc956+67108864',
  58                          ('wrong locator from Keep.put(<binarydata>): ' +
  59                           blob_locator))
  60         self.assertEqual(arvados.Keep.get(blob_locator),
  61                          blob_str,
  62                          'wrong content from Keep.get(md5(<binarydata>))')
  63
  64     def test_KeepSingleCopyRWTest(self):
  65         blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
  66         blob_locator = arvados.Keep.put(blob_str, copies=1)
  67         self.assertEqual(blob_locator,
  68                          'c902006bc98a3eb4a3663b65ab4a6fab+8',
  69                          ('wrong locator from Keep.put(<binarydata>): ' +
  70                           blob_locator))
  71         self.assertEqual(arvados.Keep.get(blob_locator),
  72                          blob_str,
  73                          'wrong content from Keep.get(md5(<binarydata>))')
  74
  75
  76 class KeepProxyTestCase(unittest.TestCase):
  77     @classmethod
  78     def setUpClass(cls):
  79         super(KeepProxyTestCase, cls).setUpClass()
  80         try:
  81             del os.environ['KEEP_LOCAL_STORE']
  82         except KeyError:
  83             pass
  84         run_test_server.run()
  85         run_test_server.run_keep()
  86         arvados.keep.global_client_object = None
  87         arvados.config._settings = None
  88         run_test_server.run_keep_proxy("admin")
  89
  90     @classmethod
  91     def tearDownClass(cls):
  92         super(KeepProxyTestCase, cls).tearDownClass()
  93         run_test_server.stop()
  94         run_test_server.stop_keep()
  95         run_test_server.stop_keep_proxy()
  96         os.environ["ARVADOS_KEEP_PROXY"] = ""
  97         os.environ["ARVADOS_EXTERNAL_CLIENT"] = ""
  98
  99     def test_KeepProxyTest1(self):
 100         # Will use ARVADOS_KEEP_PROXY environment variable that is set by
 101         # run_keep_proxy()
 102
 103         baz_locator = arvados.Keep.put('baz')
 104         self.assertEqual(baz_locator,
 105                          '73feffa4b7f6bb68e44cf984c85f6e88+3',
 106                          'wrong md5 hash from Keep.put("baz"): ' + baz_locator)
 107         self.assertEqual(arvados.Keep.get(baz_locator),
 108                          'baz',
 109                          'wrong content from Keep.get(md5("baz"))')
 110
 111         self.assertEqual(True, arvados.Keep.global_client_object().using_proxy)
 112
 113     def test_KeepProxyTest2(self):
 114         os.environ["ARVADOS_KEEP_PROXY"] = ""
 115         os.environ["ARVADOS_EXTERNAL_CLIENT"] = "true"
 116         arvados.config._settings = None
 117
 118         # Will send X-External-Client to server and get back the proxy from
 119         # keep_services/accessible
 120
 121         baz_locator = arvados.Keep.put('baz2')
 122         self.assertEqual(baz_locator,
 123                          '91f372a266fe2bf2823cb8ec7fda31ce+4',
 124                          'wrong md5 hash from Keep.put("baz2"): ' + baz_locator)
 125         self.assertEqual(arvados.Keep.get(baz_locator),
 126                          'baz2',
 127                          'wrong content from Keep.get(md5("baz2"))')
 128
 129         self.assertEqual(True, arvados.Keep.global_client_object().using_proxy)