def4b17934
nova/db/__init__.py was importing * from nova.db.api. This meant that any time any code anywhere within the nova.db package was imported then nova.db.api was too, leading to a cascade of imports that may not have been desired. Also, in general, code in __init__.py is a pain. Therefore, this change adjusts code that so that either: * nova.db.api is used directly * nova.db.api is imported as 'db' In either case, the functionality remains the same. The primary goal of this change was to make it possible to import the model files without having to import the db api. Moving the model files to a different place in the directory hierarchy was considered, but given that "code in __init__.py is a pain" this mode was chosen. This looks like a very large change, but it is essentially adjusting package names, many in mocks. Change-Id: Ic1fd7c87ceda05eeb96735da2a415ef37060bb1a
274 lines
12 KiB
Python
274 lines
12 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.
|
|
"""Tests for keypair API."""
|
|
|
|
import mock
|
|
from oslo_concurrency import processutils
|
|
from oslo_config import cfg
|
|
import six
|
|
|
|
from nova.compute import api as compute_api
|
|
from nova import context
|
|
from nova import exception
|
|
from nova.objects import keypair as keypair_obj
|
|
from nova import quota
|
|
from nova.tests.unit.compute import test_compute
|
|
from nova.tests.unit import fake_crypto
|
|
from nova.tests.unit import fake_notifier
|
|
from nova.tests.unit.objects import test_keypair
|
|
from nova.tests.unit import utils as test_utils
|
|
|
|
CONF = cfg.CONF
|
|
|
|
|
|
class KeypairAPITestCase(test_compute.BaseTestCase):
|
|
def setUp(self):
|
|
super(KeypairAPITestCase, self).setUp()
|
|
self.keypair_api = compute_api.KeypairAPI()
|
|
self.ctxt = context.RequestContext('fake', 'fake')
|
|
self._keypair_db_call_stubs()
|
|
self.existing_key_name = 'fake existing key name'
|
|
self.pub_key = ('ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDLnVkqJu9WVf'
|
|
'/5StU3JCrBR2r1s1j8K1tux+5XeSvdqaM8lMFNorzbY5iyoBbR'
|
|
'S56gy1jmm43QsMPJsrpfUZKcJpRENSe3OxIIwWXRoiapZe78u/'
|
|
'a9xKwj0avFYMcws9Rk9iAB7W4K1nEJbyCPl5lRBoyqeHBqrnnu'
|
|
'XWEgGxJCK0Ah6wcOzwlEiVjdf4kxzXrwPHyi7Ea1qvnNXTziF8'
|
|
'yYmUlH4C8UXfpTQckwSwpDyxZUc63P8q+vPbs3Q2kw+/7vvkCK'
|
|
'HJAXVI+oCiyMMfffoTq16M1xfV58JstgtTqAXG+ZFpicGajREU'
|
|
'E/E3hO5MGgcHmyzIrWHKpe1n3oEGuz')
|
|
self.fingerprint = '4e:48:c6:a0:4a:f9:dd:b5:4c:85:54:5a:af:43:47:5a'
|
|
self.keypair_type = keypair_obj.KEYPAIR_TYPE_SSH
|
|
self.key_destroyed = False
|
|
|
|
def _keypair_db_call_stubs(self):
|
|
|
|
def db_key_pair_get_all_by_user(context, user_id, limit, marker):
|
|
return [dict(test_keypair.fake_keypair,
|
|
name=self.existing_key_name,
|
|
public_key=self.pub_key,
|
|
fingerprint=self.fingerprint)]
|
|
|
|
def db_key_pair_create(context, keypair):
|
|
return dict(test_keypair.fake_keypair, **keypair)
|
|
|
|
def db_key_pair_destroy(context, user_id, name):
|
|
if name == self.existing_key_name:
|
|
self.key_destroyed = True
|
|
|
|
def db_key_pair_get(context, user_id, name):
|
|
if name == self.existing_key_name and not self.key_destroyed:
|
|
return dict(test_keypair.fake_keypair,
|
|
name=self.existing_key_name,
|
|
public_key=self.pub_key,
|
|
fingerprint=self.fingerprint)
|
|
else:
|
|
raise exception.KeypairNotFound(user_id=user_id, name=name)
|
|
|
|
self.stub_out("nova.db.api.key_pair_get_all_by_user",
|
|
db_key_pair_get_all_by_user)
|
|
self.stub_out("nova.db.api.key_pair_create", db_key_pair_create)
|
|
self.stub_out("nova.db.api.key_pair_destroy", db_key_pair_destroy)
|
|
self.stub_out("nova.db.api.key_pair_get", db_key_pair_get)
|
|
|
|
def _check_notifications(self, action='create', key_name='foo'):
|
|
self.assertEqual(2, len(fake_notifier.NOTIFICATIONS))
|
|
|
|
n1 = fake_notifier.NOTIFICATIONS[0]
|
|
self.assertEqual('INFO', n1.priority)
|
|
self.assertEqual('keypair.%s.start' % action, n1.event_type)
|
|
self.assertEqual('api.%s' % CONF.host, n1.publisher_id)
|
|
self.assertEqual('fake', n1.payload['user_id'])
|
|
self.assertEqual('fake', n1.payload['tenant_id'])
|
|
self.assertEqual(key_name, n1.payload['key_name'])
|
|
|
|
n2 = fake_notifier.NOTIFICATIONS[1]
|
|
self.assertEqual('INFO', n2.priority)
|
|
self.assertEqual('keypair.%s.end' % action, n2.event_type)
|
|
self.assertEqual('api.%s' % CONF.host, n2.publisher_id)
|
|
self.assertEqual('fake', n2.payload['user_id'])
|
|
self.assertEqual('fake', n2.payload['tenant_id'])
|
|
self.assertEqual(key_name, n2.payload['key_name'])
|
|
|
|
|
|
class CreateImportSharedTestMixIn(object):
|
|
"""Tests shared between create and import_key.
|
|
|
|
Mix-in pattern is used here so that these `test_*` methods aren't picked
|
|
up by the test runner unless they are part of a 'concrete' test case.
|
|
"""
|
|
|
|
def assertKeypairRaises(self, exc_class, expected_message, name):
|
|
func = getattr(self.keypair_api, self.func_name)
|
|
|
|
args = []
|
|
if self.func_name == 'import_key_pair':
|
|
args.append(self.pub_key)
|
|
args.append(self.keypair_type)
|
|
|
|
exc = self.assertRaises(exc_class, func, self.ctxt, self.ctxt.user_id,
|
|
name, *args)
|
|
self.assertEqual(expected_message, six.text_type(exc))
|
|
|
|
def assertInvalidKeypair(self, expected_message, name):
|
|
msg = 'Keypair data is invalid: %s' % expected_message
|
|
self.assertKeypairRaises(exception.InvalidKeypair, msg, name)
|
|
|
|
def test_name_too_short(self):
|
|
msg = ('Keypair name must be string and between 1 '
|
|
'and 255 characters long')
|
|
self.assertInvalidKeypair(msg, '')
|
|
|
|
def test_name_too_long(self):
|
|
msg = ('Keypair name must be string and between 1 '
|
|
'and 255 characters long')
|
|
self.assertInvalidKeypair(msg, 'x' * 256)
|
|
|
|
def test_invalid_chars(self):
|
|
msg = "Keypair name contains unsafe characters"
|
|
self.assertInvalidKeypair(msg, '* BAD CHARACTERS! *')
|
|
|
|
def test_already_exists(self):
|
|
def db_key_pair_create_duplicate(context, keypair):
|
|
raise exception.KeyPairExists(key_name=keypair.get('name', ''))
|
|
|
|
self.stub_out("nova.db.api.key_pair_create",
|
|
db_key_pair_create_duplicate)
|
|
|
|
msg = ("Key pair '%(key_name)s' already exists." %
|
|
{'key_name': self.existing_key_name})
|
|
self.assertKeypairRaises(exception.KeyPairExists, msg,
|
|
self.existing_key_name)
|
|
|
|
@mock.patch.object(quota.QUOTAS, 'count_as_dict',
|
|
return_value={'user': {
|
|
'key_pairs': CONF.quota.key_pairs}})
|
|
def test_quota_limit(self, mock_count_as_dict):
|
|
msg = "Maximum number of key pairs exceeded"
|
|
self.assertKeypairRaises(exception.KeypairLimitExceeded, msg, 'foo')
|
|
|
|
|
|
class CreateKeypairTestCase(KeypairAPITestCase, CreateImportSharedTestMixIn):
|
|
func_name = 'create_key_pair'
|
|
|
|
@mock.patch('nova.compute.utils.notify_about_keypair_action')
|
|
def _check_success(self, mock_notify):
|
|
keypair, private_key = self.keypair_api.create_key_pair(
|
|
self.ctxt, self.ctxt.user_id, 'foo', key_type=self.keypair_type)
|
|
self.assertEqual('foo', keypair['name'])
|
|
self.assertEqual(self.keypair_type, keypair['type'])
|
|
mock_notify.assert_has_calls([
|
|
mock.call(context=self.ctxt, keypair=keypair,
|
|
action='create', phase='start'),
|
|
mock.call(context=self.ctxt, keypair=keypair,
|
|
action='create', phase='end')])
|
|
self._check_notifications()
|
|
|
|
def test_success_ssh(self):
|
|
self._check_success()
|
|
|
|
def test_success_x509(self):
|
|
self.keypair_type = keypair_obj.KEYPAIR_TYPE_X509
|
|
self._check_success()
|
|
|
|
def test_x509_subject_too_long(self):
|
|
# X509 keypairs will fail if the Subject they're created with
|
|
# is longer than 64 characters. The previous unit tests could not
|
|
# detect the issue because the ctxt.user_id was too short.
|
|
# This unit tests is added to prove this issue.
|
|
self.keypair_type = keypair_obj.KEYPAIR_TYPE_X509
|
|
self.ctxt.user_id = 'a' * 65
|
|
self.assertRaises(processutils.ProcessExecutionError,
|
|
self._check_success)
|
|
|
|
|
|
class ImportKeypairTestCase(KeypairAPITestCase, CreateImportSharedTestMixIn):
|
|
func_name = 'import_key_pair'
|
|
|
|
@mock.patch('nova.compute.utils.notify_about_keypair_action')
|
|
def _check_success(self, mock_notify):
|
|
keypair = self.keypair_api.import_key_pair(self.ctxt,
|
|
self.ctxt.user_id,
|
|
'foo',
|
|
self.pub_key,
|
|
self.keypair_type)
|
|
|
|
self.assertEqual('foo', keypair['name'])
|
|
self.assertEqual(self.keypair_type, keypair['type'])
|
|
self.assertEqual(self.fingerprint, keypair['fingerprint'])
|
|
self.assertEqual(self.pub_key, keypair['public_key'])
|
|
self.assertEqual(self.keypair_type, keypair['type'])
|
|
mock_notify.assert_has_calls([
|
|
mock.call(context=self.ctxt, keypair=keypair,
|
|
action='import', phase='start'),
|
|
mock.call(context=self.ctxt, keypair=keypair,
|
|
action='import', phase='end')])
|
|
self._check_notifications(action='import')
|
|
|
|
def test_success_ssh(self):
|
|
self._check_success()
|
|
|
|
def test_success_x509(self):
|
|
self.keypair_type = keypair_obj.KEYPAIR_TYPE_X509
|
|
certif, fingerprint = fake_crypto.get_x509_cert_and_fingerprint()
|
|
self.pub_key = certif
|
|
self.fingerprint = fingerprint
|
|
self._check_success()
|
|
|
|
def test_bad_key_data(self):
|
|
exc = self.assertRaises(exception.InvalidKeypair,
|
|
self.keypair_api.import_key_pair,
|
|
self.ctxt, self.ctxt.user_id, 'foo',
|
|
'bad key data')
|
|
msg = u'Keypair data is invalid: failed to generate fingerprint'
|
|
self.assertEqual(msg, six.text_type(exc))
|
|
|
|
|
|
class GetKeypairTestCase(KeypairAPITestCase):
|
|
def test_success(self):
|
|
keypair = self.keypair_api.get_key_pair(self.ctxt,
|
|
self.ctxt.user_id,
|
|
self.existing_key_name)
|
|
self.assertEqual(self.existing_key_name, keypair['name'])
|
|
|
|
|
|
class GetKeypairsTestCase(KeypairAPITestCase):
|
|
def test_success(self):
|
|
keypairs = self.keypair_api.get_key_pairs(self.ctxt, self.ctxt.user_id)
|
|
self.assertEqual([self.existing_key_name],
|
|
[k['name'] for k in keypairs])
|
|
|
|
|
|
class DeleteKeypairTestCase(KeypairAPITestCase):
|
|
@mock.patch('nova.compute.utils.notify_about_keypair_action')
|
|
def test_success(self, mock_notify):
|
|
self.keypair_api.delete_key_pair(self.ctxt, self.ctxt.user_id,
|
|
self.existing_key_name)
|
|
self.assertRaises(exception.KeypairNotFound,
|
|
self.keypair_api.get_key_pair, self.ctxt, self.ctxt.user_id,
|
|
self.existing_key_name)
|
|
|
|
match_by_name = test_utils.CustomMockCallMatcher(
|
|
lambda keypair: keypair['name'] == self.existing_key_name)
|
|
|
|
mock_notify.assert_has_calls([
|
|
mock.call(context=self.ctxt,
|
|
keypair=match_by_name,
|
|
action='delete', phase='start'),
|
|
mock.call(context=self.ctxt,
|
|
keypair=match_by_name,
|
|
action='delete', phase='end')])
|
|
self._check_notifications(action='delete',
|
|
key_name=self.existing_key_name)
|