if not proxy_uris[i].endswith('/'):
proxy_uris[i] += '/'
# URL validation
- url = urlparse(proxy_uris[i])
+ url = urlparse.urlparse(proxy_uris[i])
if not (url.scheme and url.netloc):
raise arvados.errors.ArgumentError("Invalid proxy URI: {}".format(proxy_uris[i]))
self.api_token = api_token
'service_type': 'proxy',
'service_ssl_flag': False,
}}).execute()
- os.environ["ARVADOS_KEEP_PROXY"] = "http://localhost:{}".format(port)
+ os.environ["ARVADOS_KEEP_SERVICES"] = "http://localhost:{}".format(port)
_setport('keepproxy', port)
_wait_until_port_listens(port)
cls._orig_environ = os.environ.copy()
cls._orig_config = arvados.config.settings().copy()
cls._cleanup_funcs = []
- os.environ.pop('ARVADOS_KEEP_PROXY', None)
+ os.environ.pop('ARVADOS_KEEP_SERVICES', None)
os.environ.pop('ARVADOS_EXTERNAL_CLIENT', None)
for server_kwargs, start_func, stop_func in (
(cls.MAIN_SERVER, run, reset),
super(KeepProxyTestCase, self).tearDown()
def test_KeepProxyTest1(self):
- # Will use ARVADOS_KEEP_PROXY environment variable that is set by
- # setUpClass().
+ # Will use ARVADOS_KEEP_SERVICES environment variable that
+ # is set by setUpClass().
keep_client = arvados.KeepClient(api_client=self.api_client,
local_store='')
baz_locator = keep_client.put('baz')
def test_KeepProxyTestMultipleURIs(self):
# Test using ARVADOS_KEEP_SERVICES env var overriding any
# existing proxy setting and setting multiple proxies
- arvados.config.settings()['ARVADOS_KEEP_SERVICES'] = 'foo bar baz'
+ arvados.config.settings()['ARVADOS_KEEP_SERVICES'] = 'http://10.0.0.1 https://foo.example.org:1234/'
keep_client = arvados.KeepClient(api_client=self.api_client,
local_store='')
uris = [x['_service_root'] for x in keep_client._keep_services]
- self.assertEqual(uris, ['foo/', 'bar/', 'baz/'])
+ self.assertEqual(uris, ['http://10.0.0.1/',
+ 'https://foo.example.org:1234/'])
+
+ def test_KeepProxyTestInvalidURI(self):
+ arvados.config.settings()['ARVADOS_KEEP_SERVICES'] = 'bad.uri.org'
+ with self.assertRaises(arvados.errors.ArgumentError):
+ keep_client = arvados.KeepClient(api_client=self.api_client,
+ local_store='')
+
class KeepClientServiceTestCase(unittest.TestCase, tutil.ApiClientMock):
def get_service_roots(self, api_client):