Merge "Add tests for compute v2.10 microversion"
diff --git a/tempest/api/compute/admin/test_keypairs_v210.py b/tempest/api/compute/admin/test_keypairs_v210.py
new file mode 100644
index 0000000..dfa7c39
--- /dev/null
+++ b/tempest/api/compute/admin/test_keypairs_v210.py
@@ -0,0 +1,76 @@
+# Copyright 2016 NEC Corporation.
+# 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.
+
+from tempest.api.compute.keypairs import base
+from tempest.common.utils import data_utils
+from tempest import test
+
+
+class KeyPairsV210TestJSON(base.BaseKeypairTest):
+ credentials = ['primary', 'admin']
+ min_microversion = '2.10'
+
+ @classmethod
+ def setup_clients(cls):
+ super(KeyPairsV210TestJSON, cls).setup_clients()
+ cls.client = cls.os_adm.keypairs_client
+ cls.non_admin_client = cls.os.keypairs_client
+
+ def _create_and_check_keypairs(self, user_id):
+ key_list = list()
+ for i in range(2):
+ k_name = data_utils.rand_name('keypair')
+ keypair = self._create_keypair(k_name,
+ keypair_type='ssh',
+ user_id=user_id)
+ self.assertEqual(k_name, keypair['name'],
+ "The created keypair name is not equal "
+ "to the requested name!")
+ self.assertEqual(user_id, keypair['user_id'],
+ "The created keypair is not for requested user!")
+ keypair.pop('private_key', None)
+ keypair.pop('user_id')
+ key_list.append(keypair)
+ return key_list
+
+ @test.idempotent_id('3c8484af-cfb3-48f6-b8ba-d5d58bbf3eac')
+ def test_admin_manage_keypairs_for_other_users(self):
+ user_id = self.non_admin_client.user_id
+ key_list = self._create_and_check_keypairs(user_id)
+ first_keyname = key_list[0]['name']
+ keypair_detail = self.client.show_keypair(first_keyname,
+ user_id=user_id)['keypair']
+ self.assertEqual(first_keyname, keypair_detail['name'])
+ self.assertEqual(user_id, keypair_detail['user_id'],
+ "The fetched keypair is not for requested user!")
+ # Create a admin keypair
+ admin_k_name = data_utils.rand_name('keypair')
+ admin_keypair = self._create_keypair(admin_k_name, keypair_type='ssh')
+ admin_keypair.pop('private_key', None)
+ admin_keypair.pop('user_id')
+
+ # Admin fetch keypairs list of non admin user
+ keypairs = self.client.list_keypairs(user_id=user_id)['keypairs']
+ fetched_list = [keypair['keypair'] for keypair in keypairs]
+
+ # Check admin keypair is not present in non admin user keypairs list
+ self.assertNotIn(admin_keypair, fetched_list,
+ "The fetched user keypairs has admin keypair!")
+
+ # Now check if all the created keypairs are in the fetched list
+ missing_kps = [kp for kp in key_list if kp not in fetched_list]
+ self.assertFalse(missing_kps,
+ "Failed to find keypairs %s in fetched list"
+ % ', '.join(m_key['name'] for m_key in missing_kps))
diff --git a/tempest/api/compute/keypairs/base.py b/tempest/api/compute/keypairs/base.py
index ebfb724..ad7f958 100644
--- a/tempest/api/compute/keypairs/base.py
+++ b/tempest/api/compute/keypairs/base.py
@@ -24,15 +24,21 @@
super(BaseKeypairTest, cls).setup_clients()
cls.client = cls.keypairs_client
- def _delete_keypair(self, keypair_name):
- self.client.delete_keypair(keypair_name)
+ def _delete_keypair(self, keypair_name, **params):
+ self.client.delete_keypair(keypair_name, **params)
- def _create_keypair(self, keypair_name, pub_key=None, keypair_type=None):
+ def _create_keypair(self, keypair_name,
+ pub_key=None, keypair_type=None,
+ user_id=None):
kwargs = {'name': keypair_name}
+ delete_params = {}
if pub_key:
kwargs.update({'public_key': pub_key})
if keypair_type:
kwargs.update({'type': keypair_type})
+ if user_id:
+ kwargs.update({'user_id': user_id})
+ delete_params['user_id'] = user_id
body = self.client.create_keypair(**kwargs)['keypair']
- self.addCleanup(self._delete_keypair, keypair_name)
+ self.addCleanup(self._delete_keypair, keypair_name, **delete_params)
return body
diff --git a/tempest/services/compute/json/keypairs_client.py b/tempest/services/compute/json/keypairs_client.py
index 2af55b2..fcb4f2a 100644
--- a/tempest/services/compute/json/keypairs_client.py
+++ b/tempest/services/compute/json/keypairs_client.py
@@ -14,6 +14,7 @@
# under the License.
from oslo_serialization import jsonutils as json
+from six.moves.urllib import parse as urllib
from tempest.api_schema.response.compute.v2_1 import keypairs as schemav21
from tempest.api_schema.response.compute.v2_2 import keypairs as schemav22
@@ -26,15 +27,21 @@
schema_versions_info = [{'min': None, 'max': '2.1', 'schema': schemav21},
{'min': '2.2', 'max': None, 'schema': schemav22}]
- def list_keypairs(self):
- resp, body = self.get("os-keypairs")
+ def list_keypairs(self, **params):
+ url = 'os-keypairs'
+ if params:
+ url += '?%s' % urllib.urlencode(params)
+ resp, body = self.get(url)
body = json.loads(body)
schema = self.get_schema(self.schema_versions_info)
self.validate_response(schema.list_keypairs, resp, body)
return rest_client.ResponseBody(resp, body)
- def show_keypair(self, keypair_name):
- resp, body = self.get("os-keypairs/%s" % keypair_name)
+ def show_keypair(self, keypair_name, **params):
+ url = "os-keypairs/%s" % keypair_name
+ if params:
+ url += '?%s' % urllib.urlencode(params)
+ resp, body = self.get(url)
body = json.loads(body)
schema = self.get_schema(self.schema_versions_info)
self.validate_response(schema.get_keypair, resp, body)
@@ -48,8 +55,11 @@
self.validate_response(schema.create_keypair, resp, body)
return rest_client.ResponseBody(resp, body)
- def delete_keypair(self, keypair_name):
- resp, body = self.delete("os-keypairs/%s" % keypair_name)
+ def delete_keypair(self, keypair_name, **params):
+ url = "os-keypairs/%s" % keypair_name
+ if params:
+ url += '?%s' % urllib.urlencode(params)
+ resp, body = self.delete(url)
schema = self.get_schema(self.schema_versions_info)
self.validate_response(schema.delete_keypair, resp, body)
return rest_client.ResponseBody(resp, body)