Merge "Implemetation of tempest client for keystone v3 OS_FEDERATION API"

This commit is contained in:
Zuul 2021-02-11 22:46:18 +00:00 committed by Gerrit Code Review
commit 42353c5d4d
9 changed files with 1002 additions and 0 deletions

View File

@ -0,0 +1,10 @@
---
features:
- |
The following tempest clients for keystone v3 OS_FEDERATION API were
implemented in this release
* identity_providers
* protocols
* mappings
* service_providers

View File

@ -0,0 +1,92 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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 oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
class IdentityProvidersClient(rest_client.RestClient):
def register_identity_provider(self, identity_provider_id, **kwargs):
"""Register an identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#register-an-identity-provider
"""
post_body = json.dumps({'identity_provider': kwargs})
resp, body = self.put(
'OS-FEDERATION/identity_providers/%s' % identity_provider_id,
post_body)
self.expected_success(201, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def list_identity_providers(self, **params):
"""List identity providers.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#list-identity-providers
"""
url = 'identity_providers'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_identity_provider(self, identity_provider_id):
"""Get identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#get-identity-provider
"""
resp, body = self.get(
'OS-FEDERATION/identity_providers/%s' % identity_provider_id)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_identity_provider(self, identity_provider_id):
"""Delete identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#delete-identity-provider
"""
resp, body = self.delete(
'OS-FEDERATION/identity_providers/%s' % identity_provider_id)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)
def update_identity_provider(self, identity_provider_id, **kwargs):
"""Update identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#update-identity-provider
"""
post_body = json.dumps({'identity_provider': kwargs})
resp, body = self.patch(
'OS-FEDERATION/identity_providers/%s' % identity_provider_id,
post_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)

View File

