Merge "Add tests for compute v2.10 microversion"

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))

tempest/api/compute/keypairs/base.py

@@ -24,15 +24,21 @@ class BaseKeypairTest(base.BaseV2ComputeTest):
         super(BaseKeypairTest, cls).setup_clients()
         cls.client = cls.keypairs_client
 
-    def _delete_keypair(self, keypair_name):
+    def _delete_keypair(self, keypair_name, **params):
-        self.client.delete_keypair(keypair_name)
+        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

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 @@ class KeyPairsClient(base_compute_client.BaseComputeClient):
     schema_versions_info = [{'min': None, 'max': '2.1', 'schema': schemav21},
                             {'min': '2.2', 'max': None, 'schema': schemav22}]
 
-    def list_keypairs(self):
+    def list_keypairs(self, **params):
-        resp, body = self.get("os-keypairs")
+        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):
+    def show_keypair(self, keypair_name, **params):
-        resp, body = self.get("os-keypairs/%s" % keypair_name)
+        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 @@ class KeyPairsClient(base_compute_client.BaseComputeClient):
         self.validate_response(schema.create_keypair, resp, body)
         return rest_client.ResponseBody(resp, body)
 
-    def delete_keypair(self, keypair_name):
+    def delete_keypair(self, keypair_name, **params):
-        resp, body = self.delete("os-keypairs/%s" % keypair_name)
+        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)