openstack
/
openstacksdk
Code Issues Proposed changes

Use Resource layer for compute KeyPairs 

Move KeyPairs logic from the cloud layer to resource.

Change-Id: I2c06a5c76d9b2fa872c1cbfab528ed805aa5063a
This commit is contained in:
Artem Goncharov 2019-06-14 15:03:20 +02:00
parent 38847204f9
commit 15baef656a
5 changed files with 97 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
openstack/cloud/_compute.py
View File

@ -80,7 +80,9 @@ class ComputeCloudMixin(_normalize.Normalizer):
return extension_name in self._nova_extensions()
def search_keypairs(self, name_or_id=None, filters=None):
keypairs = self.list_keypairs()
keypairs = self.list_keypairs(
filters=filters if isinstance(filters, dict) else None
)
return _utils._filter_list(keypairs, name_or_id, filters)
def search_flavors(self, name_or_id=None, filters=None, get_extra=True):
@ -115,17 +117,16 @@ class ComputeCloudMixin(_normalize.Normalizer):
server_groups = self.list_server_groups()
return _utils._filter_list(server_groups, name_or_id, filters)
def list_keypairs(self):
def list_keypairs(self, filters=None):
"""List all available keypairs.
:returns: A list of ``munch.Munch`` containing keypair info.
"""
data = proxy._json_response(
self.compute.get('/os-keypairs'),
error_message="Error fetching keypair list")
return self._normalize_keypairs([
k['keypair'] for k in self._get_and_munchify('keypairs', data)])
if not filters:
filters = {}
return list(self.compute.keypairs(allow_unknown_params=True,
**filters))
@_utils.cache_on_arguments()
def list_availability_zone_names(self, unavailable=False):
@ -620,13 +621,7 @@ class ComputeCloudMixin(_normalize.Normalizer):
}
if public_key:
keypair['public_key'] = public_key
data = proxy._json_response(
self.compute.post(
'/os-keypairs',
json={'keypair': keypair}),
error_message="Unable to create keypair {name}".format(name=name))
return self._normalize_keypair(
self._get_and_munchify('keypair', data))
return self.compute.create_keypair(**keypair)
def delete_keypair(self, name):
"""Delete a keypair.
@ -638,9 +633,8 @@ class ComputeCloudMixin(_normalize.Normalizer):
:raises: OpenStackCloudException on operation error.
"""
try:
proxy._json_response(self.compute.delete(
'/os-keypairs/{name}'.format(name=name)))
except exc.OpenStackCloudURINotFound:
self.compute.delete_keypair(name, ignore_missing=False)
except exceptions.ResourceNotFound:
self.log.debug("Keypair %s not found for deleting", name)
return False
return True

6
openstack/compute/v2/_proxy.py
View File