@ -0,0 +1,90 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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 oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
class MappingsClient(rest_client.RestClient):
def create_mapping(self, mapping_id, **kwargs):
"""Create a mapping.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#create-a-mapping
"""
post_body = json.dumps({'mapping': kwargs})
resp, body = self.put(
'OS-FEDERATION/mappings/%s' % mapping_id, post_body)
self.expected_success(201, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_mapping(self, mapping_id):
"""Get a mapping.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#get-a-mapping
"""
resp, body = self.get(
'OS-FEDERATION/mappings/%s' % mapping_id)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def update_mapping(self, mapping_id, **kwargs):
"""Update a mapping.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#update-a-mapping
"""
post_body = json.dumps({'mapping': kwargs})
resp, body = self.patch(
'OS-FEDERATION/mappings/%s' % mapping_id, post_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def list_mappings(self, **kwargs):
"""List mappings.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#list-mappings
"""
url = 'OS-FEDERATION/mappings'
if kwargs:
url += '?%s' % urllib.urlencode(kwargs)
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_mapping(self, mapping_id):
"""Delete a mapping.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#delete-a-mapping
"""
resp, body = self.delete(
'OS-FEDERATION/mappings/%s' % mapping_id)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -0,0 +1,96 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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 oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
class ProtocolsClient(rest_client.RestClient):
def add_protocol_to_identity_provider(self, idp_id, protocol_id,
**kwargs):
"""Add protocol to identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#add-protocol-to-identity-provider
"""
post_body = json.dumps({'protocol': kwargs})
resp, body = self.put(
'OS-FEDERATION/identity_providers/%s/protocols/%s'
% (idp_id, protocol_id), post_body)
self.expected_success(201, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def list_protocols_of_identity_provider(self, idp_id, **kwargs):
"""List protocols of identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#list-protocols-of-identity-provider
"""
url = 'OS-FEDERATION/identity_providers/%s/protocols' % idp_id
if kwargs:
url += '?%s' % urllib.urlencode(kwargs)
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_protocol_for_identity_provider(self, idp_id, protocol_id):
"""Get protocol for identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#get-protocol-for-identity-provider
"""
resp, body = self.get(
'OS-FEDERATION/identity_providers/%s/protocols/%s'
% (idp_id, protocol_id))
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def update_mapping_for_identity_provider(self, idp_id, protocol_id,
**kwargs):
"""Update attribute mapping for identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#update-attribute-mapping-for-identity-provider
"""
post_body = json.dumps({'protocol': kwargs})
resp, body = self.patch(
'OS-FEDERATION/identity_providers/%s/protocols/%s'
% (idp_id, protocol_id), post_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_protocol_from_identity_provider(self, idp_id, protocol_id):
"""Delete a protocol from identity provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#delete-a-protocol-from-identity-provider
"""
resp, body = self.delete(
'OS-FEDERATION/identity_providers/%s/protocols/%s'
% (idp_id, protocol_id))
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -0,0 +1,92 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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 oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
class ServiceProvidersClient(rest_client.RestClient):
def register_service_provider(self, service_provider_id, **kwargs):
"""Register a service provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#register-a-service-provider
"""
post_body = json.dumps({'service_provider': kwargs})
resp, body = self.put(
'OS-FEDERATION/service_providers/%s' % service_provider_id,
post_body)
self.expected_success(201, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def list_service_providers(self, **kwargs):
"""List service providers.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#list-service-providers
"""
url = 'OS-FEDERATION/service_providers'
if kwargs:
url += '?%s' % urllib.urlencode(kwargs)
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_service_provider(self, service_provider_id):
"""Get a service provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#get-service-provider
"""
resp, body = self.get(
'OS-FEDERATION/service_providers/%s' % service_provider_id)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_service_provider(self, service_provider_id):
"""Delete a service provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#delete-service-provider
"""
resp, body = self.delete(
'OS-FEDERATION/service_providers/%s' % service_provider_id)
self.expected_success(204, resp.status)
return rest_client.ResponseBody(resp, body)
def update_service_provider(self, service_provider_id, **kwargs):
"""Update a service provider.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/identity/v3-ext/index.html#update-service-provider
"""
post_body = json.dumps({'service_provider': kwargs})
resp, body = self.patch(
'OS-FEDERATION/service_providers/%s' % service_provider_id,
post_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)

View File

@ -0,0 +1,142 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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.lib.services.identity.v3 import identity_providers_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestIdentityProvidersClient(base.BaseServiceTest):
FAKE_IDENTITY_PROVIDERS_INFO = {
"identity_providers": [
{
"domain_id": "FAKE_DOMAIN_ID",
"description": "FAKE IDENTITY PROVIDER",
"remote_ids": ["fake_id_1", "fake_id_2"],
"enabled": True,
"id": "FAKE_ID",
"links": {
"protocols": "http://example.com/identity/v3/" +
"OS-FEDERATION/identity_providers/" +
"FAKE_ID/protocols",
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID"
}
}
],
"links": {
"next": None,
"previous": None,
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers"
}
}
FAKE_IDENTITY_PROVIDER_INFO = {
"identity_provider": {
"authorization_ttl": None,
"domain_id": "FAKE_DOMAIN_ID",
"description": "FAKE IDENTITY PROVIDER",
"remote_ids": ["fake_id_1", "fake_id_2"],
"enabled": True,
"id": "ACME",
"links": {
"protocols": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID/protocols",
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID"
}
}
}
def setUp(self):
super(TestIdentityProvidersClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = identity_providers_client.IdentityProvidersClient(
fake_auth, 'identity', 'regionOne')
def _test_register_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.register_identity_provider,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_IDENTITY_PROVIDER_INFO,
bytes_body,
identity_provider_id="FAKE_ID",
status=201)
def _test_list_identity_providers(self, bytes_body=False):
self.check_service_client_function(
self.client.list_identity_providers,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_IDENTITY_PROVIDERS_INFO,
bytes_body,
status=200)
def _test_get_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.get_identity_provider,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_IDENTITY_PROVIDER_INFO,
bytes_body,
identity_provider_id="FAKE_ID",
status=200)
def _test_delete_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.delete_identity_provider,
'tempest.lib.common.rest_client.RestClient.delete',
{},
bytes_body,
identity_provider_id="FAKE_ID",
status=204)
def _test_update_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.update_identity_provider,
'tempest.lib.common.rest_client.RestClient.patch',
self.FAKE_IDENTITY_PROVIDER_INFO,
bytes_body,
identity_provider_id="FAKE_ID",
status=200)
def test_register_identity_provider_with_str_body(self):
self._test_register_identity_provider()
def test_register_identity_provider_with_bytes_body(self):
self._test_register_identity_provider(bytes_body=True)
def test_list_identity_providers_with_str_body(self):
self._test_list_identity_providers()
def test_list_identity_providers_with_bytes_body(self):
self._test_list_identity_providers(bytes_body=True)
def test_get_identity_provider_with_str_body(self):
self._test_get_identity_provider()
def test_get_identity_provider_with_bytes_body(self):
self._test_get_identity_provider(bytes_body=True)
def test_delete_identity_provider_with_str_body(self):
self._test_delete_identity_provider()
def test_delete_identity_provider_with_bytes_body(self):
self._test_delete_identity_provider(bytes_body=True)
def test_update_identity_provider_with_str_body(self):
self._test_update_identity_provider()
def test_update_identity_provider_with_bytes_body(self):
self._test_update_identity_provider(bytes_body=True)

View File

@ -0,0 +1,183 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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.lib.services.identity.v3 import mappings_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestMappingsClient(base.BaseServiceTest):
FAKE_MAPPING_INFO = {
"mapping": {
"id": "fake123",
"links": {
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"mappings/fake123"
},
"rules": [
{
"local": [
{
"user": {
"name": "{0}"
}
},
{
"group": {
"id": "0cd5e9"
}
}
],
"remote": [
{
"type": "UserName"
},
{
"type": "orgPersonType",
"not_any_of": [
"Contractor",
"Guest"
]
}
]
}
]
}
}
FAKE_MAPPINGS_INFO = {
"links": {
"next": None,
"previous": None,
"self": "http://example.com/identity/v3/OS-FEDERATION/mappings"
},
"mappings": [
{
"id": "fake123",
"links": {
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"mappings/fake123"
},
"rules": [
{
"local": [
{
"user": {
"name": "{0}"
}
},
{
"group": {
"id": "0cd5e9"
}
}
],
"remote": [
{
"type": "UserName"
},
{
"type": "orgPersonType",
"any_one_of": [
"Contractor",
"SubContractor"
]
}
]
}
]
}
]
}
def setUp(self):
super(TestMappingsClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = mappings_client.MappingsClient(
fake_auth, 'identity', 'regionOne')
def _test_create_mapping(self, bytes_body=False):
self.check_service_client_function(
self.client.create_mapping,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_MAPPING_INFO,
bytes_body,
mapping_id="fake123",
status=201)
def _test_get_mapping(self, bytes_body=False):
self.check_service_client_function(
self.client.get_mapping,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_MAPPING_INFO,
bytes_body,
mapping_id="fake123",
status=200)
def _test_update_mapping(self, bytes_body=False):
self.check_service_client_function(
self.client.update_mapping,
'tempest.lib.common.rest_client.RestClient.patch',
self.FAKE_MAPPING_INFO,
bytes_body,
mapping_id="fake123",
status=200)
def _test_list_mappings(self, bytes_body=False):
self.check_service_client_function(
self.client.list_mappings,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_MAPPINGS_INFO,
bytes_body,
status=200)
def _test_delete_mapping(self, bytes_body=False):
self.check_service_client_function(
self.client.delete_mapping,
'tempest.lib.common.rest_client.RestClient.delete',
{},
bytes_body,
mapping_id="fake123",
status=204)
def test_create_mapping_with_str_body(self):
self._test_create_mapping()
def test_create_mapping_with_bytes_body(self):
self._test_create_mapping(bytes_body=True)
def test_get_mapping_with_str_body(self):
self._test_get_mapping()
def test_get_mapping_with_bytes_body(self):
self._test_get_mapping(bytes_body=True)
def test_update_mapping_with_str_body(self):
self._test_update_mapping()
def test_update_mapping_with_bytes_body(self):
self._test_update_mapping(bytes_body=True)
def test_list_mappings_with_str_body(self):
self._test_list_mappings()
def test_list_mappings_with_bytes_body(self):
self._test_list_mappings(bytes_body=True)
def test_delete_mapping_with_str_body(self):
self._test_delete_mapping()
def test_delete_mapping_with_bytes_body(self):
self._test_delete_mapping(bytes_body=True)

View File

@ -0,0 +1,140 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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.lib.services.identity.v3 import protocols_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestProtocolsClient(base.BaseServiceTest):
FAKE_PROTOCOLS_INFO = {
"links": {
"next": None,
"previous": None,
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID/protocols"
},
"protocols": [
{
"id": "fake_id1",
"links": {
"identity_provider": "http://example.com/identity/v3/" +
"OS-FEDERATION/identity_providers/" +
"FAKE_ID",
"self": "http://example.com/identity/v3/OS-FEDERATION/"
"identity_providers/FAKE_ID/protocols/fake_id1"
},
"mapping_id": "fake123"
}
]
}
FAKE_PROTOCOL_INFO = {
"protocol": {
"id": "fake_id1",
"links": {
"identity_provider": "http://example.com/identity/v3/OS-" +
"FEDERATION/identity_providers/FAKE_ID",
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID/protocols/fake_id1"
},
"mapping_id": "fake123"
}
}
def setUp(self):
super(TestProtocolsClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = protocols_client.ProtocolsClient(
fake_auth, 'identity', 'regionOne')
def _test_add_protocol_to_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.add_protocol_to_identity_provider,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_PROTOCOL_INFO,
bytes_body,
idp_id="FAKE_ID",
protocol_id="fake_id1",
status=201)
def _test_list_protocols_of_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.list_protocols_of_identity_provider,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_PROTOCOLS_INFO,
bytes_body,
idp_id="FAKE_ID",
status=200)
def _test_get_protocol_for_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.get_protocol_for_identity_provider,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_PROTOCOL_INFO,
bytes_body,
idp_id="FAKE_ID",
protocol_id="fake_id1",
status=200)
def _test_update_mapping_for_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.update_mapping_for_identity_provider,
'tempest.lib.common.rest_client.RestClient.patch',
self.FAKE_PROTOCOL_INFO,
bytes_body,
idp_id="FAKE_ID",
protocol_id="fake_id1",
status=200)
def _test_delete_protocol_from_identity_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.delete_protocol_from_identity_provider,
'tempest.lib.common.rest_client.RestClient.delete',
{},
bytes_body,
idp_id="FAKE_ID",
protocol_id="fake_id1",
status=204)
def test_add_protocol_to_identity_provider_with_str_body(self):
self._test_add_protocol_to_identity_provider()
def test_add_protocol_to_identity_provider_with_bytes_body(self):
self._test_add_protocol_to_identity_provider(bytes_body=True)
def test_list_protocols_of_identity_provider_with_str_body(self):
self._test_list_protocols_of_identity_provider()
def test_list_protocols_of_identity_provider_with_bytes_body(self):
self._test_list_protocols_of_identity_provider(bytes_body=True)
def test_get_protocol_for_identity_provider_with_str_body(self):
self._test_get_protocol_for_identity_provider()
def test_get_protocol_for_identity_provider_with_bytes_body(self):
self._test_get_protocol_for_identity_provider(bytes_body=True)
def test_update_mapping_for_identity_provider_with_str_body(self):
self._test_update_mapping_for_identity_provider()
def test_update_mapping_for_identity_provider_with_bytes_body(self):
self._test_update_mapping_for_identity_provider(bytes_body=True)
def test_delete_protocol_from_identity_provider_with_str_body(self):
self._test_delete_protocol_from_identity_provider()
def test_delete_protocol_from_identity_provider_with_bytes_body(self):
self._test_delete_protocol_from_identity_provider(bytes_body=False)

