1f11840dd8
As a stepping stone to the os-client-config patch, first switch to using keystoneauth, its Session and its argparse registration and plugin loading to sort out any issues with that level of plumbing. The next patch will layer on the ability to use os-client-config for argument processing and client construction. Change-Id: Id681e5eb56b47d06000620f7c92c9b0c5f8d4408
457 lines
20 KiB
Python
457 lines
20 KiB
Python
# Copyright 2012 OpenStack Foundation
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
|
|
import json
|
|
import logging
|
|
|
|
import fixtures
|
|
from keystoneauth1 import adapter
|
|
import mock
|
|
import requests
|
|
|
|
import novaclient.client
|
|
import novaclient.extension
|
|
from novaclient.tests.unit import utils
|
|
import novaclient.v2.client
|
|
|
|
|
|
class ClientConnectionPoolTest(utils.TestCase):
|
|
|
|
@mock.patch("keystoneauth1.session.TCPKeepAliveAdapter")
|
|
def test_get(self, mock_http_adapter):
|
|
mock_http_adapter.side_effect = lambda: mock.Mock()
|
|
pool = novaclient.client._ClientConnectionPool()
|
|
self.assertEqual(pool.get("abc"), pool.get("abc"))
|
|
self.assertNotEqual(pool.get("abc"), pool.get("def"))
|
|
|
|
|
|
class ClientTest(utils.TestCase):
|
|
|
|
def test_client_with_timeout(self):
|
|
instance = novaclient.client.HTTPClient(user='user',
|
|
password='password',
|
|
projectid='project',
|
|
timeout=2,
|
|
auth_url="http://www.blah.com")
|
|
self.assertEqual(2, instance.timeout)
|
|
mock_request = mock.Mock()
|
|
mock_request.return_value = requests.Response()
|
|
mock_request.return_value.status_code = 200
|
|
mock_request.return_value.headers = {
|
|
'x-server-management-url': 'blah.com',
|
|
'x-auth-token': 'blah',
|
|
}
|
|
with mock.patch('requests.request', mock_request):
|
|
instance.authenticate()
|
|
requests.request.assert_called_with(
|
|
mock.ANY, mock.ANY, timeout=2, headers=mock.ANY,
|
|
verify=mock.ANY)
|
|
|
|
def test_client_reauth(self):
|
|
instance = novaclient.client.HTTPClient(user='user',
|
|
password='password',
|
|
projectid='project',
|
|
timeout=2,
|
|
auth_url="http://www.blah.com")
|
|
instance.auth_token = 'foobar'
|
|
instance.management_url = 'http://example.com'
|
|
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
|
instance.version = 'v2.0'
|
|
mock_request = mock.Mock()
|
|
mock_request.side_effect = novaclient.exceptions.Unauthorized(401)
|
|
with mock.patch('requests.request', mock_request):
|
|
try:
|
|
instance.get('/servers/detail')
|
|
except Exception:
|
|
pass
|
|
get_headers = {'X-Auth-Project-Id': 'project',
|
|
'X-Auth-Token': 'foobar',
|
|
'User-Agent': 'python-novaclient',
|
|
'Accept': 'application/json'}
|
|
reauth_headers = {'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'User-Agent': 'python-novaclient'}
|
|
data = {
|
|
"auth": {
|
|
"tenantName": "project",
|
|
"passwordCredentials": {
|
|
"username": "user",
|
|
"password": "password"
|
|
}
|
|
}
|
|
}
|
|
|
|
expected = [mock.call('GET',
|
|
'http://example.com/servers/detail',
|
|
timeout=mock.ANY,
|
|
headers=get_headers,
|
|
verify=mock.ANY),
|
|
mock.call('POST', 'http://www.blah.com/tokens',
|
|
timeout=mock.ANY,
|
|
headers=reauth_headers,
|
|
allow_redirects=mock.ANY,
|
|
data=mock.ANY,
|
|
verify=mock.ANY)]
|
|
self.assertEqual(expected, mock_request.call_args_list)
|
|
token_post_call = mock_request.call_args_list[1]
|
|
self.assertEqual(data, json.loads(token_post_call[1]['data']))
|
|
|
|
@mock.patch.object(novaclient.client.HTTPClient, 'request',
|
|
return_value=(200, "{'versions':[]}"))
|
|
def _check_version_url(self, management_url, version_url, mock_request):
|
|
projectid = '25e469aa1848471b875e68cde6531bc5'
|
|
instance = novaclient.client.HTTPClient(user='user',
|
|
password='password',
|
|
projectid=projectid,
|
|
auth_url="http://www.blah.com")
|
|
instance.auth_token = 'foobar'
|
|
instance.management_url = management_url % projectid
|
|
mock_get_service_url = mock.Mock(return_value=instance.management_url)
|
|
instance.get_service_url = mock_get_service_url
|
|
instance.version = 'v2.0'
|
|
|
|
# If passing None as the part of url, a client accesses the url which
|
|
# doesn't include "v2/<projectid>" for getting API version info.
|
|
instance.get(None)
|
|
mock_request.assert_called_once_with(version_url, 'GET',
|
|
headers=mock.ANY)
|
|
mock_request.reset_mock()
|
|
|
|
# Otherwise, a client accesses the url which includes "v2/<projectid>".
|
|
instance.get('servers')
|
|
url = instance.management_url + 'servers'
|
|
mock_request.assert_called_once_with(url, 'GET', headers=mock.ANY)
|
|
|
|
def test_client_version_url(self):
|
|
self._check_version_url('http://foo.com/v2/%s', 'http://foo.com/')
|
|
self._check_version_url('http://foo.com/v2.1/%s', 'http://foo.com/')
|
|
self._check_version_url('http://foo.com/v3.785/%s', 'http://foo.com/')
|
|
|
|
def test_client_version_url_with_project_name(self):
|
|
self._check_version_url('http://foo.com/nova/v2/%s',
|
|
'http://foo.com/nova/')
|
|
self._check_version_url('http://foo.com/nova/v2.1/%s',
|
|
'http://foo.com/nova/')
|
|
self._check_version_url('http://foo.com/nova/v3.785/%s',
|
|
'http://foo.com/nova/')
|
|
|
|
def test_get_client_class_v2(self):
|
|
output = novaclient.client.get_client_class('2')
|
|
self.assertEqual(output, novaclient.v2.client.Client)
|
|
|
|
def test_get_client_class_v2_int(self):
|
|
output = novaclient.client.get_client_class(2)
|
|
self.assertEqual(output, novaclient.v2.client.Client)
|
|
|
|
def test_get_client_class_v1_1(self):
|
|
output = novaclient.client.get_client_class('1.1')
|
|
self.assertEqual(output, novaclient.v2.client.Client)
|
|
|
|
def test_get_client_class_unknown(self):
|
|
self.assertRaises(novaclient.exceptions.UnsupportedVersion,
|
|
novaclient.client.get_client_class, '0')
|
|
|
|
def test_get_client_class_latest(self):
|
|
self.assertRaises(novaclient.exceptions.UnsupportedVersion,
|
|
novaclient.client.get_client_class, 'latest')
|
|
self.assertRaises(novaclient.exceptions.UnsupportedVersion,
|
|
novaclient.client.get_client_class, '2.latest')
|
|
|
|
def test_client_with_os_cache_enabled(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", os_cache=True)
|
|
self.assertTrue(cs.os_cache)
|
|
self.assertTrue(cs.client.os_cache)
|
|
|
|
def test_client_with_os_cache_disabled(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", os_cache=False)
|
|
self.assertFalse(cs.os_cache)
|
|
self.assertFalse(cs.client.os_cache)
|
|
|
|
def test_client_with_no_cache_enabled(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", no_cache=True)
|
|
self.assertFalse(cs.os_cache)
|
|
self.assertFalse(cs.client.os_cache)
|
|
|
|
def test_client_with_no_cache_disabled(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", no_cache=False)
|
|
self.assertTrue(cs.os_cache)
|
|
self.assertTrue(cs.client.os_cache)
|
|
|
|
def test_client_set_management_url_v1_1(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2")
|
|
cs.set_management_url("blabla")
|
|
self.assertEqual("blabla", cs.client.management_url)
|
|
|
|
def test_client_get_reset_timings_v1_1(self):
|
|
cs = novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2")
|
|
self.assertEqual(0, len(cs.get_timings()))
|
|
cs.client.times.append("somevalue")
|
|
self.assertEqual(1, len(cs.get_timings()))
|
|
self.assertEqual("somevalue", cs.get_timings()[0])
|
|
|
|
cs.reset_timings()
|
|
self.assertEqual(0, len(cs.get_timings()))
|
|
|
|
@mock.patch('novaclient.client.HTTPClient')
|
|
def test_contextmanager_v1_1(self, mock_http_client):
|
|
fake_client = mock.Mock()
|
|
mock_http_client.return_value = fake_client
|
|
with novaclient.v2.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2"):
|
|
pass
|
|
self.assertTrue(fake_client.open_session.called)
|
|
self.assertTrue(fake_client.close_session.called)
|
|
|
|
def test_get_password_simple(self):
|
|
cs = novaclient.client.HTTPClient("user", "password", "", "")
|
|
cs.password_func = mock.Mock()
|
|
self.assertEqual("password", cs._get_password())
|
|
self.assertFalse(cs.password_func.called)
|
|
|
|
def test_get_password_none(self):
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
self.assertIsNone(cs._get_password())
|
|
|
|
def test_get_password_func(self):
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
cs.password_func = mock.Mock(return_value="password")
|
|
self.assertEqual("password", cs._get_password())
|
|
cs.password_func.assert_called_once_with()
|
|
|
|
cs.password_func = mock.Mock()
|
|
self.assertEqual("password", cs._get_password())
|
|
self.assertFalse(cs.password_func.called)
|
|
|
|
def test_auth_url_rstrip_slash(self):
|
|
cs = novaclient.client.HTTPClient("user", "password", "project_id",
|
|
auth_url="foo/v2/")
|
|
self.assertEqual("foo/v2", cs.auth_url)
|
|
|
|
def test_token_and_bypass_url(self):
|
|
cs = novaclient.client.HTTPClient(None, None, None,
|
|
auth_token="12345",
|
|
bypass_url="compute/v100/")
|
|
self.assertIsNone(cs.auth_url)
|
|
self.assertEqual("12345", cs.auth_token)
|
|
self.assertEqual("compute/v100", cs.bypass_url)
|
|
self.assertEqual("compute/v100", cs.management_url)
|
|
|
|
def test_service_url_lookup(self):
|
|
service_type = 'compute'
|
|
cs = novaclient.client.HTTPClient(None, None, None,
|
|
auth_url='foo/v2',
|
|
service_type=service_type)
|
|
|
|
@mock.patch.object(cs, 'get_service_url', return_value='compute/v5')
|
|
@mock.patch.object(cs, 'request', return_value=(200, '{}'))
|
|
@mock.patch.object(cs, 'authenticate')
|
|
def do_test(mock_auth, mock_request, mock_get):
|
|
|
|
def set_service_catalog():
|
|
cs.service_catalog = 'catalog'
|
|
|
|
mock_auth.side_effect = set_service_catalog
|
|
cs.get('/servers')
|
|
mock_get.assert_called_once_with(service_type)
|
|
mock_request.assert_called_once_with('compute/v5/servers',
|
|
'GET', headers=mock.ANY)
|
|
mock_auth.assert_called_once_with()
|
|
|
|
do_test()
|
|
|
|
def test_bypass_url_no_service_url_lookup(self):
|
|
bypass_url = 'compute/v100'
|
|
cs = novaclient.client.HTTPClient(None, None, None,
|
|
auth_url='foo/v2',
|
|
bypass_url=bypass_url)
|
|
|
|
@mock.patch.object(cs, 'get_service_url')
|
|
@mock.patch.object(cs, 'request', return_value=(200, '{}'))
|
|
def do_test(mock_request, mock_get):
|
|
cs.get('/servers')
|
|
self.assertFalse(mock_get.called)
|
|
mock_request.assert_called_once_with(bypass_url + '/servers',
|
|
'GET', headers=mock.ANY)
|
|
|
|
do_test()
|
|
|
|
@mock.patch("novaclient.client.requests.Session")
|
|
def test_session(self, mock_session):
|
|
fake_session = mock.Mock()
|
|
mock_session.return_value = fake_session
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
cs.open_session()
|
|
self.assertEqual(cs._session, fake_session)
|
|
cs.close_session()
|
|
self.assertIsNone(cs._session)
|
|
|
|
def test_session_connection_pool(self):
|
|
cs = novaclient.client.HTTPClient("user", None, "",
|
|
"", connection_pool=True)
|
|
cs.open_session()
|
|
self.assertIsNone(cs._session)
|
|
cs.close_session()
|
|
self.assertIsNone(cs._session)
|
|
|
|
def test_get_session(self):
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
self.assertIsNone(cs._get_session("http://nooooooooo.com"))
|
|
|
|
@mock.patch("novaclient.client.requests.Session")
|
|
def test_get_session_open_session(self, mock_session):
|
|
fake_session = mock.Mock()
|
|
mock_session.return_value = fake_session
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
cs.open_session()
|
|
self.assertEqual(fake_session, cs._get_session("http://example.com"))
|
|
|
|
@mock.patch("novaclient.client.requests.Session")
|
|
@mock.patch("novaclient.client._ClientConnectionPool")
|
|
def test_get_session_connection_pool(self, mock_pool, mock_session):
|
|
service_url = "http://example.com"
|
|
|
|
pool = mock.MagicMock()
|
|
pool.get.return_value = "http_adapter"
|
|
mock_pool.return_value = pool
|
|
cs = novaclient.client.HTTPClient("user", None, "",
|
|
"", connection_pool=True)
|
|
cs._current_url = "http://another.com"
|
|
|
|
session = cs._get_session(service_url)
|
|
self.assertEqual(session, mock_session.return_value)
|
|
pool.get.assert_called_once_with(service_url)
|
|
mock_session().mount.assert_called_once_with(service_url,
|
|
'http_adapter')
|
|
|
|
def test_init_without_connection_pool(self):
|
|
cs = novaclient.client.HTTPClient("user", None, "", "")
|
|
self.assertIsNone(cs._connection_pool)
|
|
|
|
@mock.patch("novaclient.client._ClientConnectionPool")
|
|
def test_init_with_proper_connection_pool(self, mock_pool):
|
|
fake_pool = mock.Mock()
|
|
mock_pool.return_value = fake_pool
|
|
cs = novaclient.client.HTTPClient("user", None, "",
|
|
connection_pool=True)
|
|
self.assertEqual(cs._connection_pool, fake_pool)
|
|
|
|
def test_log_req(self):
|
|
self.logger = self.useFixture(
|
|
fixtures.FakeLogger(
|
|
format="%(message)s",
|
|
level=logging.DEBUG,
|
|
nuke_handlers=True
|
|
)
|
|
)
|
|
cs = novaclient.client.HTTPClient("user", None, "",
|
|
connection_pool=True)
|
|
cs.http_log_debug = True
|
|
cs.http_log_req('GET', '/foo', {'headers': {}})
|
|
cs.http_log_req('GET', '/foo', {'headers':
|
|
{'X-Auth-Token': None}})
|
|
cs.http_log_req('GET', '/foo', {'headers':
|
|
{'X-Auth-Token': 'totally_bogus'}})
|
|
cs.http_log_req('GET', '/foo', {'headers':
|
|
{'X-Foo': 'bar',
|
|
'X-Auth-Token': 'totally_bogus'}})
|
|
cs.http_log_req('GET', '/foo', {'headers': {},
|
|
'data':
|
|
'{"auth": {"passwordCredentials": '
|
|
'{"password": "zhaoqin"}}}'})
|
|
|
|
output = self.logger.output.split('\n')
|
|
|
|
self.assertIn("REQ: curl -g -i '/foo' -X GET", output)
|
|
self.assertIn(
|
|
"REQ: curl -g -i '/foo' -X GET -H "
|
|
'"X-Auth-Token: None"',
|
|
output)
|
|
self.assertIn(
|
|
"REQ: curl -g -i '/foo' -X GET -H "
|
|
'"X-Auth-Token: {SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d"',
|
|
output)
|
|
self.assertIn(
|
|
"REQ: curl -g -i '/foo' -X GET -H "
|
|
'"X-Auth-Token: {SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d"'
|
|
' -H "X-Foo: bar"',
|
|
output)
|
|
self.assertIn(
|
|
"REQ: curl -g -i '/foo' -X GET -d "
|
|
'\'{"auth": {"passwordCredentials": {"password":'
|
|
' "{SHA1}4fc49c6a671ce889078ff6b250f7066cf6d2ada2"}}}\'',
|
|
output)
|
|
|
|
def test_log_resp(self):
|
|
self.logger = self.useFixture(
|
|
fixtures.FakeLogger(
|
|
format="%(message)s",
|
|
level=logging.DEBUG,
|
|
nuke_handlers=True
|
|
)
|
|
)
|
|
|
|
cs = novaclient.client.HTTPClient("user", None, "",
|
|
connection_pool=True)
|
|
cs.http_log_debug = True
|
|
text = ('{"access": {"token": {"id": "zhaoqin"}}}')
|
|
resp = utils.TestResponse({'status_code': 200, 'headers': {},
|
|
'text': text})
|
|
|
|
cs.http_log_resp(resp)
|
|
output = self.logger.output.split('\n')
|
|
|
|
self.assertIn('RESP: [200] {}', output)
|
|
self.assertIn('RESP BODY: {"access": {"token": {"id":'
|
|
' "{SHA1}4fc49c6a671ce889078ff6b250f7066cf6d2ada2"}}}',
|
|
output)
|
|
|
|
@mock.patch.object(novaclient.client.HTTPClient, 'request')
|
|
def test_timings(self, m_request):
|
|
m_request.return_value = (None, None)
|
|
|
|
client = novaclient.client.HTTPClient(user='zqfan', password='')
|
|
client._time_request("http://no.where", 'GET')
|
|
self.assertEqual(0, len(client.times))
|
|
|
|
client = novaclient.client.HTTPClient(user='zqfan', password='',
|
|
timings=True)
|
|
client._time_request("http://no.where", 'GET')
|
|
self.assertEqual(1, len(client.times))
|
|
self.assertEqual('GET http://no.where', client.times[0][0])
|
|
|
|
|
|
class SessionClientTest(utils.TestCase):
|
|
|
|
@mock.patch.object(adapter.LegacyJsonAdapter, 'request')
|
|
def test_timings(self, m_request):
|
|
m_request.return_value = (mock.MagicMock(status_code=200), None)
|
|
|
|
client = novaclient.client.SessionClient(session=mock.MagicMock())
|
|
client.request("http://no.where", 'GET')
|
|
self.assertEqual(0, len(client.times))
|
|
|
|
client = novaclient.client.SessionClient(session=mock.MagicMock(),
|
|
timings=True)
|
|
client.request("http://no.where", 'GET')
|
|
self.assertEqual(1, len(client.times))
|
|
self.assertEqual('GET http://no.where', client.times[0][0])
|