path: root/keystoneclient/tests/unit/v3/test_federation.py

# 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 copy
import fixtures
import uuid

from keystoneauth1 import exceptions
from keystoneauth1 import fixture
from keystoneauth1.identity import v3
from keystoneauth1 import session
from keystoneauth1.tests.unit import k2k_fixtures
import six
from testtools import matchers

from keystoneclient import access
from keystoneclient.tests.unit.v3 import utils
from keystoneclient.v3 import client
from keystoneclient.v3.contrib.federation import base
from keystoneclient.v3.contrib.federation import identity_providers
from keystoneclient.v3.contrib.federation import mappings
from keystoneclient.v3.contrib.federation import protocols
from keystoneclient.v3.contrib.federation import service_providers
from keystoneclient.v3 import domains
from keystoneclient.v3 import projects


class IdentityProviderTests(utils.ClientTestCase, utils.CrudTests):
    def setUp(self):
        super(IdentityProviderTests, self).setUp()
        self.key = 'identity_provider'
        self.collection_key = 'identity_providers'
        self.model = identity_providers.IdentityProvider
        self.manager = self.client.federation.identity_providers
        self.path_prefix = 'OS-FEDERATION'

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('description', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        return kwargs

    def test_positional_parameters_expect_fail(self):
        """Ensure CrudManager raises TypeError exceptions.

        After passing wrong number of positional arguments
        an exception should be raised.

        Operations to be tested:
            * create()
            * get()
            * list()
            * delete()
            * update()

        """
        POS_PARAM_1 = uuid.uuid4().hex
        POS_PARAM_2 = uuid.uuid4().hex
        POS_PARAM_3 = uuid.uuid4().hex

        PARAMETERS = {
            'create': (POS_PARAM_1, POS_PARAM_2),
            'get': (POS_PARAM_1, POS_PARAM_2),
            'list': (POS_PARAM_1, POS_PARAM_2),
            'update': (POS_PARAM_1, POS_PARAM_2, POS_PARAM_3),
            'delete': (POS_PARAM_1, POS_PARAM_2)
        }

        for f_name, args in PARAMETERS.items():
            self.assertRaises(TypeError, getattr(self.manager, f_name),
                              *args)

    def test_create(self):
        ref = self.new_ref()
        req_ref = ref.copy()
        req_ref.pop('id')

        self.stub_entity('PUT', entity=ref, id=ref['id'], status_code=201)

        returned = self.manager.create(**ref)
        self.assertIsInstance(returned, self.model)
        for attr in req_ref:
            self.assertEqual(
                getattr(returned, attr),
                req_ref[attr],
                'Expected different %s' % attr)
        self.assertEntityRequestBodyIs(req_ref)


class MappingTests(utils.ClientTestCase, utils.CrudTests):
    def setUp(self):
        super(MappingTests, self).setUp()
        self.key = 'mapping'
        self.collection_key = 'mappings'
        self.model = mappings.Mapping
        self.manager = self.client.federation.mappings
        self.path_prefix = 'OS-FEDERATION'

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('rules', [uuid.uuid4().hex,
                                    uuid.uuid4().hex])
        return kwargs

    def test_create(self):
        ref = self.new_ref()
        manager_ref = ref.copy()
        mapping_id = manager_ref.pop('id')
        req_ref = ref.copy()

        self.stub_entity('PUT', entity=req_ref, id=mapping_id,
                         status_code=201)

        returned = self.manager.create(mapping_id=mapping_id, **manager_ref)
        self.assertIsInstance(returned, self.model)
        for attr in req_ref:
            self.assertEqual(
                getattr(returned, attr),
                req_ref[attr],
                'Expected different %s' % attr)
        self.assertEntityRequestBodyIs(manager_ref)


class ProtocolTests(utils.ClientTestCase, utils.CrudTests):
    def setUp(self):
        super(ProtocolTests, self).setUp()
        self.key = 'protocol'
        self.collection_key = 'protocols'
        self.model = protocols.Protocol
        self.manager = self.client.federation.protocols
        self.path_prefix = 'OS-FEDERATION/identity_providers'

    def _transform_to_response(self, ref):
        """Construct a response body from a dictionary."""
        response = copy.deepcopy(ref)
        del response['identity_provider']
        return response

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('mapping_id', uuid.uuid4().hex)
        kwargs.setdefault('identity_provider', uuid.uuid4().hex)
        return kwargs

    def build_parts(self, idp_id, protocol_id=None):
        """Build array used to construct mocking URL.

        Construct and return array with URL parts later used
        by methods like utils.TestCase.stub_entity().
        Example of URL:
        ``OS-FEDERATION/identity_providers/{idp_id}/
        protocols/{protocol_id}``

        """
        parts = ['OS-FEDERATION', 'identity_providers',
                 idp_id, 'protocols']
        if protocol_id:
            parts.append(protocol_id)
        return parts

    def test_build_url_provide_base_url(self):
        base_url = uuid.uuid4().hex
        parameters = {'base_url': base_url}
        url = self.manager.build_url(dict_args_in_out=parameters)
        self.assertEqual('/'.join([base_url, self.collection_key]), url)

    def test_build_url_w_idp_id(self):
        """Test whether kwargs ``base_url`` discards object's base_url.

        This test shows, that when ``base_url`` is specified in the
        dict_args_in_out dictionary,  values like ``identity_provider_id``
        are not taken into consideration while building the url.

        """
        base_url, identity_provider_id = uuid.uuid4().hex, uuid.uuid4().hex
        parameters = {
            'base_url': base_url,
            'identity_provider_id': identity_provider_id
        }
        url = self.manager.build_url(dict_args_in_out=parameters)
        self.assertEqual('/'.join([base_url, self.collection_key]), url)

    def test_build_url_default_base_url(self):
        identity_provider_id = uuid.uuid4().hex
        parameters = {
            'identity_provider_id': identity_provider_id
        }

        url = self.manager.build_url(dict_args_in_out=parameters)
        self.assertEqual(
            '/'.join([self.manager.base_url, identity_provider_id,
                      self.manager.collection_key]), url)

    def test_create(self):
        """Test creating federation protocol tied to an Identity Provider.

        URL to be tested: PUT /OS-FEDERATION/identity_providers/
        $identity_provider/protocols/$protocol

        """
        ref = self.new_ref()
        expected = self._transform_to_response(ref)
        parts = self.build_parts(
            idp_id=ref['identity_provider'],
            protocol_id=ref['id'])
        self.stub_entity('PUT', entity=expected,
                         parts=parts, status_code=201)
        returned = self.manager.create(
            protocol_id=ref['id'],
            identity_provider=ref['identity_provider'],
            mapping=ref['mapping_id'])
        self.assertEqual(expected, returned.to_dict())
        request_body = {'mapping_id': ref['mapping_id']}
        self.assertEntityRequestBodyIs(request_body)

    def test_get(self):
        """Fetch federation protocol object.

        URL to be tested: GET /OS-FEDERATION/identity_providers/
        $identity_provider/protocols/$protocol

        """
        ref = self.new_ref()
        expected = self._transform_to_response(ref)

        parts = self.build_parts(
            idp_id=ref['identity_provider'],
            protocol_id=ref['id'])
        self.stub_entity('GET', entity=expected,
                         parts=parts, status_code=201)

        returned = self.manager.get(ref['identity_provider'],
                                    ref['id'])
        self.assertIsInstance(returned, self.model)
        self.assertEqual(expected, returned.to_dict())

    def test_delete(self):
        """Delete federation protocol object.

        URL to be tested: DELETE /OS-FEDERATION/identity_providers/
        $identity_provider/protocols/$protocol

        """
        ref = self.new_ref()
        parts = self.build_parts(
            idp_id=ref['identity_provider'],
            protocol_id=ref['id'])

        self.stub_entity('DELETE', parts=parts, status_code=204)

        self.manager.delete(ref['identity_provider'],
                            ref['id'])

    def test_list(self):
        """Test listing all federation protocols tied to the Identity Provider.

        URL to be tested: GET /OS-FEDERATION/identity_providers/
        $identity_provider/protocols

        """
        def _ref_protocols():
            return {
                'id': uuid.uuid4().hex,
                'mapping_id': uuid.uuid4().hex
            }

        ref = self.new_ref()
        expected = [_ref_protocols() for _ in range(3)]
        parts = self.build_parts(idp_id=ref['identity_provider'])
        self.stub_entity('GET', parts=parts,
                         entity=expected, status_code=200)

        returned = self.manager.list(ref['identity_provider'])
        for obj, ref_obj in zip(returned, expected):
            self.assertEqual(obj.to_dict(), ref_obj)

    def test_list_by_id(self):
        # The test in the parent class needs to be overridden because it
        # assumes globally unique IDs, which is not the case with protocol IDs
        # (which are contextualized per identity provider).
        ref = self.new_ref()
        super(ProtocolTests, self).test_list_by_id(
            ref=ref,
            identity_provider=ref['identity_provider'],
            id=ref['id'])

    def test_list_params(self):
        request_args = self.new_ref()
        filter_kwargs = {uuid.uuid4().hex: uuid.uuid4().hex}
        parts = self.build_parts(request_args['identity_provider'])

        # Return HTTP 401 as we don't accept such requests.
        self.stub_entity('GET', parts=parts, status_code=401)
        self.assertRaises(exceptions.Unauthorized,
                          self.manager.list,
                          request_args['identity_provider'],
                          **filter_kwargs)
        self.assertQueryStringContains(**filter_kwargs)

    def test_update(self):
        """Test updating federation protocol.

        URL to be tested: PATCH /OS-FEDERATION/identity_providers/
        $identity_provider/protocols/$protocol

        """
        ref = self.new_ref()
        expected = self._transform_to_response(ref)

        parts = self.build_parts(
            idp_id=ref['identity_provider'],
            protocol_id=ref['id'])

        self.stub_entity('PATCH', parts=parts,
                         entity=expected, status_code=200)

        returned = self.manager.update(ref['identity_provider'],
                                       ref['id'],
                                       mapping=ref['mapping_id'])
        self.assertIsInstance(returned, self.model)
        self.assertEqual(expected, returned.to_dict())
        request_body = {'mapping_id': ref['mapping_id']}
        self.assertEntityRequestBodyIs(request_body)


class EntityManagerTests(utils.ClientTestCase):
    def test_create_object_expect_fail(self):
        self.assertRaises(TypeError,
                          base.EntityManager,
                          self.client)


class FederationProjectTests(utils.ClientTestCase):

    def setUp(self):
        super(FederationProjectTests, self).setUp()
        self.key = 'project'
        self.collection_key = 'projects'
        self.model = projects.Project
        self.manager = self.client.federation.projects
        self.URL = "%s%s" % (self.TEST_URL, '/auth/projects')

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('domain_id', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        kwargs.setdefault('name', uuid.uuid4().hex)
        return kwargs

    def test_list_accessible_projects(self):
        projects_ref = [self.new_ref(), self.new_ref()]
        projects_json = {
            self.collection_key: [self.new_ref(), self.new_ref()]
        }
        self.requests_mock.get(self.URL, json=projects_json)
        returned_list = self.manager.list()

        self.assertEqual(len(projects_ref), len(returned_list))
        for project in returned_list:
            self.assertIsInstance(project, self.model)


class K2KFederatedProjectTests(utils.TestCase):

    TEST_ROOT_URL = 'http://127.0.0.1:5000/'
    TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
    TEST_PASS = 'password'
    REQUEST_ECP_URL = TEST_URL + '/auth/OS-FEDERATION/saml2/ecp'

    SP_ID = 'sp1'
    SP_ROOT_URL = 'https://example.com/v3'
    SP_URL = 'https://example.com/Shibboleth.sso/SAML2/ECP'
    SP_AUTH_URL = (SP_ROOT_URL +
                   '/OS-FEDERATION/identity_providers'
                   '/testidp/protocols/saml2/auth')

    def setUp(self):
        super(K2KFederatedProjectTests, self).setUp()
        self.token_v3 = fixture.V3Token()
        self.token_v3.add_service_provider(
            self.SP_ID, self.SP_AUTH_URL, self.SP_URL)
        self.session = session.Session()
        self.collection_key = 'projects'
        self.model = projects.Project
        self.URL = '%s%s' % (self.SP_ROOT_URL, '/auth/projects')
        self.k2kplugin = self.get_plugin()
        self._mock_k2k_flow_urls()

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('domain_id', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        kwargs.setdefault('name', uuid.uuid4().hex)
        return kwargs

    def _get_base_plugin(self):
        self.stub_url('POST', ['auth', 'tokens'],
                      headers={'X-Subject-Token': uuid.uuid4().hex},
                      json=self.token_v3)
        return v3.Password(self.TEST_URL,
                           username=self.TEST_USER,
                           password=self.TEST_PASS)

    def _mock_k2k_flow_urls(self):
        # We need to check the auth versions available
        self.requests_mock.get(
            self.TEST_URL,
            json={'version': fixture.V3Discovery(self.TEST_URL)},
            headers={'Content-Type': 'application/json'})

        # The identity provider receives a request for an ECP wrapped
        # assertion. This assertion contains the user authentication info
        # and will be presented to the service provider
        self.requests_mock.register_uri(
            'POST',
            self.REQUEST_ECP_URL,
            content=six.b(k2k_fixtures.ECP_ENVELOPE),
            headers={'Content-Type': 'application/vnd.paos+xml'},
            status_code=200)

        # The service provider is presented with the ECP wrapped assertion
        # generated by the identity provider and should return a redirect
        # (302 or 303) upon successful authentication
        self.requests_mock.register_uri(
            'POST',
            self.SP_URL,
            content=six.b(k2k_fixtures.TOKEN_BASED_ECP),
            headers={'Content-Type': 'application/vnd.paos+xml'},
            status_code=302)

        # Should not follow the redirect URL, but use the auth_url attribute
        self.requests_mock.register_uri(
            'GET',
            self.SP_AUTH_URL,
            json=k2k_fixtures.UNSCOPED_TOKEN,
            headers={'X-Subject-Token': k2k_fixtures.UNSCOPED_TOKEN_HEADER})

    def get_plugin(self, **kwargs):
        kwargs.setdefault('base_plugin', self._get_base_plugin())
        kwargs.setdefault('service_provider', self.SP_ID)
        return v3.Keystone2Keystone(**kwargs)

    def test_list_projects(self):
        k2k_client = client.Client(session=self.session, auth=self.k2kplugin)
        self.requests_mock.get(self.URL, json={
            self.collection_key: [self.new_ref(), self.new_ref()]
        })
        self.requests_mock.get(self.SP_ROOT_URL, json={
            'version': fixture.discovery.V3Discovery(self.SP_ROOT_URL)
        })
        returned_list = k2k_client.federation.projects.list()

        self.assertThat(returned_list, matchers.HasLength(2))
        for project in returned_list:
            self.assertIsInstance(project, self.model)


class FederationDomainTests(utils.ClientTestCase):

    def setUp(self):
        super(FederationDomainTests, self).setUp()
        self.key = 'domain'
        self.collection_key = 'domains'
        self.model = domains.Domain
        self.manager = self.client.federation.domains

        self.URL = "%s%s" % (self.TEST_URL, '/auth/domains')

    def new_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        kwargs.setdefault('name', uuid.uuid4().hex)
        kwargs.setdefault('description', uuid.uuid4().hex)
        return kwargs

    def test_list_accessible_domains(self):
        domains_ref = [self.new_ref(), self.new_ref()]
        domains_json = {
            self.collection_key: domains_ref
        }
        self.requests_mock.get(self.URL, json=domains_json)
        returned_list = self.manager.list()
        self.assertEqual(len(domains_ref), len(returned_list))
        for domain in returned_list:
            self.assertIsInstance(domain, self.model)


class FederatedTokenTests(utils.ClientTestCase):

    def setUp(self):
        super(FederatedTokenTests, self).setUp()
        token = fixture.V3FederationToken()
        token.set_project_scope()
        token.add_role()
        self.federated_token = access.AccessInfo.factory(body=token)

    def test_federated_property_federated_token(self):
        """Check if is_federated property returns expected value."""
        self.assertTrue(self.federated_token.is_federated)

    def test_get_user_domain_name(self):
        """Ensure a federated user's domain name does not exist."""
        self.assertIsNone(self.federated_token.user_domain_name)

    def test_get_user_domain_id(self):
        """Ensure a federated user's domain ID does not exist."""
        self.assertEqual('Federated', self.federated_token.user_domain_id)


class ServiceProviderTests(utils.ClientTestCase, utils.CrudTests):
    def setUp(self):
        super(ServiceProviderTests, self).setUp()
        self.key = 'service_provider'
        self.collection_key = 'service_providers'
        self.model = service_providers.ServiceProvider
        self.manager = self.client.federation.service_providers
        self.path_prefix = 'OS-FEDERATION'

    def new_ref(self, **kwargs):
        kwargs.setdefault('auth_url', uuid.uuid4().hex)
        kwargs.setdefault('description', uuid.uuid4().hex)
        kwargs.setdefault('enabled', True)
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('sp_url', uuid.uuid4().hex)
        return kwargs

    def test_positional_parameters_expect_fail(self):
        """Ensure CrudManager raises TypeError exceptions.

        After passing wrong number of positional arguments
        an exception should be raised.

        Operations to be tested:
            * create()
            * get()
            * list()
            * delete()
            * update()

        """
        POS_PARAM_1 = uuid.uuid4().hex
        POS_PARAM_2 = uuid.uuid4().hex
        POS_PARAM_3 = uuid.uuid4().hex

        PARAMETERS = {
            'create': (POS_PARAM_1, POS_PARAM_2),
            'get': (POS_PARAM_1, POS_PARAM_2),
            'list': (POS_PARAM_1, POS_PARAM_2),
            'update': (POS_PARAM_1, POS_PARAM_2, POS_PARAM_3),
            'delete': (POS_PARAM_1, POS_PARAM_2)
        }

        for f_name, args in PARAMETERS.items():
            self.assertRaises(TypeError, getattr(self.manager, f_name),
                              *args)

    def test_create(self):
        ref = self.new_ref()

        # req_ref argument allows you to specify a different
        # signature for the request when the manager does some
        # conversion before doing the request (e.g. converting
        # from datetime object to timestamp string)
        req_ref = ref.copy()
        req_ref.pop('id')

        self.stub_entity('PUT', entity=ref, id=ref['id'], status_code=201)

        returned = self.manager.create(**ref)
        self.assertIsInstance(returned, self.model)
        for attr in req_ref:
            self.assertEqual(
                getattr(returned, attr),
                req_ref[attr],
                'Expected different %s' % attr)
        self.assertEntityRequestBodyIs(req_ref)


class IdentityProviderRequestIdTests(utils.TestRequestId):

    def setUp(self):
        super(IdentityProviderRequestIdTests, self).setUp()
        self.mgr = identity_providers.IdentityProviderManager(self.client)

    def _mock_request_method(self, method=None, body=None):
        return self.useFixture(fixtures.MockPatchObject(
            self.client, method, autospec=True,
            return_value=(self.resp, body))
        ).mock

    def test_get_identity_provider(self):
        body = {"identity_provider": {"name": "admin"}}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.get("admin")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/admin')

    def test_list_identity_provider(self):
        body = {"identity_providers": [{"name": "admin"}]}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.list()
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with('OS-FEDERATION/identity_providers?')

    def test_create_identity_provider(self):
        body = {"identity_provider": {"name": "admin"}}
        self._mock_request_method(method='post', body=body)
        put_mock = self._mock_request_method(method='put', body=body)

        response = self.mgr.create(id="admin", description='fake')
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        put_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/admin',
            body={'identity_provider': {'description': 'fake'}})

    def test_update_identity_provider(self):
        body = {"identity_provider": {"name": "admin"}}
        patch_mock = self._mock_request_method(method='patch', body=body)
        self._mock_request_method(method='post', body=body)

        response = self.mgr.update("admin")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        patch_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/admin', body={
                'identity_provider': {}})

    def test_delete_identity_provider(self):
        get_mock = self._mock_request_method(method='delete')

        _, resp = self.mgr.delete("admin")
        self.assertEqual(resp.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/admin')


class MappingRequestIdTests(utils.TestRequestId):

    def setUp(self):
        super(MappingRequestIdTests, self).setUp()
        self.mgr = mappings.MappingManager(self.client)

    def _mock_request_method(self, method=None, body=None):
        return self.useFixture(fixtures.MockPatchObject(
            self.client, method, autospec=True,
            return_value=(self.resp, body))
        ).mock

    def test_get_mapping(self):
        body = {"mapping": {"name": "admin"}}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.get("admin")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with('OS-FEDERATION/mappings/admin')

    def test_list_mapping(self):
        body = {"mappings": [{"name": "admin"}]}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.list()
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with('OS-FEDERATION/mappings?')

    def test_create_mapping(self):
        body = {"mapping": {"name": "admin"}}
        self._mock_request_method(method='post', body=body)
        put_mock = self._mock_request_method(method='put', body=body)

        response = self.mgr.create(mapping_id="admin", description='fake')
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        put_mock.assert_called_once_with(
            'OS-FEDERATION/mappings/admin', body={
                'mapping': {'description': 'fake'}})

    def test_update_mapping(self):
        body = {"mapping": {"name": "admin"}}
        patch_mock = self._mock_request_method(method='patch', body=body)
        self._mock_request_method(method='post', body=body)

        response = self.mgr.update("admin")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        patch_mock.assert_called_once_with(
            'OS-FEDERATION/mappings/admin', body={'mapping': {}})

    def test_delete_mapping(self):
        get_mock = self._mock_request_method(method='delete')

        _, resp = self.mgr.delete("admin")
        self.assertEqual(resp.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with('OS-FEDERATION/mappings/admin')


class ProtocolRequestIdTests(utils.TestRequestId):

    def setUp(self):
        super(ProtocolRequestIdTests, self).setUp()
        self.mgr = protocols.ProtocolManager(self.client)

    def _mock_request_method(self, method=None, body=None):
        return self.useFixture(fixtures.MockPatchObject(
            self.client, method, autospec=True,
            return_value=(self.resp, body))
        ).mock

    def test_get_protocol(self):
        body = {"protocol": {"name": "admin"}}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.get("admin", "protocol")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/admin/protocols/protocol')

    def test_list_protocol(self):
        body = {"protocols": [{"name": "admin"}]}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.list("identity_provider")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/identity_provider/protocols?')

    def test_create_protocol(self):
        body = {"protocol": {"name": "admin"}}
        self._mock_request_method(method='post', body=body)
        put_mock = self._mock_request_method(method='put', body=body)

        response = self.mgr.create(
            protocol_id="admin", identity_provider='fake', mapping='fake')
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        put_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/fake/protocols/admin', body={
                'protocol': {'mapping_id': 'fake'}})

    def test_update_protocol(self):
        body = {"protocol": {"name": "admin"}}
        patch_mock = self._mock_request_method(method='patch', body=body)
        self._mock_request_method(method='post', body=body)

        response = self.mgr.update(protocol="admin", identity_provider='fake',
                                   mapping='fake')
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        patch_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/fake/protocols/admin', body={
                'protocol': {'mapping_id': 'fake'}})

    def test_delete_protocol(self):
        get_mock = self._mock_request_method(method='delete')

        _, resp = self.mgr.delete("identity_provider", "protocol")
        self.assertEqual(resp.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/identity_providers/'
            'identity_provider/protocols/protocol')


