Add client resource base class
Change-Id: I0e1179fbf1804087c753e529ca67a6e503b6d1d4
This commit is contained in:
Rui Chen
parent
ba4e567fb3
commit
edc0ccd973
356
nimbleclient/common/base.py
Normal file
356
nimbleclient/common/base.py
Normal file
@ -0,0 +1,356 @@
|
||||
# Copyright 2016 Huawei, Inc. 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.
|
||||
#
|
||||
|
||||
"""
|
||||
Base utilities to build API operation managers and objects on top of.
|
||||
"""
|
||||
|
||||
import abc
|
||||
import copy
|
||||
|
||||
from requests import Response
|
||||
import six
|
||||
|
||||
from nimbleclient.common import exceptions
|
||||
|
||||
|
||||
# Python 2.4 compat
|
||||
try:
|
||||
all
|
||||
except NameError:
|
||||
def all(iterable):
|
||||
return True not in (not x for x in iterable)
|
||||
|
||||
|
||||
def getid(obj):
|
||||
"""Get obj's id or object itself if no id
|
||||
|
||||
Abstracts the common pattern of allowing both an object or
|
||||
an object's ID (UUID) as a parameter when dealing with relationships.
|
||||
"""
|
||||
try:
|
||||
return obj.id
|
||||
except AttributeError:
|
||||
return obj
|
||||
|
||||
|
||||
class Manager(object):
|
||||
"""Interacts with type of API
|
||||
|
||||
Managers interact with a particular type of API (instances, types, etc.)
|
||||
and provide CRUD operations for them.
|
||||
"""
|
||||
resource_class = None
|
||||
|
||||
def __init__(self, api):
|
||||
self.api = api
|
||||
|
||||
def _list(self, url, response_key=None, obj_class=None,
|
||||
data=None, headers=None):
|
||||
|
||||
if headers is None:
|
||||
headers = {}
|
||||
resp, body = self.api.get(url, headers=headers)
|
||||
|
||||
if obj_class is None:
|
||||
obj_class = self.resource_class
|
||||
|
||||
if response_key:
|
||||
if response_key not in body:
|
||||
body[response_key] = []
|
||||
data = body[response_key]
|
||||
else:
|
||||
data = body
|
||||
items = [obj_class(self, res, loaded=True) for res in data if res]
|
||||
|
||||
return ListWithMeta(items, resp)
|
||||
|
||||
def _delete(self, url, headers=None):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
resp, body = self.api.delete(url, headers=headers)
|
||||
|
||||
return self.convert_into_with_meta(body, resp)
|
||||
|
||||
def _update(self, url, data, response_key=None, headers=None):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
resp, body = self.api.patch(url, data=data, headers=headers)
|
||||
# PUT requests may not return a body
|
||||
if body:
|
||||
if response_key:
|
||||
return self.resource_class(self, body[response_key], resp=resp)
|
||||
return self.resource_class(self, body, resp=resp)
|
||||
else:
|
||||
return StrWithMeta(body, resp)
|
||||
|
||||
def _create(self, url, data=None, response_key=None, return_raw=False,
|
||||
headers=None):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
if data:
|
||||
resp, body = self.api.post(url, data=data, headers=headers)
|
||||
else:
|
||||
resp, body = self.api.post(url, headers=headers)
|
||||
if return_raw:
|
||||
if response_key:
|
||||
body = body[response_key]
|
||||
return self.convert_into_with_meta(body, resp)
|
||||
|
||||
if response_key:
|
||||
return self.resource_class(self, body[response_key], resp=resp)
|
||||
return self.resource_class(self, body, resp=resp)
|
||||
|
||||
def _get(self, url, response_key=None, return_raw=False, headers=None):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
resp, body = self.api.get(url, headers=headers)
|
||||
if return_raw:
|
||||
if response_key:
|
||||
body = body[response_key]
|
||||
return self.convert_into_with_meta(body, resp)
|
||||
|
||||
if response_key:
|
||||
return self.resource_class(self, body[response_key], loaded=True,
|
||||
resp=resp)
|
||||
return self.resource_class(self, body, loaded=True, resp=resp)
|
||||
|
||||
def convert_into_with_meta(self, item, resp):
|
||||
if isinstance(item, six.string_types):
|
||||
if six.PY2 and isinstance(item, six.text_type):
|
||||
return UnicodeWithMeta(item, resp)
|
||||
else:
|
||||
return StrWithMeta(item, resp)
|
||||
elif isinstance(item, six.binary_type):
|
||||
return BytesWithMeta(item, resp)
|
||||
elif isinstance(item, list):
|
||||
return ListWithMeta(item, resp)
|
||||
elif isinstance(item, tuple):
|
||||
return TupleWithMeta(item, resp)
|
||||
elif item is None:
|
||||
return TupleWithMeta((), resp)
|
||||
else:
|
||||
return DictWithMeta(item, resp)
|
||||
|
||||
|
||||
@six.add_metaclass(abc.ABCMeta)
|
||||
class ManagerWithFind(Manager):
|
||||
"""Manager with additional `find()`/`findall()` methods."""
|
||||
|
||||
@abc.abstractmethod
|
||||
def list(self):
|
||||
pass
|
||||
|
||||
def find(self, **kwargs):
|
||||
"""Find a single item with attributes matching ``**kwargs``.
|
||||
|
||||
This isn't very efficient: it loads the entire list then filters on
|
||||
the Python side.
|
||||
"""
|
||||
matches = self.findall(**kwargs)
|
||||
num = len(matches)
|
||||
|
||||
if num == 0:
|
||||
msg = "No %s matching %s." % (self.resource_class.__name__, kwargs)
|
||||
raise exceptions.NotFound(msg)
|
||||
elif num > 1:
|
||||
raise exceptions.NoUniqueMatch
|
||||
else:
|
||||
return self.get(matches[0].id)
|
||||
|
||||
def findall(self, **kwargs):
|
||||
"""Find all items with attributes matching ``**kwargs``.
|
||||
|
||||
This isn't very efficient: it loads the entire list then filters on
|
||||
the Python side.
|
||||
"""
|
||||
found = []
|
||||
searches = kwargs.items()
|
||||
|
||||
for obj in self.list():
|
||||
try:
|
||||
if all(getattr(obj, attr) == value
|
||||
for (attr, value) in searches):
|
||||
found.append(obj)
|
||||
except AttributeError:
|
||||
continue
|
||||
|
||||
return found
|
||||
|
||||
|
||||
class RequestIdMixin(object):
|
||||
"""Wrapper class to expose x-openstack-request-id to the caller."""
|
||||
def request_ids_setup(self):
|
||||
self.x_openstack_request_ids = []
|
||||
|
||||
@property
|
||||
def request_ids(self):
|
||||
return self.x_openstack_request_ids
|
||||
|
||||
def append_request_ids(self, resp):
|
||||
"""Add request_ids as an attribute to the object
|
||||
|
||||
:param resp: Response object or list of Response objects
|
||||
"""
|
||||
if isinstance(resp, list):
|
||||
# Add list of request_ids if response is of type list.
|
||||
for resp_obj in resp:
|
||||
self._append_request_id(resp_obj)
|
||||
elif resp is not None:
|
||||
# Add request_ids if response contains single object.
|
||||
self._append_request_id(resp)
|
||||
|
||||
def _append_request_id(self, resp):
|
||||
if isinstance(resp, Response):
|
||||
# Extract 'x-openstack-request-id' from headers if
|
||||
# response is a Response object.
|
||||
request_id = (resp.headers.get('x-openstack-request-id') or
|
||||
resp.headers.get('x-compute-request-id'))
|
||||
else:
|
||||
# If resp is of type string or None.
|
||||
request_id = resp
|
||||
if request_id not in self.x_openstack_request_ids:
|
||||
self.x_openstack_request_ids.append(request_id)
|
||||
|
||||
|
||||
class Resource(RequestIdMixin):
|
||||
"""Represents an instance of an object
|
||||
|
||||
A resource represents a particular instance of an object (instance, type,
|
||||
etc). This is pretty much just a bag for attributes.
|
||||
|
||||
:param manager: BaseManager object
|
||||
:param info: dictionary representing resource attributes
|
||||
:param loaded: prevent lazy-loading if set to True
|
||||
:param resp: Response or list of Response objects
|
||||
"""
|
||||
|
||||
def __init__(self, manager, info, loaded=False, resp=None):
|
||||
self.manager = manager
|
||||
self._info = info
|
||||
self._add_details(info)
|
||||
self._loaded = loaded
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
def _add_details(self, info):
|
||||
for (k, v) in six.iteritems(info):
|
||||
try:
|
||||
setattr(self, k, v)
|
||||
self._info[k] = v
|
||||
except AttributeError:
|
||||
# In this case we already defined the attribute on the class
|
||||
pass
|
||||
|
||||
def __setstate__(self, d):
|
||||
for k, v in d.items():
|
||||
setattr(self, k, v)
|
||||
|
||||
def __getattr__(self, k):
|
||||
if k not in self.__dict__:
|
||||
# NOTE(RuiChen): disallow lazy-loading if already loaded once
|
||||
if not self.is_loaded():
|
||||
self.get()
|
||||
return self.__getattr__(k)
|
||||
raise AttributeError(k)
|
||||
else:
|
||||
return self.__dict__[k]
|
||||
|
||||
def __repr__(self):
|
||||
reprkeys = sorted(k for k in self.__dict__.keys() if k[0] != '_' and
|
||||
k not in ('manager', 'x_openstack_request_ids'))
|
||||
info = ", ".join("%s=%s" % (k, getattr(self, k)) for k in reprkeys)
|
||||
return "<%s %s>" % (self.__class__.__name__, info)
|
||||
|
||||
def get(self):
|
||||
# set_loaded() first ... so if we have to bail, we know we tried.
|
||||
self.set_loaded(True)
|
||||
if not hasattr(self.manager, 'get'):
|
||||
return
|
||||
|
||||
new = self.manager.get(self.id)
|
||||
if new:
|
||||
self._add_details(new._info)
|
||||
# The 'request_ids' attribute has been added,
|
||||
# so store the request id to it instead of _info
|
||||
self.append_request_ids(new.request_ids)
|
||||
|
||||
def __eq__(self, other):
|
||||
if not isinstance(other, self.__class__):
|
||||
return False
|
||||
return self._info == other._info
|
||||
|
||||
def __ne__(self, other):
|
||||
return not self.__eq__(other)
|
||||
|
||||
def is_loaded(self):
|
||||
return self._loaded
|
||||
|
||||
def set_loaded(self, val):
|
||||
self._loaded = val
|
||||
|
||||
def to_dict(self):
|
||||
return copy.deepcopy(self._info)
|
||||
|
||||
|
||||
class ListWithMeta(list, RequestIdMixin):
|
||||
def __init__(self, values, resp):
|
||||
super(ListWithMeta, self).__init__(values)
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
|
||||
class DictWithMeta(dict, RequestIdMixin):
|
||||
def __init__(self, values, resp):
|
||||
super(DictWithMeta, self).__init__(values)
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
|
||||
class TupleWithMeta(tuple, RequestIdMixin):
|
||||
def __new__(cls, values, resp):
|
||||
return super(TupleWithMeta, cls).__new__(cls, values)
|
||||
|
||||
def __init__(self, values, resp):
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
|
||||
class StrWithMeta(str, RequestIdMixin):
|
||||
def __new__(cls, value, resp):
|
||||
return super(StrWithMeta, cls).__new__(cls, value)
|
||||
|
||||
def __init__(self, values, resp):
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
|
||||
class BytesWithMeta(six.binary_type, RequestIdMixin):
|
||||
def __new__(cls, value, resp):
|
||||
return super(BytesWithMeta, cls).__new__(cls, value)
|
||||
|
||||
def __init__(self, values, resp):
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
|
||||
|
||||
if six.PY2:
|
||||
class UnicodeWithMeta(six.text_type, RequestIdMixin):
|
||||
def __new__(cls, value, resp):
|
||||
return super(UnicodeWithMeta, cls).__new__(cls, value)
|
||||
|
||||
def __init__(self, values, resp):
|
||||
self.request_ids_setup()
|
||||
self.append_request_ids(resp)
|
||||
277
nimbleclient/tests/unit/common/test_base.py
Normal file
277
nimbleclient/tests/unit/common/test_base.py
Normal file
@ -0,0 +1,277 @@
|
||||
# Copyright 2016 Huawei, Inc. 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 copy
|
||||
|
||||
import mock
|
||||
import six
|
||||
|
||||
from nimbleclient.common import base
|
||||
from nimbleclient.common import exceptions
|
||||
from nimbleclient.tests.unit import base as test_base
|
||||
from nimbleclient.tests.unit import fakes
|
||||
|
||||
|
||||
class TestResource(test_base.TestBase):
|
||||
|
||||
def test_resource_repr(self):
|
||||
r = base.Resource(None, dict(foo='bar', baz='spam'))
|
||||
self.assertEqual('<Resource baz=spam, foo=bar>', repr(r))
|
||||
|
||||
def test_getid(self):
|
||||
self.assertEqual(4, base.getid(4))
|
||||
|
||||
class TmpObject(object):
|
||||
id = 4
|
||||
self.assertEqual(4, base.getid(TmpObject))
|
||||
|
||||
def test_init_with_attribute_info(self):
|
||||
r = base.Resource(None, dict(foo='bar', baz='spam'))
|
||||
self.assertTrue(hasattr(r, 'foo'))
|
||||
self.assertEqual('bar', r.foo)
|
||||
self.assertTrue(hasattr(r, 'baz'))
|
||||
self.assertEqual('spam', r.baz)
|
||||
|
||||
def test_resource_lazy_getattr(self):
|
||||
fake_manager = mock.Mock()
|
||||
return_resource = base.Resource(None, dict(id=mock.sentinel.fake_id,
|
||||
foo='bar',
|
||||
name='fake_name'))
|
||||
fake_manager.get.return_value = return_resource
|
||||
|
||||
r = base.Resource(fake_manager,
|
||||
dict(id=mock.sentinel.fake_id, foo='bar'))
|
||||
self.assertTrue(hasattr(r, 'foo'))
|
||||
self.assertEqual('bar', r.foo)
|
||||
self.assertFalse(r.is_loaded())
|
||||
|
||||
# Trigger load
|
||||
self.assertEqual('fake_name', r.name)
|
||||
fake_manager.get.assert_called_once_with(mock.sentinel.fake_id)
|
||||
self.assertTrue(r.is_loaded())
|
||||
|
||||
# Missing stuff still fails after a second get
|
||||
self.assertRaises(AttributeError, getattr, r, 'blahblah')
|
||||
|
||||
def test_eq(self):
|
||||
# Two resources of the same type with the same id: not equal
|
||||
r1 = base.Resource(None, {'id': 1, 'name': 'hi'})
|
||||
r2 = base.Resource(None, {'id': 1, 'name': 'hello'})
|
||||
self.assertNotEqual(r1, r2)
|
||||
|
||||
# Two resources of different types: never equal
|
||||
r1 = base.Resource(None, {'id': 1})
|
||||
r2 = fakes.FaksResource(None, {'id': 1})
|
||||
self.assertNotEqual(r1, r2)
|
||||
|
||||
# Two resources with no ID: equal if their info is equal
|
||||
r1 = base.Resource(None, {'name': 'joe', 'age': 12})
|
||||
r2 = base.Resource(None, {'name': 'joe', 'age': 12})
|
||||
self.assertEqual(r1, r2)
|
||||
|
||||
def test_resource_object_with_request_ids(self):
|
||||
resp_obj = fakes.create_response_obj_with_header()
|
||||
r = base.Resource(None, {'name': '1'}, resp=resp_obj)
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, r.request_ids)
|
||||
|
||||
def test_resource_object_with_compute_request_ids(self):
|
||||
resp_obj = fakes.create_response_obj_with_compute_header()
|
||||
r = base.Resource(None, {'name': '1'}, resp=resp_obj)
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, r.request_ids)
|
||||
|
||||
|
||||
class TestManager(test_base.TestBase):
|
||||
fake_manager = fakes.create_resource_manager()
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'get')
|
||||
def test_manager_get(self, mock_get):
|
||||
mock_get.return_value = (fakes.create_response_obj_with_header(),
|
||||
mock.MagicMock())
|
||||
fake_resource = fakes.FaksResource(
|
||||
None, dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME))
|
||||
result = self.fake_manager.get(fake_resource)
|
||||
self.assertIsInstance(result, base.Resource)
|
||||
self.assertIsInstance(result._info, mock.MagicMock)
|
||||
self.assertTrue(result.is_loaded())
|
||||
expect_url = (fakes.FAKE_RESOURCE_ITEM_URL % fakes.FAKE_RESOURCE_ID)
|
||||
mock_get.assert_called_once_with(expect_url, headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'get')
|
||||
def test_manager_list(self, mock_get):
|
||||
mock_get.return_value = (fakes.create_response_obj_with_header(),
|
||||
mock.MagicMock())
|
||||
result = self.fake_manager.list()
|
||||
self.assertIsInstance(result, base.ListWithMeta)
|
||||
self.assertEqual([], result)
|
||||
expect_url = fakes.FAKE_RESOURCE_COLLECTION_URL
|
||||
mock_get.assert_called_once_with(expect_url, headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'patch')
|
||||
def test_manager_update(self, mock_patch):
|
||||
mock_patch.return_value = (fakes.create_response_obj_with_header(),
|
||||
mock.MagicMock())
|
||||
fake_resource = fakes.FaksResource(
|
||||
None, dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME))
|
||||
result = self.fake_manager.update(fake_resource)
|
||||
self.assertIsInstance(result, base.Resource)
|
||||
self.assertIsInstance(result._info, mock.MagicMock)
|
||||
self.assertFalse(result.is_loaded())
|
||||
expect_url = (fakes.FAKE_RESOURCE_ITEM_URL % fakes.FAKE_RESOURCE_ID)
|
||||
mock_patch.assert_called_once_with(expect_url, data=fake_resource,
|
||||
headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'delete')
|
||||
def test_manager_delete(self, mock_delete):
|
||||
mock_delete.return_value = (fakes.create_response_obj_with_header(),
|
||||
None)
|
||||
fake_resource = fakes.FaksResource(
|
||||
None, dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME))
|
||||
result = self.fake_manager.delete(fake_resource)
|
||||
self.assertIsInstance(result, base.TupleWithMeta)
|
||||
self.assertEqual(tuple(), result)
|
||||
expect_url = (fakes.FAKE_RESOURCE_ITEM_URL % fakes.FAKE_RESOURCE_ID)
|
||||
mock_delete.assert_called_once_with(expect_url, headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'post')
|
||||
def test_manager_create(self, mock_post):
|
||||
mock_post.return_value = (fakes.create_response_obj_with_header(),
|
||||
mock.MagicMock())
|
||||
fake_resource = fakes.FaksResource(
|
||||
None, dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME))
|
||||
result = self.fake_manager.create(fake_resource)
|
||||
self.assertIsInstance(result, base.Resource)
|
||||
self.assertIsInstance(result._info, mock.MagicMock)
|
||||
self.assertFalse(result.is_loaded())
|
||||
expect_url = fakes.FAKE_RESOURCE_COLLECTION_URL
|
||||
mock_post.assert_called_once_with(expect_url, data=fake_resource,
|
||||
headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'get')
|
||||
def test_manager_find(self, mock_get):
|
||||
fake_json_body_1 = dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME)
|
||||
fake_json_body_2 = dict(id='no_existed_id',
|
||||
name='no_existed_name')
|
||||
mock_get.side_effect = [
|
||||
(fakes.create_response_obj_with_header(),
|
||||
{'resources': [fake_json_body_1,
|
||||
fake_json_body_2]}),
|
||||
(fakes.create_response_obj_with_header(),
|
||||
fake_json_body_1)
|
||||
]
|
||||
result = self.fake_manager.find(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME)
|
||||
self.assertIsInstance(result, base.Resource)
|
||||
self.assertEqual(fakes.FAKE_RESOURCE_ID, result.id)
|
||||
self.assertEqual(fakes.FAKE_RESOURCE_NAME, result.name)
|
||||
self.assertTrue(result.is_loaded())
|
||||
expect_collection_url = fakes.FAKE_RESOURCE_COLLECTION_URL
|
||||
expect_item_url = (fakes.FAKE_RESOURCE_ITEM_URL %
|
||||
fakes.FAKE_RESOURCE_ID)
|
||||
mock_get.assert_has_calls(
|
||||
[mock.call(expect_collection_url, headers={}),
|
||||
mock.call(expect_item_url, headers={})])
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'get')
|
||||
def test_manager_find_no_result(self, mock_get):
|
||||
mock_get.return_value = (fakes.create_response_obj_with_header(),
|
||||
{'resources': []})
|
||||
self.assertRaises(exceptions.NotFound,
|
||||
self.fake_manager.find,
|
||||
id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME)
|
||||
expect_collection_url = fakes.FAKE_RESOURCE_COLLECTION_URL
|
||||
mock_get.assert_called_once_with(expect_collection_url, headers={})
|
||||
|
||||
@mock.patch.object(fakes.FakeHTTPClient, 'get')
|
||||
def test_manager_find_more_than_one_result(self, mock_get):
|
||||
fake_json_body_1 = dict(id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME)
|
||||
fake_json_body_2 = copy.deepcopy(fake_json_body_1)
|
||||
mock_get.return_value = (fakes.create_response_obj_with_header(),
|
||||
{'resources': [fake_json_body_1,
|
||||
fake_json_body_2]})
|
||||
self.assertRaises(exceptions.NoUniqueMatch,
|
||||
self.fake_manager.find,
|
||||
id=fakes.FAKE_RESOURCE_ID,
|
||||
name=fakes.FAKE_RESOURCE_NAME)
|
||||
expect_collection_url = fakes.FAKE_RESOURCE_COLLECTION_URL
|
||||
mock_get.assert_called_once_with(expect_collection_url, headers={})
|
||||
|
||||
|
||||
class ListWithMetaTest(test_base.TestBase):
|
||||
def test_list_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
obj = base.ListWithMeta([], resp)
|
||||
self.assertEqual([], obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
|
||||
|
||||
class DictWithMetaTest(test_base.TestBase):
|
||||
def test_dict_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
obj = base.DictWithMeta({}, resp)
|
||||
self.assertEqual({}, obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
|
||||
|
||||
class TupleWithMetaTest(test_base.TestBase):
|
||||
def test_tuple_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
expected_tuple = (1, 2)
|
||||
obj = base.TupleWithMeta(expected_tuple, resp)
|
||||
self.assertEqual(expected_tuple, obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
|
||||
|
||||
class StrWithMetaTest(test_base.TestBase):
|
||||
def test_str_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
obj = base.StrWithMeta('test-str', resp)
|
||||
self.assertEqual('test-str', obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
|
||||
|
||||
class BytesWithMetaTest(test_base.TestBase):
|
||||
def test_bytes_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
obj = base.BytesWithMeta(b'test-bytes', resp)
|
||||
self.assertEqual(b'test-bytes', obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
|
||||
|
||||
if six.PY2:
|
||||
class UnicodeWithMetaTest(test_base.TestBase):
|
||||
def test_unicode_with_meta(self):
|
||||
resp = fakes.create_response_obj_with_header()
|
||||
obj = base.UnicodeWithMeta(u'test-unicode', resp)
|
||||
self.assertEqual(u'test-unicode', obj)
|
||||
# Check request_ids attribute is added to obj
|
||||
self.assertTrue(hasattr(obj, 'request_ids'))
|
||||
self.assertEqual(fakes.FAKE_REQUEST_ID_LIST, obj.request_ids)
|
||||
@ -14,6 +14,88 @@
|
||||
#
|
||||
|
||||
from oslo_serialization import jsonutils
|
||||
from requests import Response
|
||||
|
||||
from nimbleclient.common import base
|
||||
|
||||
# fake request id
|
||||
FAKE_REQUEST_ID = 'req-0594c66b-6973-405c-ae2c-43fcfc00f2e3'
|
||||
FAKE_REQUEST_ID_LIST = [FAKE_REQUEST_ID]
|
||||
|
||||
# fake resource id
|
||||
FAKE_RESOURCE_ID = '0594c66b-6973-405c-ae2c-43fcfc00f2e3'
|
||||
FAKE_RESOURCE_NAME = 'name-0594c66b-6973-405c-ae2c-43fcfc00f2e3'
|
||||
|
||||
# fake resource response key
|
||||
FAKE_RESOURCE_ITEM_URL = '/resources/%s'
|
||||
FAKE_RESOURCE_COLLECTION_URL = '/resources'
|
||||
|
||||
|
||||
def create_response_obj_with_header():
|
||||
resp = Response()
|
||||
resp.headers['x-openstack-request-id'] = FAKE_REQUEST_ID
|
||||
return resp
|
||||
|
||||
|
||||
def create_response_obj_with_compute_header():
|
||||
resp = Response()
|
||||
resp.headers['x-compute-request-id'] = FAKE_REQUEST_ID
|
||||
return resp
|
||||
|
||||
|
||||
def create_resource_manager():
|
||||
return FakeManager()
|
||||
|
||||
|
||||
class FakeHTTPClient(object):
|
||||
|
||||
def get(self):
|
||||
pass
|
||||
|
||||
def head(self):
|
||||
pass
|
||||
|
||||
def post(self):
|
||||
pass
|
||||
|
||||
def put(self):
|
||||
pass
|
||||
|
||||
def delete(self):
|
||||
pass
|
||||
|
||||
def patch(self):
|
||||
pass
|
||||
|
||||
|
||||
class FaksResource(base.Resource):
|
||||
id = 'N/A'
|
||||
|
||||
|
||||
class FakeManager(base.ManagerWithFind):
|
||||
resource_class = FaksResource
|
||||
|
||||
def __init__(self, api=None):
|
||||
if not api:
|
||||
api = FakeHTTPClient()
|
||||
super(FakeManager, self).__init__(api)
|
||||
|
||||
def get(self, resource):
|
||||
return self._get(FAKE_RESOURCE_ITEM_URL % base.getid(resource))
|
||||
|
||||
def list(self):
|
||||
return self._list(FAKE_RESOURCE_COLLECTION_URL,
|
||||
response_key='resources')
|
||||
|
||||
def update(self, resource):
|
||||
return self._update(FAKE_RESOURCE_ITEM_URL % base.getid(resource),
|
||||
resource)
|
||||
|
||||
def create(self, resource):
|
||||
return self._create(FAKE_RESOURCE_COLLECTION_URL, resource)
|
||||
|
||||
def delete(self, resource):
|
||||
return self._delete(FAKE_RESOURCE_ITEM_URL % base.getid(resource))
|
||||
|
||||
|
||||
class FakeRaw(object):
|
||||
|
||||
@ -8,6 +8,7 @@ mock>=2.0 # BSD
|
||||
python-openstackclient>=2.1.0 # Apache-2.0
|
||||
python-subunit>=0.0.18 # Apache-2.0/BSD
|
||||
oslosphinx!=3.4.0,>=2.5.0 # Apache-2.0
|
||||
oslotest>=1.10.0 # Apache-2.0
|
||||
reno>=1.8.0 # Apache2
|
||||
requests-mock>=1.0 # Apache-2.0
|
||||
sphinx!=1.3b1,<1.3,>=1.2.1 # BSD
|
||||
|
||||
2
tox.ini
2
tox.ini
@ -28,7 +28,7 @@ commands =
|
||||
sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html
|
||||
|
||||
[testenv:debug]
|
||||
commands = oslo_debug_helper {posargs}
|
||||
commands = oslo_debug_helper -t nimbleclient/tests {posargs}
|
||||
|
||||
[flake8]
|
||||
# E123, E125 skipped as they are invalid PEP-8.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user