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 self.assertEqual(0, self.keep_client.upload_counter.get())
33 foo_locator = self.keep_client.put('foo')
34 self.assertRegexpMatches(
36 '^acbd18db4cc2f85cedef654fccc4a4d8\+3',
37 'wrong md5 hash from Keep.put("foo"): ' + foo_locator)
39 # 6 bytes because uploaded 2 copies
40 self.assertEqual(6, self.keep_client.upload_counter.get())
42 self.assertEqual(0, self.keep_client.download_counter.get())
43 self.assertEqual(self.keep_client.get(foo_locator),
45 'wrong content from Keep.get(md5("foo"))')
46 self.assertEqual(3, self.keep_client.download_counter.get())
48 def test_KeepBinaryRWTest(self):
49 blob_str = '\xff\xfe\xf7\x00\x01\x02'
50 blob_locator = self.keep_client.put(blob_str)
51 self.assertRegexpMatches(
53 '^7fc7c53b45e53926ba52821140fef396\+6',
54 ('wrong locator from Keep.put(<binarydata>):' + blob_locator))
55 self.assertEqual(self.keep_client.get(blob_locator),
57 'wrong content from Keep.get(md5(<binarydata>))')
59 def test_KeepLongBinaryRWTest(self):
60 blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
62 blob_str = blob_str + blob_str
63 blob_locator = self.keep_client.put(blob_str)
64 self.assertRegexpMatches(
66 '^84d90fc0d8175dd5dcfab04b999bc956\+67108864',
67 ('wrong locator from Keep.put(<binarydata>): ' + blob_locator))
68 self.assertEqual(self.keep_client.get(blob_locator),
70 'wrong content from Keep.get(md5(<binarydata>))')
72 @unittest.skip("unreliable test - please fix and close #8752")
73 def test_KeepSingleCopyRWTest(self):
74 blob_str = '\xff\xfe\xfd\xfc\x00\x01\x02\x03'
75 blob_locator = self.keep_client.put(blob_str, copies=1)
76 self.assertRegexpMatches(
78 '^c902006bc98a3eb4a3663b65ab4a6fab\+8',
79 ('wrong locator from Keep.put(<binarydata>): ' + blob_locator))
80 self.assertEqual(self.keep_client.get(blob_locator),
82 'wrong content from Keep.get(md5(<binarydata>))')
84 def test_KeepEmptyCollectionTest(self):
85 blob_locator = self.keep_client.put('', copies=1)
86 self.assertRegexpMatches(
88 '^d41d8cd98f00b204e9800998ecf8427e\+0',
89 ('wrong locator from Keep.put(""): ' + blob_locator))
91 def test_unicode_must_be_ascii(self):
92 # If unicode type, must only consist of valid ASCII
93 foo_locator = self.keep_client.put(u'foo')
94 self.assertRegexpMatches(
96 '^acbd18db4cc2f85cedef654fccc4a4d8\+3',
97 'wrong md5 hash from Keep.put("foo"): ' + foo_locator)
99 with self.assertRaises(UnicodeEncodeError):
100 # Error if it is not ASCII
101 self.keep_client.put(u'\xe2')
103 with self.assertRaises(arvados.errors.ArgumentError):
104 # Must be a string type
105 self.keep_client.put({})
107 class KeepPermissionTestCase(run_test_server.TestCaseWithServers):
109 KEEP_SERVER = {'blob_signing_key': 'abcdefghijk0123456789',
110 'enforce_permissions': True}
112 def test_KeepBasicRWTest(self):
113 run_test_server.authorize_with('active')
114 keep_client = arvados.KeepClient()
115 foo_locator = keep_client.put('foo')
116 self.assertRegexpMatches(
118 r'^acbd18db4cc2f85cedef654fccc4a4d8\+3\+A[a-f0-9]+@[a-f0-9]+$',
119 'invalid locator from Keep.put("foo"): ' + foo_locator)
120 self.assertEqual(keep_client.get(foo_locator),
122 'wrong content from Keep.get(md5("foo"))')
124 # GET with an unsigned locator => NotFound
125 bar_locator = keep_client.put('bar')
126 unsigned_bar_locator = "37b51d194a7513e45b56f6524f2d51f2+3"
127 self.assertRegexpMatches(
129 r'^37b51d194a7513e45b56f6524f2d51f2\+3\+A[a-f0-9]+@[a-f0-9]+$',
130 'invalid locator from Keep.put("bar"): ' + bar_locator)
131 self.assertRaises(arvados.errors.NotFoundError,
133 unsigned_bar_locator)
135 # GET from a different user => NotFound
136 run_test_server.authorize_with('spectator')
137 self.assertRaises(arvados.errors.NotFoundError,
141 # Unauthenticated GET for a signed locator => NotFound
142 # Unauthenticated GET for an unsigned locator => NotFound
143 keep_client.api_token = ''
144 self.assertRaises(arvados.errors.NotFoundError,
147 self.assertRaises(arvados.errors.NotFoundError,
149 unsigned_bar_locator)
152 # KeepOptionalPermission: starts Keep with --permission-key-file
153 # but not --enforce-permissions (i.e. generate signatures on PUT
154 # requests, but do not require them for GET requests)
156 # All of these requests should succeed when permissions are optional:
157 # * authenticated request, signed locator
158 # * authenticated request, unsigned locator
159 # * unauthenticated request, signed locator
160 # * unauthenticated request, unsigned locator
161 class KeepOptionalPermission(run_test_server.TestCaseWithServers):
163 KEEP_SERVER = {'blob_signing_key': 'abcdefghijk0123456789',
164 'enforce_permissions': False}
168 super(KeepOptionalPermission, cls).setUpClass()
169 run_test_server.authorize_with("admin")
170 cls.api_client = arvados.api('v1')
173 super(KeepOptionalPermission, self).setUp()
174 self.keep_client = arvados.KeepClient(api_client=self.api_client,
175 proxy='', local_store='')
177 def _put_foo_and_check(self):
178 signed_locator = self.keep_client.put('foo')
179 self.assertRegexpMatches(
181 r'^acbd18db4cc2f85cedef654fccc4a4d8\+3\+A[a-f0-9]+@[a-f0-9]+$',
182 'invalid locator from Keep.put("foo"): ' + signed_locator)
183 return signed_locator
185 def test_KeepAuthenticatedSignedTest(self):
186 signed_locator = self._put_foo_and_check()
187 self.assertEqual(self.keep_client.get(signed_locator),
189 'wrong content from Keep.get(md5("foo"))')
191 def test_KeepAuthenticatedUnsignedTest(self):
192 signed_locator = self._put_foo_and_check()
193 self.assertEqual(self.keep_client.get("acbd18db4cc2f85cedef654fccc4a4d8"),
195 'wrong content from Keep.get(md5("foo"))')
197 def test_KeepUnauthenticatedSignedTest(self):
198 # Check that signed GET requests work even when permissions
199 # enforcement is off.
200 signed_locator = self._put_foo_and_check()
201 self.keep_client.api_token = ''
202 self.assertEqual(self.keep_client.get(signed_locator),
204 'wrong content from Keep.get(md5("foo"))')
206 def test_KeepUnauthenticatedUnsignedTest(self):
207 # Since --enforce-permissions is not in effect, GET requests
208 # need not be authenticated.
209 signed_locator = self._put_foo_and_check()
210 self.keep_client.api_token = ''
211 self.assertEqual(self.keep_client.get("acbd18db4cc2f85cedef654fccc4a4d8"),
213 'wrong content from Keep.get(md5("foo"))')
216 class KeepProxyTestCase(run_test_server.TestCaseWithServers):
219 KEEP_PROXY_SERVER = {}
223 super(KeepProxyTestCase, cls).setUpClass()
224 run_test_server.authorize_with('active')
225 cls.api_client = arvados.api('v1')
228 arvados.config.settings().pop('ARVADOS_EXTERNAL_CLIENT', None)
229 super(KeepProxyTestCase, self).tearDown()
231 def test_KeepProxyTest1(self):
232 # Will use ARVADOS_KEEP_PROXY environment variable that is set by
234 keep_client = arvados.KeepClient(api_client=self.api_client,
236 baz_locator = keep_client.put('baz')
237 self.assertRegexpMatches(
239 '^73feffa4b7f6bb68e44cf984c85f6e88\+3',
240 'wrong md5 hash from Keep.put("baz"): ' + baz_locator)
241 self.assertEqual(keep_client.get(baz_locator),
243 'wrong content from Keep.get(md5("baz"))')
244 self.assertTrue(keep_client.using_proxy)
246 def test_KeepProxyTest2(self):
247 # Don't instantiate the proxy directly, but set the X-External-Client
248 # header. The API server should direct us to the proxy.
249 arvados.config.settings()['ARVADOS_EXTERNAL_CLIENT'] = 'true'
250 keep_client = arvados.KeepClient(api_client=self.api_client,
251 proxy='', local_store='')
252 baz_locator = keep_client.put('baz2')
253 self.assertRegexpMatches(
255 '^91f372a266fe2bf2823cb8ec7fda31ce\+4',
256 'wrong md5 hash from Keep.put("baz2"): ' + baz_locator)
257 self.assertEqual(keep_client.get(baz_locator),
259 'wrong content from Keep.get(md5("baz2"))')
260 self.assertTrue(keep_client.using_proxy)
263 class KeepClientServiceTestCase(unittest.TestCase, tutil.ApiClientMock):
264 def get_service_roots(self, api_client):
265 keep_client = arvados.KeepClient(api_client=api_client)
266 services = keep_client.weighted_service_roots(arvados.KeepLocator('0'*32))
267 return [urlparse.urlparse(url) for url in sorted(services)]
269 def test_ssl_flag_respected_in_roots(self):
270 for ssl_flag in [False, True]:
271 services = self.get_service_roots(self.mock_keep_services(
272 service_ssl_flag=ssl_flag))
274 ('https' if ssl_flag else 'http'), services[0].scheme)
276 def test_correct_ports_with_ipv6_addresses(self):
277 service = self.get_service_roots(self.mock_keep_services(
278 service_type='proxy', service_host='100::1', service_port=10, count=1))[0]
279 self.assertEqual('100::1', service.hostname)
280 self.assertEqual(10, service.port)
282 # test_*_timeout verify that KeepClient instructs pycurl to use
283 # the appropriate connection and read timeouts. They don't care
284 # whether pycurl actually exhibits the expected timeout behavior
285 # -- those tests are in the KeepClientTimeout test class.
287 def test_get_timeout(self):
288 api_client = self.mock_keep_services(count=1)
289 force_timeout = socket.timeout("timed out")
290 with tutil.mock_keep_responses(force_timeout, 0) as mock:
291 keep_client = arvados.KeepClient(api_client=api_client)
292 with self.assertRaises(arvados.errors.KeepReadError):
293 keep_client.get('ffffffffffffffffffffffffffffffff')
295 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
296 int(arvados.KeepClient.DEFAULT_TIMEOUT[0]*1000))
298 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
299 int(arvados.KeepClient.DEFAULT_TIMEOUT[1]))
301 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
302 int(arvados.KeepClient.DEFAULT_TIMEOUT[2]))
304 def test_put_timeout(self):
305 api_client = self.mock_keep_services(count=1)
306 force_timeout = socket.timeout("timed out")
307 with tutil.mock_keep_responses(force_timeout, 0) as mock:
308 keep_client = arvados.KeepClient(api_client=api_client)
309 with self.assertRaises(arvados.errors.KeepWriteError):
310 keep_client.put('foo')
312 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
313 int(arvados.KeepClient.DEFAULT_TIMEOUT[0]*1000))
315 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
316 int(arvados.KeepClient.DEFAULT_TIMEOUT[1]))
318 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
319 int(arvados.KeepClient.DEFAULT_TIMEOUT[2]))
321 def test_proxy_get_timeout(self):
322 api_client = self.mock_keep_services(service_type='proxy', count=1)
323 force_timeout = socket.timeout("timed out")
324 with tutil.mock_keep_responses(force_timeout, 0) as mock:
325 keep_client = arvados.KeepClient(api_client=api_client)
326 with self.assertRaises(arvados.errors.KeepReadError):
327 keep_client.get('ffffffffffffffffffffffffffffffff')
329 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
330 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[0]*1000))
332 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
333 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[1]))
335 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
336 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[2]))
338 def test_proxy_put_timeout(self):
339 api_client = self.mock_keep_services(service_type='proxy', count=1)
340 force_timeout = socket.timeout("timed out")
341 with tutil.mock_keep_responses(force_timeout, 0) as mock:
342 keep_client = arvados.KeepClient(api_client=api_client)
343 with self.assertRaises(arvados.errors.KeepWriteError):
344 keep_client.put('foo')
346 mock.responses[0].getopt(pycurl.CONNECTTIMEOUT_MS),
347 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[0]*1000))
349 mock.responses[0].getopt(pycurl.LOW_SPEED_TIME),
350 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[1]))
352 mock.responses[0].getopt(pycurl.LOW_SPEED_LIMIT),
353 int(arvados.KeepClient.DEFAULT_PROXY_TIMEOUT[2]))
355 def check_no_services_error(self, verb, exc_class):
356 api_client = mock.MagicMock(name='api_client')
357 api_client.keep_services().accessible().execute.side_effect = (
358 arvados.errors.ApiError)
359 keep_client = arvados.KeepClient(api_client=api_client)
360 with self.assertRaises(exc_class) as err_check:
361 getattr(keep_client, verb)('d41d8cd98f00b204e9800998ecf8427e+0')
362 self.assertEqual(0, len(err_check.exception.request_errors()))
364 def test_get_error_with_no_services(self):
365 self.check_no_services_error('get', arvados.errors.KeepReadError)
367 def test_put_error_with_no_services(self):
368 self.check_no_services_error('put', arvados.errors.KeepWriteError)
370 def check_errors_from_last_retry(self, verb, exc_class):
371 api_client = self.mock_keep_services(count=2)
372 req_mock = tutil.mock_keep_responses(
373 "retry error reporting test", 500, 500, 403, 403)
374 with req_mock, tutil.skip_sleep, \
375 self.assertRaises(exc_class) as err_check:
376 keep_client = arvados.KeepClient(api_client=api_client)
377 getattr(keep_client, verb)('d41d8cd98f00b204e9800998ecf8427e+0',
379 self.assertEqual([403, 403], [
380 getattr(error, 'status_code', None)
381 for error in err_check.exception.request_errors().itervalues()])
383 def test_get_error_reflects_last_retry(self):
384 self.check_errors_from_last_retry('get', arvados.errors.KeepReadError)
386 def test_put_error_reflects_last_retry(self):
387 self.check_errors_from_last_retry('put', arvados.errors.KeepWriteError)
389 def test_put_error_does_not_include_successful_puts(self):
390 data = 'partial failure test'
391 data_loc = tutil.str_keep_locator(data)
392 api_client = self.mock_keep_services(count=3)
393 with tutil.mock_keep_responses(data_loc, 200, 500, 500) as req_mock, \
394 self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
395 keep_client = arvados.KeepClient(api_client=api_client)
396 keep_client.put(data)
397 self.assertEqual(2, len(exc_check.exception.request_errors()))
399 def test_proxy_put_with_no_writable_services(self):
400 data = 'test with no writable services'
401 data_loc = tutil.str_keep_locator(data)
402 api_client = self.mock_keep_services(service_type='proxy', read_only=True, count=1)
403 with tutil.mock_keep_responses(data_loc, 200, 500, 500) as req_mock, \
404 self.assertRaises(arvados.errors.KeepWriteError) as exc_check:
405 keep_client = arvados.KeepClient(api_client=api_client)
406 keep_client.put(data)
407 self.assertEqual(True, ("no Keep services available" in str(exc_check.exception)))
408 self.assertEqual(0, len(exc_check.exception.request_errors()))
410 def test_oddball_service_get(self):
411 body = 'oddball service get'
412 api_client = self.mock_keep_services(service_type='fancynewblobstore')
413 with tutil.mock_keep_responses(body, 200):
414 keep_client = arvados.KeepClient(api_client=api_client)
415 actual = keep_client.get(tutil.str_keep_locator(body))
416 self.assertEqual(body, actual)
418 def test_oddball_service_put(self):
419 body = 'oddball service put'
420 pdh = tutil.str_keep_locator(body)
421 api_client = self.mock_keep_services(service_type='fancynewblobstore')
422 with tutil.mock_keep_responses(pdh, 200):
423 keep_client = arvados.KeepClient(api_client=api_client)
424 actual = keep_client.put(body, copies=1)
425 self.assertEqual(pdh, actual)
427 def test_oddball_service_writer_count(self):
428 body = 'oddball service writer count'
429 pdh = tutil.str_keep_locator(body)
430 api_client = self.mock_keep_services(service_type='fancynewblobstore',
432 headers = {'x-keep-replicas-stored': 3}
433 with tutil.mock_keep_responses(pdh, 200, 418, 418, 418,
434 **headers) as req_mock:
435 keep_client = arvados.KeepClient(api_client=api_client)
436 actual = keep_client.put(body, copies=2)
437 self.assertEqual(pdh, actual)
438 self.assertEqual(1, req_mock.call_count)
442 class KeepClientRendezvousTestCase(unittest.TestCase, tutil.ApiClientMock):
445 # expected_order[i] is the probe order for
446 # hash=md5(sprintf("%064x",i)) where there are 16 services
447 # with uuid sprintf("anything-%015x",j) with j in 0..15. E.g.,
448 # the first probe for the block consisting of 64 "0"
449 # characters is the service whose uuid is
450 # "zzzzz-bi6l4-000000000000003", so expected_order[0][0]=='3'.
452 self.expected_order = [
453 list('3eab2d5fc9681074'),
454 list('097dba52e648f1c3'),
455 list('c5b4e023f8a7d691'),
456 list('9d81c02e76a3bf54'),
460 for x in range(len(self.expected_order))]
462 hashlib.md5(self.blocks[x]).hexdigest()
463 for x in range(len(self.expected_order))]
464 self.api_client = self.mock_keep_services(count=self.services)
465 self.keep_client = arvados.KeepClient(api_client=self.api_client)
467 def test_weighted_service_roots_against_reference_set(self):
468 # Confirm weighted_service_roots() returns the correct order
469 for i, hash in enumerate(self.hashes):
470 roots = self.keep_client.weighted_service_roots(arvados.KeepLocator(hash))
472 re.search(r'//\[?keep0x([0-9a-f]+)', root).group(1)
474 self.assertEqual(self.expected_order[i], got_order)
476 def test_get_probe_order_against_reference_set(self):
477 self._test_probe_order_against_reference_set(
478 lambda i: self.keep_client.get(self.hashes[i], num_retries=1))
480 def test_put_probe_order_against_reference_set(self):
481 # copies=1 prevents the test from being sensitive to races
482 # between writer threads.
483 self._test_probe_order_against_reference_set(
484 lambda i: self.keep_client.put(self.blocks[i], num_retries=1, copies=1))
486 def _test_probe_order_against_reference_set(self, op):
487 for i in range(len(self.blocks)):
488 with tutil.mock_keep_responses('', *[500 for _ in range(self.services*2)]) as mock, \
489 self.assertRaises(arvados.errors.KeepRequestError):
492 re.search(r'//\[?keep0x([0-9a-f]+)', resp.getopt(pycurl.URL)).group(1)
493 for resp in mock.responses]
494 self.assertEqual(self.expected_order[i]*2, got_order)
496 def test_put_probe_order_multiple_copies(self):
497 for copies in range(2, 4):
498 for i in range(len(self.blocks)):
499 with tutil.mock_keep_responses('', *[500 for _ in range(self.services*3)]) as mock, \
500 self.assertRaises(arvados.errors.KeepWriteError):
501 self.keep_client.put(self.blocks[i], num_retries=2, copies=copies)
503 re.search(r'//\[?keep0x([0-9a-f]+)', resp.getopt(pycurl.URL)).group(1)
504 for resp in mock.responses]
505 # With T threads racing to make requests, the position
506 # of a given server in the sequence of HTTP requests
507 # (got_order) cannot be more than T-1 positions
508 # earlier than that server's position in the reference
509 # probe sequence (expected_order).
511 # Loop invariant: we have accounted for +pos+ expected
512 # probes, either by seeing them in +got_order+ or by
513 # putting them in +pending+ in the hope of seeing them
514 # later. As long as +len(pending)<T+, we haven't
515 # started a request too early.
517 for pos, expected in enumerate(self.expected_order[i]*3):
518 got = got_order[pos-len(pending)]
519 while got in pending:
520 del pending[pending.index(got)]
521 got = got_order[pos-len(pending)]
523 pending.append(expected)
525 len(pending), copies,
526 "pending={}, with copies={}, got {}, expected {}".format(
527 pending, copies, repr(got_order), repr(self.expected_order[i]*3)))
529 def test_probe_waste_adding_one_server(self):
531 hashlib.md5("{:064x}".format(x)).hexdigest() for x in range(100)]
532 initial_services = 12
533 self.api_client = self.mock_keep_services(count=initial_services)
534 self.keep_client = arvados.KeepClient(api_client=self.api_client)
536 self.keep_client.weighted_service_roots(arvados.KeepLocator(hash)) for hash in hashes]
537 for added_services in range(1, 12):
538 api_client = self.mock_keep_services(count=initial_services+added_services)
539 keep_client = arvados.KeepClient(api_client=api_client)
541 for hash_index in range(len(hashes)):
542 probe_after = keep_client.weighted_service_roots(
543 arvados.KeepLocator(hashes[hash_index]))
544 penalty = probe_after.index(probes_before[hash_index][0])
545 self.assertLessEqual(penalty, added_services)
546 total_penalty += penalty
547 # Average penalty per block should not exceed
548 # N(added)/N(orig) by more than 20%, and should get closer
549 # to the ideal as we add data points.
552 len(hashes) / initial_services)
555 (120 - added_services)/100)
557 expect_penalty * 8/10)
559 min_penalty <= total_penalty <= max_penalty,
560 "With {}+{} services, {} blocks, penalty {} but expected {}..{}".format(
568 def check_64_zeros_error_order(self, verb, exc_class):
571 data = tutil.str_keep_locator(data)
572 # Arbitrary port number:
573 aport = random.randint(1024,65535)
574 api_client = self.mock_keep_services(service_port=aport, count=self.services)
575 keep_client = arvados.KeepClient(api_client=api_client)
576 with mock.patch('pycurl.Curl') as curl_mock, \
577 self.assertRaises(exc_class) as err_check:
578 curl_mock.return_value.side_effect = socket.timeout
579 getattr(keep_client, verb)(data)
580 urls = [urlparse.urlparse(url)
581 for url in err_check.exception.request_errors()]
582 self.assertEqual([('keep0x' + c, aport) for c in '3eab2d5fc9681074'],
583 [(url.hostname, url.port) for url in urls])
585 def test_get_error_shows_probe_order(self):
586 self.check_64_zeros_error_order('get', arvados.errors.KeepReadError)
588 def test_put_error_shows_probe_order(self):
589 self.check_64_zeros_error_order('put', arvados.errors.KeepWriteError)
592 class KeepClientTimeout(unittest.TestCase, tutil.ApiClientMock):
593 # BANDWIDTH_LOW_LIM must be less than len(DATA) so we can transfer
594 # 1s worth of data and then trigger bandwidth errors before running
597 BANDWIDTH_LOW_LIM = 1024
600 class assertTakesBetween(unittest.TestCase):
601 def __init__(self, tmin, tmax):
606 self.t0 = time.time()
608 def __exit__(self, *args, **kwargs):
609 # Round times to milliseconds, like CURL. Otherwise, we
610 # fail when CURL reaches a 1s timeout at 0.9998s.
611 delta = round(time.time() - self.t0, 3)
612 self.assertGreaterEqual(delta, self.tmin)
613 self.assertLessEqual(delta, self.tmax)
615 class assertTakesGreater(unittest.TestCase):
616 def __init__(self, tmin):
620 self.t0 = time.time()
622 def __exit__(self, *args, **kwargs):
623 delta = round(time.time() - self.t0, 3)
624 self.assertGreaterEqual(delta, self.tmin)
627 sock = socket.socket()
628 sock.bind(('0.0.0.0', 0))
629 self.port = sock.getsockname()[1]
631 self.server = keepstub.Server(('0.0.0.0', self.port), keepstub.Handler)
632 self.thread = threading.Thread(target=self.server.serve_forever)
633 self.thread.daemon = True # Exit thread if main proc exits
635 self.api_client = self.mock_keep_services(
637 service_host='localhost',
638 service_port=self.port,
642 self.server.shutdown()
644 def keepClient(self, timeouts=(0.1, TIMEOUT_TIME, BANDWIDTH_LOW_LIM)):
645 return arvados.KeepClient(
646 api_client=self.api_client,
649 def test_timeout_slow_connect(self):
650 # Can't simulate TCP delays with our own socket. Leave our
651 # stub server running uselessly, and try to connect to an
652 # unroutable IP address instead.
653 self.api_client = self.mock_keep_services(
655 service_host='240.0.0.0',
657 with self.assertTakesBetween(0.1, 0.5):
658 with self.assertRaises(arvados.errors.KeepWriteError):
659 self.keepClient().put(self.DATA, copies=1, num_retries=0)
661 def test_low_bandwidth_no_delays_success(self):
662 self.server.setbandwidth(2*self.BANDWIDTH_LOW_LIM)
663 kc = self.keepClient()
664 loc = kc.put(self.DATA, copies=1, num_retries=0)
665 self.assertEqual(self.DATA, kc.get(loc, num_retries=0))
667 def test_too_low_bandwidth_no_delays_failure(self):
668 # Check that lessening bandwidth corresponds to failing
669 kc = self.keepClient()
670 loc = kc.put(self.DATA, copies=1, num_retries=0)
671 self.server.setbandwidth(0.5*self.BANDWIDTH_LOW_LIM)
672 with self.assertTakesGreater(self.TIMEOUT_TIME):
673 with self.assertRaises(arvados.errors.KeepReadError) as e:
674 kc.get(loc, num_retries=0)
675 with self.assertTakesGreater(self.TIMEOUT_TIME):
676 with self.assertRaises(arvados.errors.KeepWriteError):
677 kc.put(self.DATA, copies=1, num_retries=0)
679 def test_low_bandwidth_with_server_response_delay_failure(self):
680 kc = self.keepClient()
681 loc = kc.put(self.DATA, copies=1, num_retries=0)
682 self.server.setbandwidth(self.BANDWIDTH_LOW_LIM)
683 self.server.setdelays(response=self.TIMEOUT_TIME)
684 with self.assertTakesGreater(self.TIMEOUT_TIME):
685 with self.assertRaises(arvados.errors.KeepReadError) as e:
686 kc.get(loc, num_retries=0)
687 with self.assertTakesGreater(self.TIMEOUT_TIME):
688 with self.assertRaises(arvados.errors.KeepWriteError):
689 kc.put(self.DATA, copies=1, num_retries=0)
691 def test_low_bandwidth_with_server_mid_delay_failure(self):
692 kc = self.keepClient()
693 loc = kc.put(self.DATA, copies=1, num_retries=0)
694 self.server.setbandwidth(self.BANDWIDTH_LOW_LIM)
695 self.server.setdelays(mid_write=self.TIMEOUT_TIME, mid_read=self.TIMEOUT_TIME)
696 with self.assertTakesGreater(self.TIMEOUT_TIME):
697 with self.assertRaises(arvados.errors.KeepReadError) as e:
698 kc.get(loc, num_retries=0)
699 with self.assertTakesGreater(self.TIMEOUT_TIME):
700 with self.assertRaises(arvados.errors.KeepWriteError):
701 kc.put(self.DATA, copies=1, num_retries=0)
703 def test_timeout_slow_request(self):
704 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
705 self.server.setdelays(request=.2)
706 self._test_connect_timeout_under_200ms(loc)
707 self.server.setdelays(request=2)
708 self._test_response_timeout_under_2s(loc)
710 def test_timeout_slow_response(self):
711 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
712 self.server.setdelays(response=.2)
713 self._test_connect_timeout_under_200ms(loc)
714 self.server.setdelays(response=2)
715 self._test_response_timeout_under_2s(loc)
717 def test_timeout_slow_response_body(self):
718 loc = self.keepClient().put(self.DATA, copies=1, num_retries=0)
719 self.server.setdelays(response_body=.2)
720 self._test_connect_timeout_under_200ms(loc)
721 self.server.setdelays(response_body=2)
722 self._test_response_timeout_under_2s(loc)
724 def _test_connect_timeout_under_200ms(self, loc):
725 # Allow 100ms to connect, then 1s for response. Everything
726 # should work, and everything should take at least 200ms to
728 kc = self.keepClient(timeouts=(.1, 1))
729 with self.assertTakesBetween(.2, .3):
730 kc.put(self.DATA, copies=1, num_retries=0)
731 with self.assertTakesBetween(.2, .3):
732 self.assertEqual(self.DATA, kc.get(loc, num_retries=0))
734 def _test_response_timeout_under_2s(self, loc):
735 # Allow 10s to connect, then 1s for response. Nothing should
736 # work, and everything should take at least 1s to return.
737 kc = self.keepClient(timeouts=(10, 1))
738 with self.assertTakesBetween(1, 1.9):
739 with self.assertRaises(arvados.errors.KeepReadError):
740 kc.get(loc, num_retries=0)
741 with self.assertTakesBetween(1, 1.9):
742 with self.assertRaises(arvados.errors.KeepWriteError):
743 kc.put(self.DATA, copies=1, num_retries=0)
746 class KeepClientGatewayTestCase(unittest.TestCase, tutil.ApiClientMock):
747 def mock_disks_and_gateways(self, disks=3, gateways=1):
749 'uuid': 'zzzzz-bi6l4-gateway{:08d}'.format(i),
750 'owner_uuid': 'zzzzz-tpzed-000000000000000',
751 'service_host': 'gatewayhost{}'.format(i),
752 'service_port': 12345,
753 'service_ssl_flag': True,
754 'service_type': 'gateway:test',
755 } for i in range(gateways)]
756 self.gateway_roots = [
757 "https://{service_host}:{service_port}/".format(**gw)
758 for gw in self.gateways]
759 self.api_client = self.mock_keep_services(
760 count=disks, additional_services=self.gateways)
761 self.keepClient = arvados.KeepClient(api_client=self.api_client)
763 @mock.patch('pycurl.Curl')
764 def test_get_with_gateway_hint_first(self, MockCurl):
765 MockCurl.return_value = tutil.FakeCurl.make(
766 code=200, body='foo', headers={'Content-Length': 3})
767 self.mock_disks_and_gateways()
768 locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@' + self.gateways[0]['uuid']
769 self.assertEqual('foo', self.keepClient.get(locator))
770 self.assertEqual(self.gateway_roots[0]+locator,
771 MockCurl.return_value.getopt(pycurl.URL))
773 @mock.patch('pycurl.Curl')
774 def test_get_with_gateway_hints_in_order(self, MockCurl):
778 tutil.FakeCurl.make(code=404, body='')
779 for _ in range(gateways+disks)
781 MockCurl.side_effect = tutil.queue_with(mocks)
782 self.mock_disks_and_gateways(gateways=gateways, disks=disks)
783 locator = '+'.join(['acbd18db4cc2f85cedef654fccc4a4d8+3'] +
784 ['K@'+gw['uuid'] for gw in self.gateways])
785 with self.assertRaises(arvados.errors.NotFoundError):
786 self.keepClient.get(locator)
787 # Gateways are tried first, in the order given.
788 for i, root in enumerate(self.gateway_roots):
789 self.assertEqual(root+locator,
790 mocks[i].getopt(pycurl.URL))
791 # Disk services are tried next.
792 for i in range(gateways, gateways+disks):
793 self.assertRegexpMatches(
794 mocks[i].getopt(pycurl.URL),
797 @mock.patch('pycurl.Curl')
798 def test_get_with_remote_proxy_hint(self, MockCurl):
799 MockCurl.return_value = tutil.FakeCurl.make(
800 code=200, body='foo', headers={'Content-Length': 3})
801 self.mock_disks_and_gateways()
802 locator = 'acbd18db4cc2f85cedef654fccc4a4d8+3+K@xyzzy'
803 self.assertEqual('foo', self.keepClient.get(locator))
804 self.assertEqual('https://keep.xyzzy.arvadosapi.com/'+locator,
805 MockCurl.return_value.getopt(pycurl.URL))
808 class KeepClientRetryTestMixin(object):
809 # Testing with a local Keep store won't exercise the retry behavior.
810 # Instead, our strategy is:
811 # * Create a client with one proxy specified (pointed at a black
812 # hole), so there's no need to instantiate an API client, and
813 # all HTTP requests come from one place.
814 # * Mock httplib's request method to provide simulated responses.
815 # This lets us test the retry logic extensively without relying on any
816 # supporting servers, and prevents side effects in case something hiccups.
817 # To use this mixin, define DEFAULT_EXPECT, DEFAULT_EXCEPTION, and
820 # Test classes must define TEST_PATCHER to a method that mocks
821 # out appropriate methods in the client.
823 PROXY_ADDR = 'http://[%s]:65535/' % (tutil.TEST_HOST,)
824 TEST_DATA = 'testdata'
825 TEST_LOCATOR = 'ef654c40ab4f1747fc699915d4f70902+8'
828 self.client_kwargs = {'proxy': self.PROXY_ADDR, 'local_store': ''}
830 def new_client(self, **caller_kwargs):
831 kwargs = self.client_kwargs.copy()
832 kwargs.update(caller_kwargs)
833 return arvados.KeepClient(**kwargs)
835 def run_method(self, *args, **kwargs):
836 raise NotImplementedError("test subclasses must define run_method")
838 def check_success(self, expected=None, *args, **kwargs):
840 expected = self.DEFAULT_EXPECT
841 self.assertEqual(expected, self.run_method(*args, **kwargs))
843 def check_exception(self, error_class=None, *args, **kwargs):
844 if error_class is None:
845 error_class = self.DEFAULT_EXCEPTION
846 self.assertRaises(error_class, self.run_method, *args, **kwargs)
848 def test_immediate_success(self):
849 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 200):
852 def test_retry_then_success(self):
853 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
854 self.check_success(num_retries=3)
856 def test_exception_then_success(self):
857 with self.TEST_PATCHER(self.DEFAULT_EXPECT, Exception('mock err'), 200):
858 self.check_success(num_retries=3)
860 def test_no_default_retry(self):
861 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
862 self.check_exception()
864 def test_no_retry_after_permanent_error(self):
865 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 403, 200):
866 self.check_exception(num_retries=3)
868 def test_error_after_retries_exhausted(self):
869 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 500, 200):
870 self.check_exception(num_retries=1)
872 def test_num_retries_instance_fallback(self):
873 self.client_kwargs['num_retries'] = 3
874 with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
879 class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
880 DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_DATA
881 DEFAULT_EXCEPTION = arvados.errors.KeepReadError
882 HINTED_LOCATOR = KeepClientRetryTestMixin.TEST_LOCATOR + '+K@xyzzy'
883 TEST_PATCHER = staticmethod(tutil.mock_keep_responses)
885 def run_method(self, locator=KeepClientRetryTestMixin.TEST_LOCATOR,
887 return self.new_client().get(locator, *args, **kwargs)
889 def test_specific_exception_when_not_found(self):
890 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 200):
891 self.check_exception(arvados.errors.NotFoundError, num_retries=3)
893 def test_general_exception_with_mixed_errors(self):
894 # get should raise a NotFoundError if no server returns the block,
895 # and a high threshold of servers report that it's not found.
896 # This test rigs up 50/50 disagreement between two servers, and
897 # checks that it does not become a NotFoundError.
898 client = self.new_client()
899 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 500):
900 with self.assertRaises(arvados.errors.KeepReadError) as exc_check:
901 client.get(self.HINTED_LOCATOR)
902 self.assertNotIsInstance(
903 exc_check.exception, arvados.errors.NotFoundError,
904 "mixed errors raised NotFoundError")
906 def test_hint_server_can_succeed_without_retries(self):
907 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 404, 200, 500):
908 self.check_success(locator=self.HINTED_LOCATOR)
910 def test_try_next_server_after_timeout(self):
911 with tutil.mock_keep_responses(
912 (socket.timeout("timed out"), 200),
913 (self.DEFAULT_EXPECT, 200)):
914 self.check_success(locator=self.HINTED_LOCATOR)
916 def test_retry_data_with_wrong_checksum(self):
917 with tutil.mock_keep_responses(
919 (self.DEFAULT_EXPECT, 200)):
920 self.check_success(locator=self.HINTED_LOCATOR)
924 class KeepClientRetryPutTestCase(KeepClientRetryTestMixin, unittest.TestCase):
925 DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_LOCATOR
926 DEFAULT_EXCEPTION = arvados.errors.KeepWriteError
927 TEST_PATCHER = staticmethod(tutil.mock_keep_responses)
929 def run_method(self, data=KeepClientRetryTestMixin.TEST_DATA,
930 copies=1, *args, **kwargs):
931 return self.new_client().put(data, copies, *args, **kwargs)
933 def test_do_not_send_multiple_copies_to_same_server(self):
934 with tutil.mock_keep_responses(self.DEFAULT_EXPECT, 200):
935 self.check_exception(copies=2, num_retries=3)
939 class RetryNeedsMultipleServices(unittest.TestCase, tutil.ApiClientMock):
940 # Test put()s that need two distinct servers to succeed, possibly
941 # requiring multiple passes through the retry loop.
944 self.api_client = self.mock_keep_services(count=2)
945 self.keep_client = arvados.KeepClient(api_client=self.api_client)
947 def test_success_after_exception(self):
948 with tutil.mock_keep_responses(
949 'acbd18db4cc2f85cedef654fccc4a4d8+3',
950 Exception('mock err'), 200, 200) as req_mock:
951 self.keep_client.put('foo', num_retries=1, copies=2)
952 self.assertTrue(3, req_mock.call_count)
954 def test_success_after_retryable_error(self):
955 with tutil.mock_keep_responses(
956 'acbd18db4cc2f85cedef654fccc4a4d8+3',
957 500, 200, 200) as req_mock:
958 self.keep_client.put('foo', num_retries=1, copies=2)
959 self.assertTrue(3, req_mock.call_count)
961 def test_fail_after_final_error(self):
962 # First retry loop gets a 200 (can't achieve replication by
963 # storing again on that server) and a 400 (can't retry that
964 # server at all), so we shouldn't try a third request.
965 with tutil.mock_keep_responses(
966 'acbd18db4cc2f85cedef654fccc4a4d8+3',
967 200, 400, 200) as req_mock:
968 with self.assertRaises(arvados.errors.KeepWriteError):
969 self.keep_client.put('foo', num_retries=1, copies=2)
970 self.assertTrue(2, req_mock.call_count)