class ServiceProviderRequestIdTests(utils.TestRequestId):

    def setUp(self):
        super(ServiceProviderRequestIdTests, self).setUp()
        self.mgr = service_providers.ServiceProviderManager(self.client)

    def _mock_request_method(self, method=None, body=None):
        return self.useFixture(fixtures.MockPatchObject(
            self.client, method, autospec=True,
            return_value=(self.resp, body))
        ).mock

    def test_get_service_provider(self):
        body = {"service_provider": {"name": "admin"}}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.get("provider")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/service_providers/provider')

    def test_list_service_provider(self):
        body = {"service_providers": [{"name": "admin"}]}
        get_mock = self._mock_request_method(method='get', body=body)

        response = self.mgr.list()
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with('OS-FEDERATION/service_providers?')

    def test_create_service_provider(self):
        body = {"service_provider": {"name": "admin"}}
        self._mock_request_method(method='post', body=body)
        put_mock = self._mock_request_method(method='put', body=body)

        response = self.mgr.create(id='provider')
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        put_mock.assert_called_once_with(
            'OS-FEDERATION/service_providers/provider', body={
                'service_provider': {}})

    def test_update_service_provider(self):
        body = {"service_provider": {"name": "admin"}}
        patch_mock = self._mock_request_method(method='patch', body=body)
        self._mock_request_method(method='post', body=body)

        response = self.mgr.update("provider")
        self.assertEqual(response.request_ids[0], self.TEST_REQUEST_ID)
        patch_mock.assert_called_once_with(
            'OS-FEDERATION/service_providers/provider', body={
                'service_provider': {}})

    def test_delete_service_provider(self):
        get_mock = self._mock_request_method(method='delete')

        _, resp = self.mgr.delete("provider")
        self.assertEqual(resp.request_ids[0], self.TEST_REQUEST_ID)
        get_mock.assert_called_once_with(
            'OS-FEDERATION/service_providers/provider')