421 lines
18 KiB
Python
421 lines
18 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
|
|
import mock
|
|
import requests
|
|
|
|
import novaclient.client
|
|
import novaclient.extension
|
|
import novaclient.tests.fakes as fakes
|
|
from novaclient.tests import utils
|
|
import novaclient.v1_1.client
|
|
import novaclient.v3.client
|
|
|
|
|
|
class ClientConnectionPoolTest(utils.TestCase):
|
|
|
|
@mock.patch("novaclient.client.adapters.HTTPAdapter")
|
|
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.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=json.dumps(data),
|
|
verify=mock.ANY)]
|
|
self.assertEqual(expected, mock_request.call_args_list)
|
|
|
|
@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
|
|
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/')
|
|
|
|
def test_client_version_url_with_project_name(self):
|
|
self._check_version_url('http://foo.com/nova/v2/%s',
|
|
'http://foo.com/nova/')
|
|
|
|
def test_get_client_class_v3(self):
|
|
output = novaclient.client.get_client_class('3')
|
|
self.assertEqual(output, novaclient.v3.client.Client)
|
|
|
|
def test_get_client_class_v2(self):
|
|
output = novaclient.client.get_client_class('2')
|
|
self.assertEqual(output, novaclient.v1_1.client.Client)
|
|
|
|
def test_get_client_class_v2_int(self):
|
|
output = novaclient.client.get_client_class(2)
|
|
self.assertEqual(output, novaclient.v1_1.client.Client)
|
|
|
|
def test_get_client_class_v1_1(self):
|
|
output = novaclient.client.get_client_class('1.1')
|
|
self.assertEqual(output, novaclient.v1_1.client.Client)
|
|
|
|
def test_get_client_class_unknown(self):
|
|
self.assertRaises(novaclient.exceptions.UnsupportedVersion,
|
|
novaclient.client.get_client_class, '0')
|
|
|
|
def test_client_with_os_cache_enabled(self):
|
|
cs = novaclient.v1_1.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", os_cache=True)
|
|
self.assertEqual(True, cs.os_cache)
|
|
self.assertEqual(True, cs.client.os_cache)
|
|
|
|
def test_client_with_os_cache_disabled(self):
|
|
cs = novaclient.v1_1.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", os_cache=False)
|
|
self.assertEqual(False, cs.os_cache)
|
|
self.assertEqual(False, cs.client.os_cache)
|
|
|
|
def test_client_with_no_cache_enabled(self):
|
|
cs = novaclient.v1_1.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", no_cache=True)
|
|
self.assertEqual(False, cs.os_cache)
|
|
self.assertEqual(False, cs.client.os_cache)
|
|
|
|
def test_client_with_no_cache_disabled(self):
|
|
cs = novaclient.v1_1.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2", no_cache=False)
|
|
self.assertEqual(True, cs.os_cache)
|
|
self.assertEqual(True, cs.client.os_cache)
|
|
|
|
def test_client_set_management_url_v1_1(self):
|
|
cs = novaclient.v1_1.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.v1_1.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()))
|
|
|
|
def test_client_set_management_url_v3(self):
|
|
cs = novaclient.v3.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_v3(self):
|
|
cs = novaclient.v3.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2")
|
|
self.assertEqual(0, len(cs.get_timings()))
|
|
cs.client.times.append("somevalue")
|
|
self.assertEqual(["somevalue"], cs.get_timings())
|
|
|
|
cs.reset_timings()
|
|
self.assertEqual(0, len(cs.get_timings()))
|
|
|
|
def test_clent_extensions_v3(self):
|
|
fake_attribute_name1 = "FakeAttribute1"
|
|
fake_attribute_name2 = "FakeAttribute2"
|
|
extensions = [
|
|
novaclient.extension.Extension(fake_attribute_name1,
|
|
fakes),
|
|
novaclient.extension.Extension(fake_attribute_name2,
|
|
utils),
|
|
]
|
|
|
|
cs = novaclient.v3.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2",
|
|
extensions=extensions)
|
|
self.assertIsInstance(getattr(cs, fake_attribute_name1, None),
|
|
fakes.FakeManager)
|
|
self.assertFalse(hasattr(cs, fake_attribute_name2))
|
|
|
|
@mock.patch.object(novaclient.client.HTTPClient, 'authenticate')
|
|
def test_authenticate_call_v3(self, mock_authenticate):
|
|
cs = novaclient.v3.client.Client("user", "password", "project_id",
|
|
auth_url="foo/v2")
|
|
cs.authenticate()
|
|
self.assertTrue(mock_authenticate.called)
|
|
|
|
@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.v1_1.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)
|
|
|
|
@mock.patch('novaclient.client.HTTPClient')
|
|
def test_contextmanager_v3(self, mock_http_client):
|
|
fake_client = mock.Mock()
|
|
mock_http_client.return_value = fake_client
|
|
with novaclient.v3.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)
|
|
|
|
@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': '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 -i '/foo' -X GET", output)
|
|
self.assertIn(
|
|
"REQ: curl -i '/foo' -X GET -H "
|
|
'"X-Auth-Token: {SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d"',
|
|
output)
|
|
self.assertIn(
|
|
"REQ: curl -i '/foo' -X GET -H "
|
|
'"X-Auth-Token: {SHA1}b42162b6ffdbd7c3c37b7c95b7ba9f51dda0236d"'
|
|
' -H "X-Foo: bar"',
|
|
output)
|
|
self.assertIn(
|
|
"REQ: curl -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)
|