@ -403,13 +403,15 @@ class Proxy(proxy.Proxy):
return self._find(_keypair.Keypair, name_or_id,
ignore_missing=ignore_missing)
def keypairs(self):
def keypairs(self, **query):
"""Return a generator of keypairs
:param kwargs query: Optional query parameters to be sent to limit
the resources being returned.
:returns: A generator of keypair objects
:rtype: :class:`~openstack.compute.v2.keypair.Keypair`
"""
return self._list(_keypair.Keypair)
return self._list(_keypair.Keypair, **query)
def get_limits(self):
"""Retrieve limits that are applied to the project's account

39
openstack/compute/v2/keypair.py
View File

@ -18,6 +18,9 @@ class Keypair(resource.Resource):
resources_key = 'keypairs'
base_path = '/os-keypairs'
_query_mapping = resource.QueryParameters(
'user_id')
# capabilities
allow_create = True
allow_fetch = True
@ -25,6 +28,10 @@ class Keypair(resource.Resource):
allow_list = True
# Properties
#: The date and time when the resource was created.
created_at = resource.Body('created_at')
#: A boolean indicates whether this keypair is deleted or not.
is_deleted = resource.Body('deleted', type=bool)
#: The short fingerprint associated with the ``public_key`` for
#: this keypair.
fingerprint = resource.Body('fingerprint')
@ -42,6 +49,10 @@ class Keypair(resource.Resource):
private_key = resource.Body('private_key')
#: The SSH public key that is paired with the server.
public_key = resource.Body('public_key')
#: The type of the keypair.
type = resource.Body('type', default='ssh')
#: The user_id for a keypair.
user_id = resource.Body('user_id')
def _consume_attrs(self, mapping, attrs):
# TODO(mordred) This should not be required. However, without doing
@ -51,16 +62,22 @@ class Keypair(resource.Resource):
return super(Keypair, self)._consume_attrs(mapping, attrs)
@classmethod
def list(cls, session, paginated=False, base_path=None):
def existing(cls, connection=None, **kwargs):
"""Create an instance of an existing remote resource.
if base_path is None:
base_path = cls.base_path
When creating the instance set the ``_synchronized`` parameter
of :class:`Resource` to ``True`` to indicate that it represents the
state of an existing server-side resource. As such, all attributes
passed in ``**kwargs`` are considered "clean", such that an immediate
:meth:`update` call would not generate a body of attributes to be
modified on the server.
resp = session.get(base_path,
headers={"Accept": "application/json"})
resp = resp.json()
resp = resp[cls.resources_key]
for data in resp:
value = cls.existing(**data[cls.resource_key])
yield value
:param dict kwargs: Each of the named arguments will be set as
attributes on the resulting Resource object.
"""
# Listing KPs return list with resource_key structure. Instead of
# overriding whole list just try to create object smart.
if cls.resource_key in kwargs:
args = kwargs.pop(cls.resource_key)
kwargs.update(**args)
return cls(_synchronized=True, connection=connection, **kwargs)

36
openstack/tests/unit/cloud/test_keypair.py
View File

@ -38,7 +38,10 @@ class TestKeypair(base.TestCase):
new_key = self.cloud.create_keypair(
self.keyname, self.key['public_key'])
self.assertEqual(new_key, self.cloud._normalize_keypair(self.key))
new_key_cmp = new_key.to_dict(ignore_none=True)
new_key_cmp.pop('location')
new_key_cmp.pop('id')
self.assertEqual(new_key_cmp, self.key)
self.assert_calls()
@ -94,7 +97,36 @@ class TestKeypair(base.TestCase):
])
keypairs = self.cloud.list_keypairs()
self.assertEqual(keypairs, self.cloud._normalize_keypairs([self.key]))
self.assertEqual(len(keypairs), 1)
self.assertEqual(keypairs[0].name, self.key['name'])
self.assert_calls()
def test_list_keypairs_empty_filters(self):
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['os-keypairs']),
json={'keypairs': [{'keypair': self.key}]}),
])
keypairs = self.cloud.list_keypairs(filters=None)
self.assertEqual(len(keypairs), 1)
self.assertEqual(keypairs[0].name, self.key['name'])
self.assert_calls()
def test_list_keypairs_notempty_filters(self):
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'compute', 'public', append=['os-keypairs'],
qs_elements=['user_id=b']),
json={'keypairs': [{'keypair': self.key}]}),
])
keypairs = self.cloud.list_keypairs(
filters={'user_id': 'b', 'fake': 'dummy'})
self.assertEqual(len(keypairs), 1)
self.assertEqual(keypairs[0].name, self.key['name'])
self.assert_calls()
def test_list_keypairs_exception(self):

21
openstack/tests/unit/compute/v2/test_keypair.py
View File

@ -15,10 +15,14 @@ from openstack.tests.unit import base
from openstack.compute.v2 import keypair
EXAMPLE = {
'created_at': 'some_time',
'deleted': False,
'fingerprint': '1',
'name': '2',
'public_key': '3',
'private_key': '3',
'private_key': '4',
'type': 'ssh',
'user_id': '5'
}
@ -35,9 +39,24 @@ class TestKeypair(base.TestCase):
self.assertTrue(sot.allow_delete)
self.assertTrue(sot.allow_list)
self.assertDictEqual({'limit': 'limit',
'marker': 'marker',
'user_id': 'user_id'},
sot._query_mapping._mapping)
def test_make_it(self):
sot = keypair.Keypair(**EXAMPLE)
self.assertEqual(EXAMPLE['created_at'], sot.created_at)
self.assertEqual(EXAMPLE['deleted'], sot.is_deleted)
self.assertEqual(EXAMPLE['fingerprint'], sot.fingerprint)
self.assertEqual(EXAMPLE['name'], sot.name)
self.assertEqual(EXAMPLE['public_key'], sot.public_key)
self.assertEqual(EXAMPLE['private_key'], sot.private_key)
self.assertEqual(EXAMPLE['type'], sot.type)
self.assertEqual(EXAMPLE['user_id'], sot.user_id)
def test_make_it_defaults(self):
EXAMPLE_DEFAULT = EXAMPLE.copy()
EXAMPLE_DEFAULT.pop('type')
sot = keypair.Keypair(**EXAMPLE_DEFAULT)
self.assertEqual(EXAMPLE['type'], sot.type)