openstack
/
keystone-tempest-plugin
Code Issues Proposed changes

Add tempest clients for limits 

This change adds tempest clients for the registered limits and limits
APIs. While those APIs are experimental, it's best to start development
of the tempest tests in the keystone plugin rather than in tempest. This
base can be used for both developing exhaustive API tests for these APIs
as well as for RBAC tests.

Change-Id: I30b5b2ac5f10fd457e436df876f872432059b655
This commit is contained in:
Colleen Murphy 2020-07-29 16:42:58 -07:00
parent aa17472706
commit 5ee9af871d
3 changed files with 168 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
keystone_tempest_plugin/clients.py
View File

@ -19,6 +19,7 @@ from keystone_tempest_plugin.services.identity.v3 import (
from keystone_tempest_plugin.services.identity.v3 import (
service_providers_client)
from keystone_tempest_plugin.services.identity.v3 import auth_client
from keystone_tempest_plugin.services.identity.v3 import limits_client
from keystone_tempest_plugin.services.identity.v3 import saml2_client
from tempest import clients
@ -29,7 +30,10 @@ class Manager(clients.Manager):
def __init__(self, credentials):
super(Manager, self).__init__(credentials)
# keystone auth client
self.auth_client = auth_client.AuthClient(self.auth_provider)
# federation clients
self.identity_providers_client = (
identity_providers_client.IdentityProvidersClient(
self.auth_provider))
@ -40,3 +44,8 @@ class Manager(clients.Manager):
self.service_providers_client = (
service_providers_client.ServiceProvidersClient(
self.auth_provider))
# unified limits clients
self.registered_limits_client = limits_client.RegisteredLimitsClient(
self.auth_provider)
self.limits_client = limits_client.LimitsClient(self.auth_provider)

156
keystone_tempest_plugin/services/identity/v3/limits_client.py Normal file
View File

@ -0,0 +1,156 @@
# Copyright 2020 SUSE LLC
#
# 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.
import json
import http.client
from tempest.lib.common import rest_client
from keystone_tempest_plugin.services.identity import clients
class RegisteredLimitsClient(clients.Identity):
subpath_prefix = 'registered_limits'
collection_url = subpath_prefix
entity_url = subpath_prefix + '/%s'
def create_registered_limits(self, payload):
"""Create a list of registered limits.
:param body: A list of registered limits objects.
"""
post_body = json.dumps({'registered_limits': payload})
resp, body = super(RegisteredLimitsClient, self).post(
self.collection_url, post_body)
self.expected_success(http.client.CREATED, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def list_registered_limits(self, **kwargs):
"""List registered limits.
:param kwargs: Filter by service_id, region_id, or resource_name
"""
resp, body = super(RegisteredLimitsClient, self).get(
self.collection_url, **kwargs)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def update_registered_limit(self, registered_limit_id, registered_limit):
"""Update a registered limit.
:param registered_limit_id: ID of registered limit to update
:param registered_limit: new registered limit object
"""
patch_body = json.dumps({'registered_limit': registered_limit})
resp, body = super(RegisteredLimitsClient, self).patch(
self.entity_url % registered_limit_id, patch_body)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def show_registered_limit(self, registered_limit_id):
"""Get a registered limit.
:param registered_limit_id: ID of registered limit to show
"""
resp, body = super(RegisteredLimitsClient, self).get(
self.entity_url % registered_limit_id)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def delete_registered_limit(self, registered_limit_id):
"""Delete a registered limit.
:param registered_limit_id: ID of registered limit to delete.
"""
resp, body = super(RegisteredLimitsClient, self).delete(
self.entity_url % registered_limit_id)
self.expected_success(http.client.NO_CONTENT, resp.status)
return rest_client.ResponseBody(resp, body)
class LimitsClient(clients.Identity):
subpath_prefix = 'limits'
collection_url = subpath_prefix
entity_url = subpath_prefix + '/%s'
def limits_model(self):
"""Get limits model from server."""
url = self.entity_url % 'model'
resp, body = super(LimitsClient, self).get(url)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def create_limits(self, payload):
"""Create a list of project limits.
:param body: A list of project limits objects.
"""
post_body = json.dumps({'limits': payload})
resp, body = super(LimitsClient, self).post(
self.collection_url, post_body)
self.expected_success(http.client.CREATED, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def list_limits(self, **kwargs):
"""List project limits.
:param kwargs: Filter by service_id, region_id, resource_name,
or project/domain ID
"""
resp, body = super(LimitsClient, self).get(
self.collection_url, **kwargs)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def update_limit(self, limit_id, limit):
"""Update a project limit.
:param limit_id: ID of project limit to update
:param limit: new project limit object
"""
patch_body = json.dumps({'limit': limit})
resp, body = super(LimitsClient, self).patch(
self.entity_url % limit_id, patch_body)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def show_limit(self, limit_id):
"""Get a project limit.
:param limit_id: ID of project limit to show
"""
resp, body = super(LimitsClient, self).get(self.entity_url % limit_id)
self.expected_success(http.client.OK, resp.status)
body = json.loads(body.decode('utf-8'))
return rest_client.ResponseBody(resp, body)
def delete_limit(self, limit_id):
"""Delete a project limit.
:param limit_id: ID of project limit to delete.
"""
resp, body = super(LimitsClient, self).delete(
self.entity_url % limit_id)
self.expected_success(http.client.NO_CONTENT, resp.status)
return rest_client.ResponseBody(resp, body)

3
keystone_tempest_plugin/tests/base.py
View File

@ -42,3 +42,6 @@ class BaseIdentityTest(test.BaseTestCase):
cls.tokens_client = cls.keystone_manager.token_v3_client
cls.consumers_client = cls.keystone_manager.oauth_consumers_client
cls.oauth_token_client = cls.keystone_manager.oauth_token_client
cls.registered_limits_client = (
cls.keystone_manager.registered_limits_client)
cls.limits_client = cls.keystone_manager.limits_client