self.assertEqual('100::1', service.hostname)
self.assertEqual(10, service.port)
+ def test_insecure_disables_tls_verify(self):
+ api_client = self.mock_keep_services(count=1)
+ force_timeout = socket.timeout("timed out")
+
+ api_client.insecure = True
+ with tutil.mock_keep_responses(b'foo', 200) as mock:
+ keep_client = arvados.KeepClient(api_client=api_client)
+ keep_client.get('acbd18db4cc2f85cedef654fccc4a4d8+3')
+ self.assertEqual(
+ mock.responses[0].getopt(pycurl.SSL_VERIFYPEER),
+ 0)
+
+ api_client.insecure = False
+ with tutil.mock_keep_responses(b'foo', 200) as mock:
+ keep_client = arvados.KeepClient(api_client=api_client)
+ keep_client.get('acbd18db4cc2f85cedef654fccc4a4d8+3')
+ # getopt()==None here means we didn't change the
+ # default. If we were using real pycurl instead of a mock,
+ # it would return the default value 1.
+ self.assertEqual(
+ mock.responses[0].getopt(pycurl.SSL_VERIFYPEER),
+ None)
+
+ def test_refresh_signature(self):
+ blk_digest = '6f5902ac237024bdd0c176cb93063dc4+11'
+ blk_sig = 'da39a3ee5e6b4b0d3255bfef95601890afd80709@53bed294'
+ local_loc = blk_digest+'+A'+blk_sig
+ remote_loc = blk_digest+'+R'+blk_sig
+ api_client = self.mock_keep_services(count=1)
+ headers = {'X-Keep-Locator':local_loc}
+ with tutil.mock_keep_responses('', 200, **headers):
+ # Check that the translated locator gets returned
+ keep_client = arvados.KeepClient(api_client=api_client)
+ self.assertEqual(local_loc, keep_client.refresh_signature(remote_loc))
+ # Check that refresh_signature() uses the correct method and headers
+ keep_client._get_or_head = mock.MagicMock()
+ keep_client.refresh_signature(remote_loc)
+ args, kwargs = keep_client._get_or_head.call_args_list[0]
+ self.assertIn(remote_loc, args)
+ self.assertEqual("HEAD", kwargs['method'])
+ self.assertIn('X-Keep-Signature', kwargs['headers'])
+
# test_*_timeout verify that KeepClient instructs pycurl to use
# the appropriate connection and read timeouts. They don't care
# whether pycurl actually exhibits the expected timeout behavior
def __getattr__(self, r):
if r == "api_token":
return "abc"
+ elif r == "insecure":
+ return False
else:
raise arvados.errors.KeepReadError()
keep_client = arvados.KeepClient(api_client=ApiMock(),