15 import arvados_testutil as tutil
17 import run_test_server
19 class KeepTestCase(run_test_server.TestCaseWithServers):
25 super(KeepTestCase, cls).setUpClass()
26 run_test_server.authorize_with("admin")
27 cls.api_client = arvados.api('v1')
28 cls.keep_client = arvados.KeepClient(api_client=cls.api_client,
29 proxy='', local_store='')
31 def test_KeepBasicRWTest(self):
32 foo_locator = self.keep_client.put('foo')
33 self.assertRegexpMatches(
35 '^acbd18db4cc2f85cedef654fccc4a4d8\+3',
36 'wrong md5 hash from Keep.put("foo"): ' + foo_locator)
37 self.assertEqual(self.keep_client.get(foo_locator),
39 'wrong content from Keep.get(md5("foo"))')
41 def test_KeepBinaryRWTest(self):
42 blob_str = '\xff\xfe\xf7\x00\x01\x02'
43 blob_locator = self.keep_client.put(blob_str)
44 self.assertRegexpMatches(
46 '^7fc7c53b45e53926ba52821140fef396\+6',
47 ('wrong locator from Keep.put(<binarydata>):' + blob_locator))
48 self.assertEqual(self.keep_client.get(blob_locator),
50 'wrong content from Keep.get(md5(<binarydata>))')
52 def test_KeepLongBinaryRWTest(self):
53 blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
55 blob_str = blob_str + blob_str
56 blob_locator = self.keep_client.put(blob_str)
57 self.assertRegexpMatches(
59 '^84d90fc0d8175dd5dcfab04b999bc956\+67108864',
60 ('wrong locator from Keep.put(<binarydata>): ' + blob_locator))
61 self.assertEqual(self.keep_client.get(blob_locator),
63 'wrong content from Keep.get(md5(<binarydata>))')
65 def test_KeepSingleCopyRWTest(self):
66 blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
67 blob_locator = self.keep_client.put(blob_str, copies=1)
68 self.assertRegexpMatches(
70 '^c902006bc98a3eb4a3663b65ab4a6fab\+8',
71 ('wrong locator from Keep.put(<binarydata>): ' + blob_locator))
72 self.assertEqual(self.keep_client.get(blob_locator),
74 'wrong content from Keep.get(md5(<binarydata>))')
76 def test_KeepEmptyCollectionTest(self):
77 blob_locator = self.keep_client.put('', copies=1)
78 self.assertRegexpMatches(
80 '^d41d8cd98f00b204e9800998ecf8427e\+0',
81 ('wrong locator from Keep.put(""): ' + blob_locator))
83 def test_unicode_must_be_ascii(self):
84 # If unicode type, must only consist of valid ASCII
85 foo_locator = self.keep_client.put(u'foo')
86 self.assertRegexpMatches(
88 '^acbd18db4cc2f85cedef654fccc4a4d8\+3',
89 'wrong md5 hash from Keep.put("foo"): ' + foo_locator)
91 with self.assertRaises(UnicodeEncodeError):
92 # Error if it is not ASCII
93 self.keep_client.put(u'\xe2')
95 with self.assertRaises(arvados.errors.ArgumentError):
96 # Must be a string type
97 self.keep_client.put({})
99 class KeepPermissionTestCase(run_test_server.TestCaseWithServers):
101 KEEP_SERVER = {'blob_signing_key': 'abcdefghijk0123456789',
102 'enforce_permissions': True}
104 def test_KeepBasicRWTest(self):
105 run_test_server.authorize_with('active')
106 keep_client = arvados.KeepClient()
107 foo_locator = keep_client.put('foo')
108 self.assertRegexpMatches(
110 r'^acbd18db4cc2f85cedef654fccc4a4d8\+3\+A[a-f0-9]+@[a-f0-9]+$',
111 'invalid locator from Keep.put("foo"): ' + foo_locator)
112 self.assertEqual(keep_client.get(foo_locator),
114 'wrong content from Keep.get(md5("foo"))')
116 # GET with an unsigned locator => NotFound
117 bar_locator = keep_client.put('bar')
118 unsigned_bar_locator = "37b51d194a7513e45b56f6524f2d51f2+3"
119 self.assertRegexpMatches(
121 r'^37b51d194a7513e45b56f6524f2d51f2\+3\+A[a-f0-9]+@[a-f0-9]+$',
122 'invalid locator from Keep.put("bar"): ' + bar_locator)
123 self.assertRaises(arvados.errors.NotFoundError,
125 unsigned_bar_locator)
127 # GET from a different user => NotFound
128 run_test_server.authorize_with('spectator')
129 self.assertRaises(arvados.errors.NotFoundError,
133 # Unauthenticated GET for a signed locator => NotFound
134 # Unauthenticated GET for an unsigned locator => NotFound
135 keep_client.api_token = ''
136 self.assertRaises(arvados.errors.NotFoundError,
139 self.assertRaises(arvados.errors.NotFoundError,
141 unsigned_bar_locator)
144 # KeepOptionalPermission: starts Keep with --permission-key-file
145 # but not --enforce-permissions (i.e. generate signatures on PUT
146 # requests, but do not require them for GET requests)
148 # All of these requests should succeed when permissions are optional:
149 # * authenticated request, signed locator
150 # * authenticated request, unsigned locator
151 # * unauthenticated request, signed locator
152 # * unauthenticated request, unsigned locator
153 class KeepOptionalPermission(run_test_server.TestCaseWithServers):
155 KEEP_SERVER = {'blob_signing_key': 'abcdefghijk0123456789',
156 'enforce_permissions': False}
160 super(KeepOptionalPermission, cls).setUpClass()
161 run_test_server.authorize_with("admin")
162 cls.api_client = arvados.api('v1')
165 super(KeepOptionalPermission, self).setUp()
166 self.keep_client = arvados.KeepClient(api_client=self.api_client,
167 proxy='', local_store='')
169 def _put_foo_and_check(self):
170 signed_locator = self.keep_client.put('foo')
171 self.assertRegexpMatches(
173 r'^acbd18db4cc2f85cedef654fccc4a4d8\+3\+A[a-f0-9]+@[a-f0-9]+$',
174 'invalid locator from Keep.put("foo"): ' + signed_locator)
175 return signed_locator
177 def test_KeepAuthenticatedSignedTest(self):
178 signed_locator = self._put_foo_and_check()
179 self.assertEqual(self.keep_client.get(signed_locator),
181 'wrong content from Keep.get(md5("foo"))')
183 def test_KeepAuthenticatedUnsignedTest(self):
184 signed_locator = self._put_foo_and_check()
185 self.assertEqual(self.keep_client.get("acbd18db4cc2f85cedef654fccc4a4d8"),
187 'wrong content from Keep.get(md5("foo"))')
189 def test_KeepUnauthenticatedSignedTest(self):
190 # Check that signed GET requests work even when permissions
191 # enforcement is off.
192 signed_locator = self._put_foo_and_check()
193 self.keep_client.api_token = ''
194 self.assertEqual(self.keep_client.get(signed_locator),
196 'wrong content from Keep.get(md5("foo"))')
198 def test_KeepUnauthenticatedUnsignedTest(self):
199 # Since --enforce-permissions is not in effect, GET requests
200 # need not be authenticated.
201 signed_locator = self._put_foo_and_check()
202 self.keep_client.api_token = ''
203 self.assertEqual(self.keep_client.get("acbd18db4cc2f85cedef654fccc4a4d8"),
205 'wrong content from Keep.get(md5("foo"))')
208 class KeepProxyTestCase(run_test_server.TestCaseWithServers):
211 KEEP_PROXY_SERVER = {}
215 super(KeepProxyTestCase, cls).setUpClass()
216 run_test_server.authorize_with('active')
217 cls.api_client = arvados.api('v1')
220 arvados.config.settings().pop('ARVADOS_EXTERNAL_CLIENT', None)
221 super(KeepProxyTestCase, self).tearDown()
223 def test_KeepProxyTest1(self):
224 # Will use ARVADOS_KEEP_PROXY environment variable that is set by
226 keep_client = arvados.KeepClient(api_client=self.api_client,
228 baz_locator = keep_client.put('baz')
229 self.assertRegexpMatches(
231 '^73feffa4b7f6bb68e44cf984c85f6e88\+3',
232 'wrong md5 hash from Keep.put("baz"): ' + baz_locator)
233 self.assertEqual(keep_client.get(baz_locator),
235 'wrong content from Keep.get(md5("baz"))')
236 self.assertTrue(keep_client.using_proxy)
238 def test_KeepProxyTest2(self):
239 # Don't instantiate the proxy directly, but set the X-External-Client
240 # header. The API server should direct us to the proxy.
241 arvados.config.settings()['ARVADOS_EXTERNAL_CLIENT'] = 'true'
242 keep_client = arvados.KeepClient(api_client=self.api_client,
243 proxy='', local_store='')
244 baz_locator = keep_client.put('baz2')
245 self.assertRegexpMatches(
247 '^91f372a266fe2bf2823cb8ec7fda31ce\+4',
248 'wrong md5 hash from Keep.put("baz2"): ' + baz_locator)
249 self.assertEqual(keep_client.get(baz_locator),
251 'wrong content from Keep.get(md5("baz2"))')
252 self.assertTrue(keep_client.using_proxy)
255 class KeepClientServiceTestCase(unittest.TestCase, tutil.ApiClientMock):
256 def get_service_roots(self, api_client):
257 keep_client = arvados.KeepClient(api_client=api_client)
258 services = keep_client.weighted_service_roots(arvados.KeepLocator('0'*32))
259 return [urlparse.urlparse(url) for url in sorted(services)]
261 def test_ssl_flag_respected_in_roots(self):
262 for ssl_flag in [False, True]:
263 services = self.get_service_roots(self.mock_keep_services(
264 service_ssl_flag=ssl_flag))
266 ('https' if ssl_flag else 'http'), services[0].scheme)
268 def test_correct_ports_with_ipv6_addresses(self):
269 service = self.get_service_roots(self.mock_keep_services(
270 service_type='proxy', service_host='100::1', service_port=10, count=1))[0]
271 self.assertEqual('100::1', service.hostname)
272 self.assertEqual(10, service.port)
274 # test_*_timeout verify that KeepClient instructs pycurl to use
275 # the appropriate connection and read timeouts. They don't care
276 # whether pycurl actually exhibits the expected timeout behavior
277 # -- those tests are in the KeepClientTimeout test class.
279 def test_get_timeout(self):
280 api_client = self.mock_keep_services(count=1)
281 force_timeout = socket.timeout("timed out")
282 with tutil.mock_keep_responses(force_timeout, 0) as mock:
283 keep_client = arvados.KeepClient(api_client=api_client)
284 with self.assertRaises(arvados.errors.KeepReadError):
285 keep_client.get('ffffffffffffffffffffffffffffffff')
287 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
288 int(arvados.KeepClient.DEFAULT_TIMEOUT[0]*1000))
290 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
291 int(arvados.KeepClient.DEFAULT_TIMEOUT[1]))
293 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
294 int(arvados.KeepClient.DEFAULT_TIMEOUT[2]))
296 def test_put_timeout(self):
297 api_client = self.mock_keep_services(count=1)
298 force_timeout = socket.timeout("timed out")
299 with tutil.mock_keep_responses(force_timeout, 0) as mock:
300 keep_client = arvados.KeepClient(api_client=api_client)
301 with self.assertRaises(arvados.errors.KeepWriteError):
302 keep_client.put('foo')
304 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
305 int(arvados.KeepClient.DEFAULT_TIMEOUT[0]*1000))
307 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
308 int(arvados.KeepClient.DEFAULT_TIMEOUT[1]))
310 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
311 int(arvados.KeepClient.DEFAULT_TIMEOUT[2]))
313 def test_proxy_get_timeout(self):
314 api_client = self.mock_keep_services(service_type='proxy', count=1)
315 force_timeout = socket.timeout("timed out")
316 with tutil.mock_keep_responses(force_timeout, 0) as mock:
317 keep_client = arvados.KeepClient(api_client=api_client)
318 with self.assertRaises(arvados.errors.KeepReadError):
319 keep_client.get('ffffffffffffffffffffffffffffffff')
321 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
322 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[0]*1000))
324 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
325 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[1]))
327 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
328 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[2]))
330 def test_proxy_put_timeout(self):
331 api_client = self.mock_keep_services(service_type='proxy', count=1)
332 force_timeout = socket.timeout("timed out")
333 with tutil.mock_keep_responses(force_timeout, 0) as mock:
334 keep_client = arvados.KeepClient(api_client=api_client)
335 with self.assertRaises(arvados.errors.KeepWriteError):
336 keep_client.put('foo')
338 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
339 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[0]*1000))
341 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
342 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[1]))
344 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
345 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[2]))
347 def check_no_services_error(self, verb, exc_class):
348 api_client = mock.MagicMock(name='api_client')
349 api_client.keep_services().accessible().execute.side_effect = (
350 arvados.errors.ApiError)
351 keep_client = arvados.KeepClient(api_client=api_client)
352 with self.assertRaises(exc_class) as err_check:
353 getattr(keep_client, verb)('d41d8cd98f00b204e9800998ecf8427e+0')
354 self.assertEqual(0, len(err_check.exception.request_errors()))
356 def test_get_error_with_no_services(self):
357 self.check_no_services_error('get', arvados.errors.KeepReadError)
359 def test_put_error_with_no_services(self):
360 self.check_no_services_error('put', arvados.errors.KeepWriteError)
362 def check_errors_from_last_retry(self, verb, exc_class):
363 api_client = self.mock_keep_services(count=2)
364 req_mock = tutil.mock_keep_responses(
365 "retry error reporting test", 500, 500, 403, 403)
366 with req_mock, tutil.skip_sleep, \
367 self.assertRaises(exc_class) as err_check:
368 keep_client = arvados.KeepClient(api_client=api_client)
369 getattr(keep_client, verb)('d41d8cd98f00b204e9800998ecf8427e+0',
371 self.assertEqual([403, 403], [
372 getattr(error, 'status_code', None)
373 for error in err_check.exception.request_errors().itervalues()])
375 def test_get_error_reflects_last_retry(self):
376 self.check_errors_from_last_retry('get', arvados.errors.KeepReadError)
378 def test_put_error_reflects_last_retry(self):
379 self.check_errors_from_last_retry('put', arvados.errors.KeepWriteError)
381 def test_put_error_does_not_include_successful_puts(self):
382 data = 'partial failure test'
383 data_loc = tutil.str_keep_locator(data)
384 api_client = self.mock_keep_services(count=3)
385 with tutil.mock_keep_responses(data_loc, 200, 500, 500) as req_mock, \
386 self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
387 keep_client = arvados.KeepClient(api_client=api_client)
388 keep_client.put(data)
389 self.assertEqual(2, len(exc_check.exception.request_errors()))
391 def test_proxy_put_with_no_writable_services(self):
392 data = 'test with no writable services'
393 data_loc = tutil.str_keep_locator(data)
394 api_client = self.mock_keep_services(service_type='proxy', read_only=True, count=1)
395 with tutil.mock_keep_responses(data_loc, 200, 500, 500) as req_mock, \
396 self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
397 keep_client = arvados.KeepClient(api_client=api_client)
398 keep_client.put(data)
399 self.assertEqual(True, ("no Keep services available" in str(exc_check.exception)))
400 self.assertEqual(0, len(exc_check.exception.request_errors()))
402 def test_oddball_service_get(self):
403 body = 'oddball service get'
404 api_client = self.mock_keep_services(service_type='fancynewblobstore')
405 with tutil.mock_keep_responses(body, 200):
406 keep_client = arvados.KeepClient(api_client=api_client)
407 actual = keep_client.get(tutil.str_keep_locator(body))
408 self.assertEqual(body, actual)
410 def test_oddball_service_put(self):
411 body = 'oddball service put'
412 pdh = tutil.str_keep_locator(body)
413 api_client = self.mock_keep_services(service_type='fancynewblobstore')
414 with tutil.mock_keep_responses(pdh, 200):
415 keep_client = arvados.KeepClient(api_client=api_client)
416 actual = keep_client.put(body, copies=1)
417 self.assertEqual(pdh, actual)
419 def test_oddball_service_writer_count(self):
420 body = 'oddball service writer count'
421 pdh = tutil.str_keep_locator(body)
422 api_client = self.mock_keep_services(service_type='fancynewblobstore',
424 headers = {'x-keep-replicas-stored': 3}
425 with tutil.mock_keep_responses(pdh, 200, 418, 418, 418,
426 **headers) as req_mock:
427 keep_client = arvados.KeepClient(api_client=api_client)
428 actual = keep_client.put(body, copies=2)
429 self.assertEqual(pdh, actual)
430 self.assertEqual(1, req_mock.call_count)
434 class KeepClientRendezvousTestCase(unittest.TestCase, tutil.ApiClientMock):
437 # expected_order[i] is the probe order for
438 # hash=md5(sprintf("%064x",i)) where there are 16 services
439 # with uuid sprintf("anything-%015x",j) with j in 0..15. E.g.,
440 # the first probe for the block consisting of 64 "0"
441 # characters is the service whose uuid is
442 # "zzzzz-bi6l4-000000000000003", so expected_order[0][0]=='3'.
444 self.expected_order = [
445 list('3eab2d5fc9681074'),
446 list('097dba52e648f1c3'),
447 list('c5b4e023f8a7d691'),
448 list('9d81c02e76a3bf54'),
452 for x in range(len(self.expected_order))]
454 hashlib.md5(self.blocks[x]).hexdigest()
455 for x in range(len(self.expected_order))]
456 self.api_client = self.mock_keep_services(count=self.services)
457 self.keep_client = arvados.KeepClient(api_client=self.api_client)
459 def test_weighted_service_roots_against_reference_set(self):
460 # Confirm weighted_service_roots() returns the correct order
461 for i, hash in enumerate(self.hashes):
462 roots = self.keep_client.weighted_service_roots(arvados.KeepLocator(hash))
464 re.search(r'//\[?keep0x([0-9a-f]+)', root).group(1)
466 self.assertEqual(self.expected_order[i], got_order)
468 def test_get_probe_order_against_reference_set(self):
469 self._test_probe_order_against_reference_set(
470 lambda i: self.keep_client.get(self.hashes[i], num_retries=1))
472 def test_put_probe_order_against_reference_set(self):
473 # copies=1 prevents the test from being sensitive to races
474 # between writer threads.
475 self._test_probe_order_against_reference_set(
476 lambda i: self.keep_client.put(self.blocks[i], num_retries=1, copies=1))
478 def _test_probe_order_against_reference_set(self, op):
479 for i in range(len(self.blocks)):
480 with tutil.mock_keep_responses('', *[500 for _ in range(self.services*2)]) as mock, \
481 self.assertRaises(arvados.errors.KeepRequestError):
484 re.search(r'//\[?keep0x([0-9a-f]+)', resp.getopt(pycurl.URL)).group(1)
485 for resp in mock.responses]
486 self.assertEqual(self.expected_order[i]*2, got_order)
488 def test_put_probe_order_multiple_copies(self):
489 for copies in range(2, 4):
490 for i in range(len(self.blocks)):
491 with tutil.mock_keep_responses('', *[500 for _ in range(self.services*3)]) as mock, \
492 self.assertRaises(arvados.errors.KeepWriteError):
493 self.keep_client.put(self.blocks[i], num_retries=2, copies=copies)
495 re.search(r'//\[?keep0x([0-9a-f]+)', resp.getopt(pycurl.URL)).group(1)
496 for resp in mock.responses]
497 # With T threads racing to make requests, the position
498 # of a given server in the sequence of HTTP requests
499 # (got_order) cannot be more than T-1 positions
500 # earlier than that server's position in the reference
501 # probe sequence (expected_order).
503 # Loop invariant: we have accounted for +pos+ expected
504 # probes, either by seeing them in +got_order+ or by
505 # putting them in +pending+ in the hope of seeing them
506 # later. As long as +len(pending)<T+, we haven't
507 # started a request too early.
509 for pos, expected in enumerate(self.expected_order[i]*3):
510 got = got_order[pos-len(pending)]
511 while got in pending:
512 del pending[pending.index(got)]
513 got = got_order[pos-len(pending)]
515 pending.append(expected)
517 len(pending), copies,
518 "pending={}, with copies={}, got {}, expected {}".format(
519 pending, copies, repr(got_order), repr(self.expected_order[i]*3)))
521 def test_probe_waste_adding_one_server(self):
523 hashlib.md5("{:064x}".format(x)).hexdigest() for x in range(100)]
524 initial_services = 12
525 self.api_client = self.mock_keep_services(count=initial_services)
526 self.keep_client = arvados.KeepClient(api_client=self.api_client)
528 self.keep_client.weighted_service_roots(arvados.KeepLocator(hash)) for hash in hashes]
529 for added_services in range(1, 12):
530 api_client = self.mock_keep_services(count=initial_services+added_services)
531 keep_client = arvados.KeepClient(api_client=api_client)
533 for hash_index in range(len(hashes)):
534 probe_after = keep_client.weighted_service_roots(
535 arvados.KeepLocator(hashes[hash_index]))
536 penalty = probe_after.index(probes_before[hash_index][0])
537 self.assertLessEqual(penalty, added_services)
538 total_penalty += penalty
539 # Average penalty per block should not exceed
540 # N(added)/N(orig) by more than 20%, and should get closer
541 # to the ideal as we add data points.
544 len(hashes) / initial_services)
547 (120 - added_services)/100)
549 expect_penalty * 8/10)
551 min_penalty <= total_penalty <= max_penalty,
552 "With {}+{} services, {} blocks, penalty {} but expected {}..{}".format(
560 def check_64_zeros_error_order(self, verb, exc_class):
563 data = tutil.str_keep_locator(data)
564 # Arbitrary port number:
565 aport = random.randint(1024,65535)
566 api_client = self.mock_keep_services(service_port=aport, count=self.services)
567 keep_client = arvados.KeepClient(api_client=api_client)
568 with mock.patch('pycurl.Curl') as curl_mock, \
569 self.assertRaises(exc_class) as err_check:
570 curl_mock.return_value.side_effect = socket.timeout
571 getattr(keep_client, verb)(data)
572 urls = [urlparse.urlparse(url)
573 for url in err_check.exception.request_errors()]
574 self.assertEqual([('keep0x' + c, aport) for c in '3eab2d5fc9681074'],
575 [(url.hostname, url.port) for url in urls])
577 def test_get_error_shows_probe_order(self):
578 self.check_64_zeros_error_order('get', arvados.errors.KeepReadError)
580 def test_put_error_shows_probe_order(self):
581 self.check_64_zeros_error_order('put', arvados.errors.KeepWriteError)
584 class KeepClientTimeout(unittest.TestCase, tutil.ApiClientMock):
585 # BANDWIDTH_LOW_LIM must be less than len(DATA) so we can transfer
586 # 1s worth of data and then trigger bandwidth errors before running
589 BANDWIDTH_LOW_LIM = 1024
592 class assertTakesBetween(unittest.TestCase):
593 def __init__(self, tmin, tmax):
598 self.t0 = time.time()
600 def __exit__(self, *args, **kwargs):
601 # Round times to milliseconds, like CURL. Otherwise, we
602 # fail when CURL reaches a 1s timeout at 0.9998s.
603 delta = round(time.time() - self.t0, 3)
604 self.assertGreaterEqual(delta, self.tmin)
605 self.assertLessEqual(delta, self.tmax)
607 class assertTakesGreater(unittest.TestCase):
608 def __init__(self, tmin):
612 self.t0 = time.time()
614 def __exit__(self, *args, **kwargs):
615 delta = round(time.time() - self.t0, 3)
616 self.assertGreaterEqual(delta, self.tmin)
619 sock = socket.socket()
620 sock.bind(('0.0.0.0', 0))
621 self.port = sock.getsockname()[1]
623 self.server = keepstub.Server(('0.0.0.0', self.port), keepstub.Handler)
624 self.thread = threading.Thread(target=self.server.serve_forever)
625 self.thread.daemon = True # Exit thread if main proc exits
627 self.api_client = self.mock_keep_services(
629 service_host='localhost',
630 service_port=self.port,
634 self.server.shutdown()
636 def keepClient(self, timeouts=(0.1, TIMEOUT_TIME, BANDWIDTH_LOW_LIM)):
637 return arvados.KeepClient(
638 api_client=self.api_client,
641 def test_timeout_slow_connect(self):
642 # Can't simulate TCP delays with our own socket. Leave our
643 # stub server running uselessly, and try to connect to an
644 # unroutable IP address instead.
645 self.api_client = self.mock_keep_services(
647 service_host='240.0.0.0',
649 with self.assertTakesBetween(0.1, 0.5):
650 with self.assertRaises(arvados.errors.KeepWriteError):
651 self.keepClient().put(self.DATA, copies=1, num_retries=0)
653 def test_low_bandwidth_no_delays_success(self):
654 self.server.setbandwidth(2*self.BANDWIDTH_LOW_LIM)
655 kc = self.keepClient()
656 loc = kc.put(self.DATA, copies=1, num_retries=0)
657 self.assertEqual(self.DATA, kc.get(loc, num_retries=0))
659 def test_too_low_bandwidth_no_delays_failure(self):
660 # Check that lessening bandwidth corresponds to failing
661 kc = self.keepClient()
662 loc = kc.put(self.DATA, copies=1, num_retries=0)
663 self.server.setbandwidth(0.5*self.BANDWIDTH_LOW_LIM)
664 with self.assertTakesGreater(self.TIMEOUT_TIME):
665 with self.assertRaises(arvados.errors.KeepReadError) as e:
666 kc.get(loc, num_retries=0)
667 with self.assertTakesGreater(self.TIMEOUT_TIME):
668 with self.assertRaises(arvados.errors.KeepWriteError):
669 kc.put(self.DATA, copies=1, num_retries=0)
671 def test_low_bandwidth_with_server_response_delay_failure(self):
672 kc = self.keepClient()
673 loc = kc.put(self.DATA, copies=1, num_retries=0)
674 self.server.setbandwidth(self.BANDWIDTH_LOW_LIM)
675 self.server.setdelays(response=self.TIMEOUT_TIME)
676 with self.assertTakesGreater(self.TIMEOUT_TIME):
677 with self.assertRaises(arvados.errors.KeepReadError) as e:
678 kc.get(loc, num_retries=0)
679 with self.assertTakesGreater(self.TIMEOUT_TIME):
680 with self.assertRaises(arvados.errors.KeepWriteError):
681 kc.put(self.DATA, copies=1, num_retries=0)
683 def test_low_bandwidth_with_server_mid_delay_failure(self):
684 kc = self.keepClient()
685 loc = kc.put(self.DATA, copies=1, num_retries=0)
686 self.server.setbandwidth(self.BANDWIDTH_LOW_LIM)
687 self.server.setdelays(mid_write=self.TIMEOUT_TIME, mid_read=self.TIMEOUT_TIME)
688 with self.assertTakesGreater(self.TIMEOUT_TIME):
689 with self.assertRaises(arvados.errors.KeepReadError) as e:
690 kc.get(loc, num_retries=0)
691 with self.assertTakesGreater(self.TIMEOUT_TIME):
692 with self.assertRaises(arvados.errors.KeepWriteError):
693 kc.put(self.DATA, copies=1, num_retries=0)
695 def test_timeout_slow_request(self):
696 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
697 self.server.setdelays(request=.2)
698 self._test_connect_timeout_under_200ms(loc)
699 self.server.setdelays(request=2)
700 self._test_response_timeout_under_2s(loc)
702 def test_timeout_slow_response(self):
703 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
704 self.server.setdelays(response=.2)
705 self._test_connect_timeout_under_200ms(loc)
706 self.server.setdelays(response=2)
707 self._test_response_timeout_under_2s(loc)
709 def test_timeout_slow_response_body(self):
710 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
711 self.server.setdelays(response_body=.2)
712 self._test_connect_timeout_under_200ms(loc)
713 self.server.setdelays(response_body=2)
714 self._test_response_timeout_under_2s(loc)
716 def _test_connect_timeout_under_200ms(self, loc):
717 # Allow 100ms to connect, then 1s for response. Everything
718 # should work, and everything should take at least 200ms to
720 kc = self.keepClient(timeouts=(.1, 1))
721 with self.assertTakesBetween(.2, .3):
722 kc.put(self.DATA, copies=1, num_retries=0)
723 with self.assertTakesBetween(.2, .3):
724 self.assertEqual(self.DATA, kc.get(loc, num_retries=0))
726 def _test_response_timeout_under_2s(self, loc):
727 # Allow 10s to connect, then 1s for response. Nothing should
728 # work, and everything should take at least 1s to return.
729 kc = self.keepClient(timeouts=(10, 1))
730 with self.assertTakesBetween(1, 1.9):
731 with self.assertRaises(arvados.errors.KeepReadError):
732 kc.get(loc, num_retries=0)
733 with self.assertTakesBetween(1, 1.9):
734 with self.assertRaises(arvados.errors.KeepWriteError):
735 kc.put(self.DATA, copies=1, num_retries=0)
738 class KeepClientGatewayTestCase(unittest.TestCase, tutil.ApiClientMock):
739 def mock_disks_and_gateways(self, disks=3, gateways=1):
741 'uuid': 'zzzzz-bi6l4-gateway{:08d}'.format(i),
742 'owner_uuid': 'zzzzz-tpzed-000000000000000',
743 'service_host': 'gatewayhost{}'.format(i),
744 'service_port': 12345,
745 'service_ssl_flag': True,
746 'service_type': 'gateway:test',
747 } for i in range(gateways)]
748 self.gateway_roots = [
749 "https://{service_host}:{service_port}/".format(**gw)
750 for gw in self.gateways]
751 self.api_client = self.mock_keep_services(
752 count=disks, additional_services=self.gateways)
753 self.keepClient = arvados.KeepClient(api_client=self.api_client)
755 @mock.patch('pycurl.Curl')
756 def test_get_with_gateway_hint_first(self, MockCurl):
757 MockCurl.return_value = tutil.FakeCurl.make(
758 code=200, body='foo', headers={'Content-Length': 3})
759 self.mock_disks_and_gateways()
760 locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@' + self.gateways[0]['uuid']
761 self.assertEqual('foo', self.keepClient.get(locator))
762 self.assertEqual(self.gateway_roots[0]+locator,
763 MockCurl.return_value.getopt(pycurl.URL))
765 @mock.patch('pycurl.Curl')
766 def test_get_with_gateway_hints_in_order(self, MockCurl):
770 tutil.FakeCurl.make(code=404, body='')
771 for _ in range(gateways+disks)
773 MockCurl.side_effect = tutil.queue_with(mocks)
774 self.mock_disks_and_gateways(gateways=gateways, disks=disks)
775 locator = '+'.join(['acbd18db4cc2f85cedef654fccc4a4d8+3'] +
776 ['K@'+gw['uuid'] for gw in self.gateways])
777 with self.assertRaises(arvados.errors.NotFoundError):
778 self.keepClient.get(locator)
779 # Gateways are tried first, in the order given.
780 for i, root in enumerate(self.gateway_roots):
781 self.assertEqual(root+locator,
782 mocks[i].getopt(pycurl.URL))
783 # Disk services are tried next.
784 for i in range(gateways, gateways+disks):
785 self.assertRegexpMatches(
786 mocks[i].getopt(pycurl.URL),
789 @mock.patch('pycurl.Curl')
790 def test_get_with_remote_proxy_hint(self, MockCurl):
791 MockCurl.return_value = tutil.FakeCurl.make(
792 code=200, body='foo', headers={'Content-Length': 3})
793 self.mock_disks_and_gateways()
794 locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@xyzzy'
795 self.assertEqual('foo', self.keepClient.get(locator))
796 self.assertEqual('https://keep.xyzzy.arvadosapi.com/'+locator,
797 MockCurl.return_value.getopt(pycurl.URL))
800 class KeepClientRetryTestMixin(object):
801 # Testing with a local Keep store won't exercise the retry behavior.
802 # Instead, our strategy is:
803 # * Create a client with one proxy specified (pointed at a black
804 # hole), so there's no need to instantiate an API client, and
805 # all HTTP requests come from one place.
806 # * Mock httplib's request method to provide simulated responses.
807 # This lets us test the retry logic extensively without relying on any
808 # supporting servers, and prevents side effects in case something hiccups.
809 # To use this mixin, define DEFAULT_EXPECT, DEFAULT_EXCEPTION, and
812 # Test classes must define TEST_PATCHER to a method that mocks
813 # out appropriate methods in the client.
815 PROXY_ADDR = 'http://[%s]:65535/' % (tutil.TEST_HOST,)
816 TEST_DATA = 'testdata'
817 TEST_LOCATOR = 'ef654c40ab4f1747fc699915d4f70902+8'
820 self.client_kwargs = {'proxy': self.PROXY_ADDR, 'local_store': ''}
822 def new_client(self, **caller_kwargs):
823 kwargs = self.client_kwargs.copy()
824 kwargs.update(caller_kwargs)
825 return arvados.KeepClient(**kwargs)
827 def run_method(self, *args, **kwargs):
828 raise NotImplementedError("test subclasses must define run_method")
830 def check_success(self, expected=None, *args, **kwargs):
832 expected = self.DEFAULT_EXPECT
833 self.assertEqual(expected, self.run_method(*args, **kwargs))
835 def check_exception(self, error_class=None, *args, **kwargs):
836 if error_class is None:
837 error_class = self.DEFAULT_EXCEPTION
838 self.assertRaises(error_class, self.run_method, *args, **kwargs)
840 def test_immediate_success(self):
841 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 200):
844 def test_retry_then_success(self):
845 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
846 self.check_success(num_retries=3)
848 def test_no_default_retry(self):
849 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
850 self.check_exception()
852 def test_no_retry_after_permanent_error(self):
853 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 403, 200):
854 self.check_exception(num_retries=3)
856 def test_error_after_retries_exhausted(self):
857 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 500, 200):
858 self.check_exception(num_retries=1)
860 def test_num_retries_instance_fallback(self):
861 self.client_kwargs['num_retries'] = 3
862 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
867 class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
868 DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_DATA
869 DEFAULT_EXCEPTION = arvados.errors.KeepReadError
870 HINTED_LOCATOR = KeepClientRetryTestMixin.TEST_LOCATOR + '+K@xyzzy'
871 TEST_PATCHER = staticmethod(tutil.mock_keep_responses)
873 def run_method(self, locator=KeepClientRetryTestMixin.TEST_LOCATOR,
875 return self.new_client().get(locator, *args, **kwargs)
877 def test_specific_exception_when_not_found(self):
878 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 200):
879 self.check_exception(arvados.errors.NotFoundError, num_retries=3)
881 def test_general_exception_with_mixed_errors(self):
882 # get should raise a NotFoundError if no server returns the block,
883 # and a high threshold of servers report that it's not found.
884 # This test rigs up 50/50 disagreement between two servers, and
885 # checks that it does not become a NotFoundError.
886 client = self.new_client()
887 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 500):
888 with self.assertRaises(arvados.errors.KeepReadError) as exc_check:
889 client.get(self.HINTED_LOCATOR)
890 self.assertNotIsInstance(
891 exc_check.exception, arvados.errors.NotFoundError,
892 "mixed errors raised NotFoundError")
894 def test_hint_server_can_succeed_without_retries(self):
895 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 200, 500):
896 self.check_success(locator=self.HINTED_LOCATOR)
898 def test_try_next_server_after_timeout(self):
899 with tutil.mock_keep_responses(
900 (socket.timeout("timed out"), 200),
901 (self.DEFAULT_EXPECT, 200)):
902 self.check_success(locator=self.HINTED_LOCATOR)
904 def test_retry_data_with_wrong_checksum(self):
905 with tutil.mock_keep_responses(
907 (self.DEFAULT_EXPECT, 200)):
908 self.check_success(locator=self.HINTED_LOCATOR)
912 class KeepClientRetryPutTestCase(KeepClientRetryTestMixin, unittest.TestCase):
913 DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_LOCATOR
914 DEFAULT_EXCEPTION = arvados.errors.KeepWriteError
915 TEST_PATCHER = staticmethod(tutil.mock_keep_responses)
917 def run_method(self, data=KeepClientRetryTestMixin.TEST_DATA,
918 copies=1, *args, **kwargs):
919 return self.new_client().put(data, copies, *args, **kwargs)
921 def test_do_not_send_multiple_copies_to_same_server(self):
922 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 200):
923 self.check_exception(copies=2, num_retries=3)