3ba4a591ee
* rpc: allow also exceptions from the builtins module. On Python 3, builtin exceptions are part of the builtins module. The exceptions module was removed in Python 3. * Fix usage of reraise(): translate_exception() returns an instance which is the second parameter of the reraise() function, not the first parameter. The first parameter is the exception type. * test_rpc: add json_dump_as_bytes() helper to serialize JSON as bytes. * JSONRequestDeserializer: don't compare None to int, it raises a TypeError on Python 3. * Fix Unicode versus bytes issues: HTTP body type is bytes. Encode JSON to UTF-8 for example. Replace StringIO with BytesIO. * Replace filter() with a list-comprehension to get a list on Python 3. * test_client, test_image_cache_client: use bytes for HTTP body. * tox.ini: add glance.tests.unit.common.test_rpc to Python 3. Change-Id: I9960d6a810375474f89c1788b7016a8fbb0770e0
124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
# Copyright 2013 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 os
|
|
|
|
import mock
|
|
|
|
from glance.common import exception
|
|
from glance.image_cache import client
|
|
from glance.tests import utils
|
|
|
|
|
|
class CacheClientTestCase(utils.BaseTestCase):
|
|
def setUp(self):
|
|
super(CacheClientTestCase, self).setUp()
|
|
self.client = client.CacheClient('test_host')
|
|
self.client.do_request = mock.Mock()
|
|
|
|
def test_delete_cached_image(self):
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse()
|
|
self.assertTrue(self.client.delete_cached_image('test_id'))
|
|
self.client.do_request.assert_called_with("DELETE",
|
|
"/cached_images/test_id")
|
|
|
|
def test_get_cached_images(self):
|
|
expected_data = b'{"cached_images": "some_images"}'
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse(
|
|
data=expected_data)
|
|
self.assertEqual("some_images", self.client.get_cached_images())
|
|
self.client.do_request.assert_called_with("GET", "/cached_images")
|
|
|
|
def test_get_queued_images(self):
|
|
expected_data = b'{"queued_images": "some_images"}'
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse(
|
|
data=expected_data)
|
|
self.assertEqual("some_images", self.client.get_queued_images())
|
|
self.client.do_request.assert_called_with("GET", "/queued_images")
|
|
|
|
def test_delete_all_cached_images(self):
|
|
expected_data = b'{"num_deleted": 4}'
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse(
|
|
data=expected_data)
|
|
self.assertEqual(4, self.client.delete_all_cached_images())
|
|
self.client.do_request.assert_called_with("DELETE", "/cached_images")
|
|
|
|
def test_queue_image_for_caching(self):
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse()
|
|
self.assertTrue(self.client.queue_image_for_caching('test_id'))
|
|
self.client.do_request.assert_called_with("PUT",
|
|
"/queued_images/test_id")
|
|
|
|
def test_delete_queued_image(self):
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse()
|
|
self.assertTrue(self.client.delete_queued_image('test_id'))
|
|
self.client.do_request.assert_called_with("DELETE",
|
|
"/queued_images/test_id")
|
|
|
|
def test_delete_all_queued_images(self):
|
|
expected_data = b'{"num_deleted": 4}'
|
|
self.client.do_request.return_value = utils.FakeHTTPResponse(
|
|
data=expected_data)
|
|
self.assertEqual(4, self.client.delete_all_queued_images())
|
|
self.client.do_request.assert_called_with("DELETE", "/queued_images")
|
|
|
|
|
|
class GetClientTestCase(utils.BaseTestCase):
|
|
def setUp(self):
|
|
super(GetClientTestCase, self).setUp()
|
|
self.host = 'test_host'
|
|
self.env = os.environ.copy()
|
|
os.environ.clear()
|
|
|
|
def tearDown(self):
|
|
os.environ = self.env
|
|
super(GetClientTestCase, self).tearDown()
|
|
|
|
def test_get_client_host_only(self):
|
|
expected_creds = {
|
|
'username': None,
|
|
'password': None,
|
|
'tenant': None,
|
|
'auth_url': None,
|
|
'strategy': 'noauth',
|
|
'region': None
|
|
}
|
|
self.assertEqual(expected_creds, client.get_client(self.host).creds)
|
|
|
|
def test_get_client_all_creds(self):
|
|
expected_creds = {
|
|
'username': 'name',
|
|
'password': 'pass',
|
|
'tenant': 'ten',
|
|
'auth_url': 'url',
|
|
'strategy': 'keystone',
|
|
'region': 'reg'
|
|
}
|
|
creds = client.get_client(
|
|
self.host,
|
|
username='name',
|
|
password='pass',
|
|
tenant='ten',
|
|
auth_url='url',
|
|
auth_strategy='strategy',
|
|
region='reg'
|
|
).creds
|
|
self.assertEqual(expected_creds, creds)
|
|
|
|
def test_get_client_client_configuration_error(self):
|
|
self.assertRaises(exception.ClientConfigurationError,
|
|
client.get_client, self.host, username='name',
|
|
password='pass', tenant='ten',
|
|
auth_strategy='keystone', region='reg')
|