View File

@ -0,0 +1,157 @@
# Copyright 2020 Samsung Electronics Co., Ltd
# 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.lib.services.identity.v3 import service_providers_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestServiceProvidersClient(base.BaseServiceTest):
FAKE_SERVICE_PROVIDER_INFO = {
"service_provider": {
"auth_url": "https://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/FAKE_ID/protocols/fake_id1/auth",
"description": "Fake Service Provider",
"enabled": True,
"id": "FAKE_ID",
"links": {
"self": "https://example.com/identity/v3/OS-FEDERATION/" +
"service_providers/FAKE_ID"
},
"relay_state_prefix": "ss:mem:",
"sp_url": "https://example.com/identity/Shibboleth.sso/" +
"FAKE_ID1/ECP"
}
}
FAKE_SERVICE_PROVIDERS_INFO = {
"links": {
"next": None,
"previous": None,
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"service_providers"
},
"service_providers": [
{
"auth_url": "https://example.com/identity/v3/OS-FEDERATION/" +
"identity_providers/acme/protocols/saml2/auth",
"description": "Stores ACME identities",
"enabled": True,
"id": "ACME",
"links": {
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"service_providers/ACME"
},
"relay_state_prefix": "ss:mem:",
"sp_url": "https://example.com/identity/Shibboleth.sso/" +
"SAML2/ECP"
},
{
"auth_url": "https://other.example.com/identity/v3/" +
"OS-FEDERATION/identity_providers/acme/" +
"protocols/saml2/auth",
"description": "Stores contractor identities",
"enabled": False,
"id": "ACME-contractors",
"links": {
"self": "http://example.com/identity/v3/OS-FEDERATION/" +
"service_providers/ACME-contractors"
},
"relay_state_prefix": "ss:mem:",
"sp_url": "https://other.example.com/identity/Shibboleth" +
".sso/SAML2/ECP"
}
]
}
def setUp(self):
super(TestServiceProvidersClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = service_providers_client.ServiceProvidersClient(
fake_auth, 'identity', 'regionOne')
def _test_register_service_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.register_service_provider,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_SERVICE_PROVIDER_INFO,
bytes_body,
service_provider_id="FAKE_ID",
status=201)
def _test_list_service_providers(self, bytes_body=False):
self.check_service_client_function(
self.client.list_service_providers,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_SERVICE_PROVIDERS_INFO,
bytes_body,
status=200)
def _test_get_service_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.get_service_provider,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_SERVICE_PROVIDER_INFO,
bytes_body,
service_provider_id="FAKE_ID",
status=200)
def _test_delete_service_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.delete_service_provider,
'tempest.lib.common.rest_client.RestClient.delete',
{},
bytes_body,
service_provider_id="FAKE_ID",
status=204)
def _test_update_service_provider(self, bytes_body=False):
self.check_service_client_function(
self.client.update_service_provider,
'tempest.lib.common.rest_client.RestClient.patch',
self.FAKE_SERVICE_PROVIDER_INFO,
bytes_body,
service_provider_id="FAKE_ID",
status=200)
def test_register_service_provider_with_str_body(self):
self._test_register_service_provider()
def test_register_service_provider_with_bytes_body(self):
self._test_register_service_provider(bytes_body=True)
def test_list_service_providers_with_str_body(self):
self._test_list_service_providers()
def test_list_service_providers_with_bytes_body(self):
self._test_list_service_providers(bytes_body=True)
def test_get_service_provider_with_str_body(self):
self._test_get_service_provider()
def test_get_service_provider_with_bytes_body(self):
self._test_get_service_provider(bytes_body=True)
def test_delete_service_provider_with_str_body(self):
self._test_delete_service_provider()
def test_delete_service_provider_with_bytes_body(self):
self._test_delete_service_provider(bytes_body=True)
def test_update_service_provider_with_str_body(self):
self._test_update_service_provider()
def test_update_service_provider_with_bytes_body(self):
self._test_update_service_provider(bytes_body=True)