+ def test_proxy_put_with_no_writable_services(self):
+ data = 'test with no writable services'
+ data_loc = '{}+{}'.format(hashlib.md5(data).hexdigest(), len(data))
+ api_client = self.mock_keep_services(service_type='proxy', read_only=True, count=1)
+ with tutil.mock_keep_responses(data_loc, 200, 500, 500) as req_mock, \
+ self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
+ keep_client = arvados.KeepClient(api_client=api_client)
+ keep_client.put(data)
+ self.assertEqual(True, ("no Keep services available" in str(exc_check.exception)))
+ self.assertEqual(0, len(exc_check.exception.request_errors()))
+
+class KeepClientTimeout(unittest.TestCase, tutil.ApiClientMock):
+ DATA = 'x' * 2**10
+
+ class assertTakesBetween(unittest.TestCase):
+ def __init__(self, tmin, tmax):
+ self.tmin = tmin
+ self.tmax = tmax
+
+ def __enter__(self):
+ self.t0 = time.time()
+
+ def __exit__(self, *args, **kwargs):
+ self.assertGreater(time.time() - self.t0, self.tmin)
+ self.assertLess(time.time() - self.t0, self.tmax)
+
+ def setUp(self):
+ sock = socket.socket()
+ sock.bind(('0.0.0.0', 0))
+ self.port = sock.getsockname()[1]
+ sock.close()
+ self.server = keepstub.Server(('0.0.0.0', self.port), keepstub.Handler)
+ self.thread = threading.Thread(target=self.server.serve_forever)
+ self.thread.daemon = True # Exit thread if main proc exits
+ self.thread.start()
+ self.api_client = self.mock_keep_services(
+ count=1,
+ service_host='localhost',
+ service_port=self.port,
+ )
+
+ def tearDown(self):
+ self.server.shutdown()
+
+ def keepClient(self, timeouts=(0.1, 1.0)):
+ return arvados.KeepClient(
+ api_client=self.api_client,
+ timeout=timeouts)
+
+ def test_timeout_slow_connect(self):
+ # Can't simulate TCP delays with our own socket. Leave our
+ # stub server running uselessly, and try to connect to an
+ # unroutable IP address instead.
+ self.api_client = self.mock_keep_services(
+ count=1,
+ service_host='240.0.0.0',
+ )
+ with self.assertTakesBetween(0.1, 0.5):
+ with self.assertRaises(arvados.errors.KeepWriteError):
+ self.keepClient((0.1, 1)).put(self.DATA, copies=1, num_retries=0)
+
+ def test_timeout_slow_request(self):
+ self.server.setdelays(request=0.2)
+ self._test_200ms()
+
+ def test_timeout_slow_response(self):
+ self.server.setdelays(response=0.2)
+ self._test_200ms()
+
+ def test_timeout_slow_response_body(self):
+ self.server.setdelays(response_body=0.2)
+ self._test_200ms()
+
+ def _test_200ms(self):
+ """Connect should be t<100ms, request should be 200ms <= t < 300ms"""
+
+ # Allow 100ms to connect, then 1s for response. Everything
+ # should work, and everything should take at least 200ms to
+ # return.
+ kc = self.keepClient((.1, 1))
+ with self.assertTakesBetween(.2, .3):
+ loc = kc.put(self.DATA, copies=1, num_retries=0)
+ with self.assertTakesBetween(.2, .3):
+ self.assertEqual(self.DATA, kc.get(loc, num_retries=0))
+
+ # Allow 1s to connect, then 100ms for response. Nothing should
+ # work, and everything should take at least 100ms to return.
+ kc = self.keepClient((1, .1))
+ with self.assertTakesBetween(.1, .2):
+ with self.assertRaises(arvados.errors.KeepReadError):
+ kc.get(loc, num_retries=0)
+ with self.assertTakesBetween(.1, .2):
+ with self.assertRaises(arvados.errors.KeepWriteError):
+ kc.put(self.DATA, copies=1, num_retries=0)
+
+
+class KeepClientGatewayTestCase(unittest.TestCase, tutil.ApiClientMock):
+ def mock_disks_and_gateways(self, disks=3, gateways=1):
+ self.gateways = [{
+ 'uuid': 'zzzzz-bi6l4-gateway{:08d}'.format(i),
+ 'owner_uuid': 'zzzzz-tpzed-000000000000000',
+ 'service_host': 'gatewayhost{}'.format(i),
+ 'service_port': 12345,
+ 'service_ssl_flag': True,
+ 'service_type': 'gateway:test',
+ } for i in range(gateways)]
+ self.gateway_roots = [
+ "https://{service_host}:{service_port}/".format(**gw)
+ for gw in self.gateways]
+ self.api_client = self.mock_keep_services(
+ count=disks, additional_services=self.gateways)
+ self.keepClient = arvados.KeepClient(api_client=self.api_client)
+
+ @mock.patch('pycurl.Curl')
+ def test_get_with_gateway_hint_first(self, MockCurl):
+ MockCurl.return_value = tutil.FakeCurl.make(
+ code=200, body='foo', headers={'Content-Length': 3})
+ self.mock_disks_and_gateways()
+ locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@' + self.gateways[0]['uuid']
+ self.assertEqual('foo', self.keepClient.get(locator))
+ self.assertEqual(self.gateway_roots[0]+locator,
+ MockCurl.return_value.getopt(pycurl.URL))
+
+ @mock.patch('pycurl.Curl')
+ def test_get_with_gateway_hints_in_order(self, MockCurl):
+ gateways = 4
+ disks = 3
+ mocks = [
+ tutil.FakeCurl.make(code=404, body='')
+ for _ in range(gateways+disks)
+ ]
+ MockCurl.side_effect = tutil.queue_with(mocks)
+ self.mock_disks_and_gateways(gateways=gateways, disks=disks)
+ locator = '+'.join(['acbd18db4cc2f85cedef654fccc4a4d8+3'] +
+ ['K@'+gw['uuid'] for gw in self.gateways])
+ with self.assertRaises(arvados.errors.NotFoundError):
+ self.keepClient.get(locator)
+ # Gateways are tried first, in the order given.
+ for i, root in enumerate(self.gateway_roots):
+ self.assertEqual(root+locator,
+ mocks[i].getopt(pycurl.URL))
+ # Disk services are tried next.
+ for i in range(gateways, gateways+disks):
+ self.assertRegexpMatches(
+ mocks[i].getopt(pycurl.URL),
+ r'keep0x')
+
+ @mock.patch('pycurl.Curl')
+ def test_get_with_remote_proxy_hint(self, MockCurl):
+ MockCurl.return_value = tutil.FakeCurl.make(
+ code=200, body='foo', headers={'Content-Length': 3})
+ self.mock_disks_and_gateways()
+ locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@xyzzy'
+ self.assertEqual('foo', self.keepClient.get(locator))
+ self.assertEqual('https://keep.xyzzy.arvadosapi.com/'+locator,
+ MockCurl.return_value.getopt(pycurl.URL))
+