summaryrefslogtreecommitdiffstats
path: root/nova/tests
diff options
context:
space:
mode:
Diffstat (limited to 'nova/tests')
-rw-r--r--nova/tests/api/openstack/compute/contrib/test_keypairs.py8
-rw-r--r--nova/tests/compute/test_compute.py3
-rw-r--r--nova/tests/test_db_api.py61
-rw-r--r--nova/tests/test_migrations.py39
4 files changed, 107 insertions, 4 deletions
diff --git a/nova/tests/api/openstack/compute/contrib/test_keypairs.py b/nova/tests/api/openstack/compute/contrib/test_keypairs.py
index 025845637..e338cad69 100644
--- a/nova/tests/api/openstack/compute/contrib/test_keypairs.py
+++ b/nova/tests/api/openstack/compute/contrib/test_keypairs.py
@@ -40,7 +40,7 @@ def db_key_pair_get_all_by_user(self, user_id):
def db_key_pair_create(self, keypair):
- pass
+ return keypair
def db_key_pair_destroy(context, user_id, name):
@@ -48,8 +48,8 @@ def db_key_pair_destroy(context, user_id, name):
raise Exception()
-def db_key_pair_get(context, user_id, name):
- pass
+def db_key_pair_create_duplicate(context, keypair):
+ raise exception.KeyPairExists(key_name=keypair.get('name', ''))
class KeypairsTest(test.TestCase):
@@ -205,7 +205,7 @@ class KeypairsTest(test.TestCase):
self.assertEqual(res.status_int, 413)
def test_keypair_create_duplicate(self):
- self.stubs.Set(db, "key_pair_get", db_key_pair_get)
+ self.stubs.Set(db, "key_pair_create", db_key_pair_create_duplicate)
body = {'keypair': {'name': 'create_duplicate'}}
req = webob.Request.blank('/v2/fake/os-keypairs')
req.method = 'POST'
diff --git a/nova/tests/compute/test_compute.py b/nova/tests/compute/test_compute.py
index 097c87848..395a075c8 100644
--- a/nova/tests/compute/test_compute.py
+++ b/nova/tests/compute/test_compute.py
@@ -7315,6 +7315,9 @@ class KeypairAPITestCase(BaseTestCase):
self.ctxt, self.ctxt.user_id, '* BAD CHARACTERS! *')
def test_create_keypair_already_exists(self):
+ def db_key_pair_create_duplicate(context, keypair):
+ raise exception.KeyPairExists(key_name=keypair.get('name', ''))
+ self.stubs.Set(db, "key_pair_create", db_key_pair_create_duplicate)
self.assertRaises(exception.KeyPairExists,
self.keypair_api.create_key_pair,
self.ctxt, self.ctxt.user_id,
diff --git a/nova/tests/test_db_api.py b/nova/tests/test_db_api.py
index ce4c8fe7f..23b8ebcb0 100644
--- a/nova/tests/test_db_api.py
+++ b/nova/tests/test_db_api.py
@@ -1066,6 +1066,67 @@ class DbApiTestCase(DbTestCase):
_compare(bw_usages[2], expected_bw_usages[2])
timeutils.clear_time_override()
+ def test_key_pair_create(self):
+ ctxt = context.get_admin_context()
+ values = {'name': 'test_keypair', 'public_key': 'test-public-key',
+ 'user_id': 'test_user_id', 'fingerprint': 'test_fingerprint'}
+ keypair = db.key_pair_create(ctxt, values)
+ self.assertNotEqual(None, keypair)
+ for name, value in values.iteritems():
+ self.assertEqual(keypair.get(name), value)
+
+ def test_key_pair_create_with_duplicate_name(self):
+ ctxt = context.get_admin_context()
+ values = {'name': 'test_keypair', 'public_key': 'test-public-key',
+ 'user_id': 'test_user_id', 'fingerprint': 'test_fingerprint'}
+ keypair = db.key_pair_create(ctxt, values)
+ self.assertRaises(exception.KeyPairExists,
+ db.key_pair_create, ctxt, values)
+
+ def test_admin_get_deleted_keypair(self):
+ # Test deleted keypair can be read by admin user.
+ ctxt = context.get_admin_context()
+ values = {'name': 'test_keypair', 'public_key': 'test-public-key',
+ 'user_id': 'test_user_id', 'fingerprint': 'test_fingerprint'}
+ keypair = db.key_pair_create(ctxt, values)
+ db.key_pair_destroy(ctxt, keypair['user_id'], keypair['name'])
+
+ # Raise exception when read_deleted is 'no'.
+ self.assertRaises(exception.KeypairNotFound, db.key_pair_get, ctxt,
+ keypair['user_id'], keypair['name'])
+ ctxt = ctxt.elevated(read_deleted='yes')
+ db_keypair = db.key_pair_get(ctxt, keypair['user_id'],
+ keypair['name'])
+ self.assertEqual(db_keypair['name'], keypair['name'])
+ self.assertEqual(db_keypair['deleted'], keypair['id'])
+
+ def test_admin_get_all_keypairs_including_deleted(self):
+ # Test all deleted/non-deleted keypairs can be read by admin user.
+ ctxt = context.get_admin_context()
+ keypair1_values = {'name': 'test_keypair1',
+ 'public_key': 'test-public-key1',
+ 'user_id': 'test_user_id',
+ 'fingerprint': 'test_fingerprint1'}
+ keypair2_values = {'name': 'test_keypair2',
+ 'public_key': 'test-public-key2',
+ 'user_id': 'test_user_id',
+ 'fingerprint': 'test_fingerprint2'}
+ keypair1 = db.key_pair_create(ctxt, keypair1_values)
+ keypair2 = db.key_pair_create(ctxt, keypair2_values)
+ db.key_pair_destroy(ctxt, keypair1['user_id'], keypair1['name'])
+ db.key_pair_destroy(ctxt, keypair2['user_id'], keypair2['name'])
+ # Returns non-deleted keypairs.
+ result = db.key_pair_get_all_by_user(ctxt, keypair1['user_id'])
+ self.assertEqual(result, [])
+ ctxt = ctxt.elevated(read_deleted='yes')
+ # Returns deleted and non-deleted keypairs.
+ db_keypairs = db.key_pair_get_all_by_user(ctxt, keypair1['user_id'])
+ expected_deleted_ids = [keypair1['id'], keypair2['id']]
+ expected_keypair_names = [keypair1['name'], keypair2['name']]
+ for keypair in db_keypairs:
+ self.assertTrue(keypair['name'] in expected_keypair_names)
+ self.assertTrue(keypair['deleted'] in expected_deleted_ids)
+
def _get_fake_aggr_values():
return {'name': 'fake_aggregate'}
diff --git a/nova/tests/test_migrations.py b/nova/tests/test_migrations.py
index 98c50aaa7..c78bb6d24 100644
--- a/nova/tests/test_migrations.py
+++ b/nova/tests/test_migrations.py
@@ -1151,6 +1151,45 @@ class TestNovaMigrations(BaseMigrationTestCase, CommonTestsMixIn):
fetchall()
self.assertEqual(1, len(uc_name2_rows))
+ # migration 173, add unique constraint to keypairs
+ def _pre_upgrade_173(self, engine):
+ created_at = [datetime.datetime.now() for x in range(0, 7)]
+ fake_keypairs = [dict(name='key1', user_id='1a',
+ created_at=created_at[0],
+ deleted=0),
+ dict(name='key1', user_id='1a',
+ created_at=created_at[1],
+ deleted=0),
+ dict(name='key1', user_id='1a',
+ created_at=created_at[2],
+ deleted=0)
+ ]
+ keypairs = get_table(engine, 'key_pairs')
+ engine.execute(keypairs.insert(), fake_keypairs)
+ return fake_keypairs
+
+ def _check_173(self, engine, data):
+ keypairs = get_table(engine, 'key_pairs')
+ # Unique constraints are not listed in table.constraints for any db.
+ # So, simply add a duplicate keypair to check if unique constraint
+ # is applied to the key_pairs table or not.
+ insert = keypairs.insert()
+ duplicate_keypair = dict(name='key4', user_id='4a',
+ created_at=datetime.datetime.now(),
+ deleted=0)
+ insert.execute(duplicate_keypair)
+ # Insert again
+ self.assertRaises(sqlalchemy.exc.IntegrityError, insert.execute,
+ duplicate_keypair)
+
+ # Get all unique records
+ rows = keypairs.select().\
+ where(keypairs.c.deleted != keypairs.c.id).\
+ execute().\
+ fetchall()
+ # Ensure the number of unique keypairs is correct
+ self.assertEqual(len(rows), 2)
+
class TestBaremetalMigrations(BaseMigrationTestCase, CommonTestsMixIn):
"""Test sqlalchemy-migrate migrations."""