FM / v3

/usr/lib/python2.6/site-packages/keystoneclient/tests/unit/v3/

/usr/lib/python2.6/site-packages/keystoneclient/tests/unit/v3

Name	Size	Mode	Actions
examples/	-	0755	rm
client_fixtures.py	5688	0644	edit dl rm
client_fixtures.pyc	5059	0644	edit dl rm
saml2_fixtures.py	10706	0644	edit dl rm
saml2_fixtures.pyc	9828	0644	edit dl rm
test_access.py	8841	0644	edit dl rm
test_access.pyc	6672	0644	edit dl rm
test_auth.py	14628	0644	edit dl rm
test_auth.pyc	10995	0644	edit dl rm
test_auth_manager.py	2435	0644	edit dl rm
test_auth_manager.pyc	2765	0644	edit dl rm
test_auth_oidc.py	7550	0644	edit dl rm
test_auth_oidc.pyc	5888	0644	edit dl rm
test_auth_saml2.py	28966	0644	edit dl rm
test_auth_saml2.pyc	27228	0644	edit dl rm
test_client.py	10167	0644	edit dl rm
test_client.pyc	8984	0644	edit dl rm
test_credentials.py	2107	0644	edit dl rm
test_credentials.pyc	2436	0644	edit dl rm
test_discover.py	3726	0644	edit dl rm
test_discover.pyc	2521	0644	edit dl rm
test_domains.py	1950	0644	edit dl rm
test_domains.pyc	2840	0644	edit dl rm
test_ec2.py	3740	0644	edit dl rm
test_ec2.pyc	3103	0644	edit dl rm
test_endpoints.py	3581	0644	edit dl rm
test_endpoints.pyc	5332	0644	edit dl rm
test_endpoint_filter.py	6041	0644	edit dl rm
test_endpoint_filter.pyc	5875	0644	edit dl rm
test_endpoint_policy.py	10583	0644	edit dl rm
test_endpoint_policy.pyc	12055	0644	edit dl rm
test_federation.py	17359	0644	edit dl rm
test_federation.pyc	20277	0644	edit dl rm
test_groups.py	2105	0644	edit dl rm
test_groups.pyc	2488	0644	edit dl rm
test_oauth1.py	11505	0644	edit dl rm
test_oauth1.pyc	11579	0644	edit dl rm
test_policies.py	1131	0644	edit dl rm
test_policies.pyc	1336	0644	edit dl rm
test_projects.py	11382	0644	edit dl rm
test_projects.pyc	9952	0644	edit dl rm
test_regions.py	1267	0644	edit dl rm
test_regions.pyc	1662	0644	edit dl rm
test_roles.py	20575	0644	edit dl rm
test_roles.pyc	20534	0644	edit dl rm
test_role_assignments.py	9111	0644	edit dl rm
test_role_assignments.pyc	8488	0644	edit dl rm
test_services.py	1701	0644	edit dl rm
test_services.pyc	2061	0644	edit dl rm
test_service_catalog.py	12560	0644	edit dl rm
test_service_catalog.pyc	10339	0644	edit dl rm
test_simple_cert.py	1611	0644	edit dl rm
test_simple_cert.pyc	1966	0644	edit dl rm
test_tokens.py	5353	0644	edit dl rm
test_tokens.pyc	5510	0644	edit dl rm
test_trusts.py	4298	0644	edit dl rm
test_trusts.pyc	5053	0644	edit dl rm
test_users.py	10411	0644	edit dl rm
test_users.pyc	8840	0644	edit dl rm
utils.py	11594	0644	edit dl rm
utils.pyc	10513	0644	edit dl rm
__init__.py	0	0644	edit dl rm
__init__.pyc	171	0644	edit dl rm

Edit: /usr/lib/python2.6/site-packages/keystoneclient/tests/unit/v3/test_endpoint_policy.py (10583B)

import uuid

from keystoneclient.tests.unit.v3 import test_endpoint_filter
from keystoneclient.tests.unit.v3 import utils

class EndpointPolicyTests(utils.TestCase,
                          test_endpoint_filter.EndpointTestUtils):
    """Test policy-endpoint associations (a.k.a. EndpointPolicy Extension)."""

def setUp(self):
        super(EndpointPolicyTests, self).setUp()
        self.manager = self.client.endpoint_policy

def new_policy_ref(self, **kwargs):
        kwargs.setdefault('id', uuid.uuid4().hex)
        kwargs.setdefault('type', uuid.uuid4().hex)
        kwargs.setdefault('blob', uuid.uuid4().hex)
        return kwargs

def new_region_ref(self, **kwargs):
        kwargs = self.new_ref(**kwargs)
        return kwargs

def new_service_ref(self, **kwargs):
        kwargs = self.new_ref(**kwargs)
        kwargs.setdefault('name', uuid.uuid4().hex)
        kwargs.setdefault('type', uuid.uuid4().hex)
        return kwargs

def _crud_policy_association_for_endpoint_via_id(
            self, http_action, manager_action):
        policy_id = uuid.uuid4().hex
        endpoint_id = uuid.uuid4().hex

self.stub_url(http_action,
                      ['policies', policy_id, self.manager.OS_EP_POLICY_EXT,
                       'endpoints', endpoint_id],
                      status_code=204)
        manager_action(policy=policy_id, endpoint=endpoint_id)

def _crud_policy_association_for_endpoint_via_obj(
            self, http_action, manager_action):
        policy_ref = self.new_policy_ref()
        endpoint_ref = self.new_endpoint_ref()
        policy = self.client.policies.resource_class(
            self.client.policies, policy_ref, loaded=True)
        endpoint = self.client.endpoints.resource_class(
            self.client.endpoints, endpoint_ref, loaded=True)

self.stub_url(http_action,
                      ['policies', policy_ref['id'],
                       self.manager.OS_EP_POLICY_EXT,
                       'endpoints', endpoint_ref['id']],
                      status_code=204)
        manager_action(policy=policy, endpoint=endpoint)

def test_create_policy_association_for_endpoint_via_id(self):
        self._crud_policy_association_for_endpoint_via_id(
            'PUT', self.manager.create_policy_association_for_endpoint)

def test_create_policy_association_for_endpoint_via_obj(self):
        self._crud_policy_association_for_endpoint_via_obj(
            'PUT', self.manager.create_policy_association_for_endpoint)

def test_check_policy_association_for_endpoint_via_id(self):
        self._crud_policy_association_for_endpoint_via_id(
            'HEAD', self.manager.check_policy_association_for_endpoint)

def test_check_policy_association_for_endpoint_via_obj(self):
        self._crud_policy_association_for_endpoint_via_obj(
            'HEAD', self.manager.check_policy_association_for_endpoint)

def test_delete_policy_association_for_endpoint_via_id(self):
        self._crud_policy_association_for_endpoint_via_id(
            'DELETE', self.manager.delete_policy_association_for_endpoint)

def test_delete_policy_association_for_endpoint_via_obj(self):
        self._crud_policy_association_for_endpoint_via_obj(
            'DELETE', self.manager.delete_policy_association_for_endpoint)

def _crud_policy_association_for_service_via_id(
            self, http_action, manager_action):
        policy_id = uuid.uuid4().hex
        service_id = uuid.uuid4().hex

self.stub_url(http_action,
                      ['policies', policy_id, self.manager.OS_EP_POLICY_EXT,
                       'services', service_id],
                      status_code=204)
        manager_action(policy=policy_id, service=service_id)

def _crud_policy_association_for_service_via_obj(
            self, http_action, manager_action):
        policy_ref = self.new_policy_ref()
        service_ref = self.new_service_ref()
        policy = self.client.policies.resource_class(
            self.client.policies, policy_ref, loaded=True)
        service = self.client.services.resource_class(
            self.client.services, service_ref, loaded=True)

self.stub_url(http_action,
                      ['policies', policy_ref['id'],
                       self.manager.OS_EP_POLICY_EXT,
                       'services', service_ref['id']],
                      status_code=204)
        manager_action(policy=policy, service=service)

def test_create_policy_association_for_service_via_id(self):
        self._crud_policy_association_for_service_via_id(
            'PUT', self.manager.create_policy_association_for_service)

def test_create_policy_association_for_service_via_obj(self):
        self._crud_policy_association_for_service_via_obj(
            'PUT', self.manager.create_policy_association_for_service)

def test_check_policy_association_for_service_via_id(self):
        self._crud_policy_association_for_service_via_id(
            'HEAD', self.manager.check_policy_association_for_service)

def test_check_policy_association_for_service_via_obj(self):
        self._crud_policy_association_for_service_via_obj(
            'HEAD', self.manager.check_policy_association_for_service)

def test_delete_policy_association_for_service_via_id(self):
        self._crud_policy_association_for_service_via_id(
            'DELETE', self.manager.delete_policy_association_for_service)

def test_delete_policy_association_for_service_via_obj(self):
        self._crud_policy_association_for_service_via_obj(
            'DELETE', self.manager.delete_policy_association_for_service)

def _crud_policy_association_for_region_and_service_via_id(
            self, http_action, manager_action):
        policy_id = uuid.uuid4().hex
        region_id = uuid.uuid4().hex
        service_id = uuid.uuid4().hex

self.stub_url(http_action,
                      ['policies', policy_id, self.manager.OS_EP_POLICY_EXT,
                       'services', service_id, 'regions', region_id],
                      status_code=204)
        manager_action(policy=policy_id, region=region_id, service=service_id)

def _crud_policy_association_for_region_and_service_via_obj(
            self, http_action, manager_action):
        policy_ref = self.new_policy_ref()
        region_ref = self.new_region_ref()
        service_ref = self.new_service_ref()
        policy = self.client.policies.resource_class(
            self.client.policies, policy_ref, loaded=True)
        region = self.client.regions.resource_class(
            self.client.regions, region_ref, loaded=True)
        service = self.client.services.resource_class(
            self.client.services, service_ref, loaded=True)

self.stub_url(http_action,
                      ['policies', policy_ref['id'],
                       self.manager.OS_EP_POLICY_EXT,
                       'services', service_ref['id'],
                       'regions', region_ref['id']],
                      status_code=204)
        manager_action(policy=policy, region=region, service=service)

def test_create_policy_association_for_region_and_service_via_id(self):
        self._crud_policy_association_for_region_and_service_via_id(
            'PUT',
            self.manager.create_policy_association_for_region_and_service)

def test_create_policy_association_for_region_and_service_via_obj(self):
        self._crud_policy_association_for_region_and_service_via_obj(
            'PUT',
            self.manager.create_policy_association_for_region_and_service)

def test_check_policy_association_for_region_and_service_via_id(self):
        self._crud_policy_association_for_region_and_service_via_id(
            'HEAD',
            self.manager.check_policy_association_for_region_and_service)

def test_check_policy_association_for_region_and_service_via_obj(self):
        self._crud_policy_association_for_region_and_service_via_obj(
            'HEAD',
            self.manager.check_policy_association_for_region_and_service)

def test_delete_policy_association_for_region_and_service_via_id(self):
        self._crud_policy_association_for_region_and_service_via_id(
            'DELETE',
            self.manager.delete_policy_association_for_region_and_service)

def test_delete_policy_association_for_region_and_service_via_obj(self):
        self._crud_policy_association_for_region_and_service_via_obj(
            'DELETE',
            self.manager.delete_policy_association_for_region_and_service)

def test_get_policy_for_endpoint(self):
        endpoint_id = uuid.uuid4().hex
        expected_policy = self.new_policy_ref()

self.stub_url('GET',
                      ['endpoints', endpoint_id, self.manager.OS_EP_POLICY_EXT,
                       'policy'],
                      json={'policy': expected_policy},
                      status_code=200)

policy_resp = self.manager.get_policy_for_endpoint(
            endpoint=endpoint_id)

self.assertEqual(expected_policy['id'], policy_resp.id)
        self.assertEqual(expected_policy['blob'], policy_resp.blob)
        self.assertEqual(expected_policy['type'], policy_resp.type)

def test_list_endpoints_for_policy(self):
        policy_id = uuid.uuid4().hex
        endpoints = {'endpoints': [self.new_endpoint_ref(),
                                   self.new_endpoint_ref()]}
        self.stub_url('GET',
                      ['policies', policy_id, self.manager.OS_EP_POLICY_EXT,
                       'endpoints'],
                      json=endpoints,
                      status_code=200)

endpoints_resp = self.manager.list_endpoints_for_policy(
            policy=policy_id)

expected_endpoint_ids = [
            endpoint['id'] for endpoint in endpoints['endpoints']]
        actual_endpoint_ids = [endpoint.id for endpoint in endpoints_resp]
        self.assertEqual(expected_endpoint_ids, actual_endpoint_ids)

# Copyright 2014 IBM Corp. # # 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 uuid from keystoneclient.tests.unit.v3 import test_endpoint_filter from keystoneclient.tests.unit.v3 import utils class EndpointPolicyTests(utils.TestCase, test_endpoint_filter.EndpointTestUtils): """Test policy-endpoint associations (a.k.a. EndpointPolicy Extension).""" def setUp(self): super(EndpointPolicyTests, self).setUp() self.manager = self.client.endpoint_policy def new_policy_ref(self, **kwargs): kwargs.setdefault('id', uuid.uuid4().hex) kwargs.setdefault('type', uuid.uuid4().hex) kwargs.setdefault('blob', uuid.uuid4().hex) return kwargs def new_region_ref(self, **kwargs): kwargs = self.new_ref(**kwargs) return kwargs def new_service_ref(self, **kwargs): kwargs = self.new_ref(**kwargs) kwargs.setdefault('name', uuid.uuid4().hex) kwargs.setdefault('type', uuid.uuid4().hex) return kwargs def _crud_policy_association_for_endpoint_via_id( self, http_action, manager_action): policy_id = uuid.uuid4().hex endpoint_id = uuid.uuid4().hex self.stub_url(http_action, ['policies', policy_id, self.manager.OS_EP_POLICY_EXT, 'endpoints', endpoint_id], status_code=204) manager_action(policy=policy_id, endpoint=endpoint_id) def _crud_policy_association_for_endpoint_via_obj( self, http_action, manager_action): policy_ref = self.new_policy_ref() endpoint_ref = self.new_endpoint_ref() policy = self.client.policies.resource_class( self.client.policies, policy_ref, loaded=True) endpoint = self.client.endpoints.resource_class( self.client.endpoints, endpoint_ref, loaded=True) self.stub_url(http_action, ['policies', policy_ref['id'], self.manager.OS_EP_POLICY_EXT, 'endpoints', endpoint_ref['id']], status_code=204) manager_action(policy=policy, endpoint=endpoint) def test_create_policy_association_for_endpoint_via_id(self): self._crud_policy_association_for_endpoint_via_id( 'PUT', self.manager.create_policy_association_for_endpoint) def test_create_policy_association_for_endpoint_via_obj(self): self._crud_policy_association_for_endpoint_via_obj( 'PUT', self.manager.create_policy_association_for_endpoint) def test_check_policy_association_for_endpoint_via_id(self): self._crud_policy_association_for_endpoint_via_id( 'HEAD', self.manager.check_policy_association_for_endpoint) def test_check_policy_association_for_endpoint_via_obj(self): self._crud_policy_association_for_endpoint_via_obj( 'HEAD', self.manager.check_policy_association_for_endpoint) def test_delete_policy_association_for_endpoint_via_id(self): self._crud_policy_association_for_endpoint_via_id( 'DELETE', self.manager.delete_policy_association_for_endpoint) def test_delete_policy_association_for_endpoint_via_obj(self): self._crud_policy_association_for_endpoint_via_obj( 'DELETE', self.manager.delete_policy_association_for_endpoint) def _crud_policy_association_for_service_via_id( self, http_action, manager_action): policy_id = uuid.uuid4().hex service_id = uuid.uuid4().hex self.stub_url(http_action, ['policies', policy_id, self.manager.OS_EP_POLICY_EXT, 'services', service_id], status_code=204) manager_action(policy=policy_id, service=service_id) def _crud_policy_association_for_service_via_obj( self, http_action, manager_action): policy_ref = self.new_policy_ref() service_ref = self.new_service_ref() policy = self.client.policies.resource_class( self.client.policies, policy_ref, loaded=True) service = self.client.services.resource_class( self.client.services, service_ref, loaded=True) self.stub_url(http_action, ['policies', policy_ref['id'], self.manager.OS_EP_POLICY_EXT, 'services', service_ref['id']], status_code=204) manager_action(policy=policy, service=service) def test_create_policy_association_for_service_via_id(self): self._crud_policy_association_for_service_via_id( 'PUT', self.manager.create_policy_association_for_service) def test_create_policy_association_for_service_via_obj(self): self._crud_policy_association_for_service_via_obj( 'PUT', self.manager.create_policy_association_for_service) def test_check_policy_association_for_service_via_id(self): self._crud_policy_association_for_service_via_id( 'HEAD', self.manager.check_policy_association_for_service) def test_check_policy_association_for_service_via_obj(self): self._crud_policy_association_for_service_via_obj( 'HEAD', self.manager.check_policy_association_for_service) def test_delete_policy_association_for_service_via_id(self): self._crud_policy_association_for_service_via_id( 'DELETE', self.manager.delete_policy_association_for_service) def test_delete_policy_association_for_service_via_obj(self): self._crud_policy_association_for_service_via_obj( 'DELETE', self.manager.delete_policy_association_for_service) def _crud_policy_association_for_region_and_service_via_id( self, http_action, manager_action): policy_id = uuid.uuid4().hex region_id = uuid.uuid4().hex service_id = uuid.uuid4().hex self.stub_url(http_action, ['policies', policy_id, self.manager.OS_EP_POLICY_EXT, 'services', service_id, 'regions', region_id], status_code=204) manager_action(policy=policy_id, region=region_id, service=service_id) def _crud_policy_association_for_region_and_service_via_obj( self, http_action, manager_action): policy_ref = self.new_policy_ref() region_ref = self.new_region_ref() service_ref = self.new_service_ref() policy = self.client.policies.resource_class( self.client.policies, policy_ref, loaded=True) region = self.client.regions.resource_class( self.client.regions, region_ref, loaded=True) service = self.client.services.resource_class( self.client.services, service_ref, loaded=True) self.stub_url(http_action, ['policies', policy_ref['id'], self.manager.OS_EP_POLICY_EXT, 'services', service_ref['id'], 'regions', region_ref['id']], status_code=204) manager_action(policy=policy, region=region, service=service) def test_create_policy_association_for_region_and_service_via_id(self): self._crud_policy_association_for_region_and_service_via_id( 'PUT', self.manager.create_policy_association_for_region_and_service) def test_create_policy_association_for_region_and_service_via_obj(self): self._crud_policy_association_for_region_and_service_via_obj( 'PUT', self.manager.create_policy_association_for_region_and_service) def test_check_policy_association_for_region_and_service_via_id(self): self._crud_policy_association_for_region_and_service_via_id( 'HEAD', self.manager.check_policy_association_for_region_and_service) def test_check_policy_association_for_region_and_service_via_obj(self): self._crud_policy_association_for_region_and_service_via_obj( 'HEAD', self.manager.check_policy_association_for_region_and_service) def test_delete_policy_association_for_region_and_service_via_id(self): self._crud_policy_association_for_region_and_service_via_id( 'DELETE', self.manager.delete_policy_association_for_region_and_service) def test_delete_policy_association_for_region_and_service_via_obj(self): self._crud_policy_association_for_region_and_service_via_obj( 'DELETE', self.manager.delete_policy_association_for_region_and_service) def test_get_policy_for_endpoint(self): endpoint_id = uuid.uuid4().hex expected_policy = self.new_policy_ref() self.stub_url('GET', ['endpoints', endpoint_id, self.manager.OS_EP_POLICY_EXT, 'policy'], json={'policy': expected_policy}, status_code=200) policy_resp = self.manager.get_policy_for_endpoint( endpoint=endpoint_id) self.assertEqual(expected_policy['id'], policy_resp.id) self.assertEqual(expected_policy['blob'], policy_resp.blob) self.assertEqual(expected_policy['type'], policy_resp.type) def test_list_endpoints_for_policy(self): policy_id = uuid.uuid4().hex endpoints = {'endpoints': [self.new_endpoint_ref(), self.new_endpoint_ref()]} self.stub_url('GET', ['policies', policy_id, self.manager.OS_EP_POLICY_EXT, 'endpoints'], json=endpoints, status_code=200) endpoints_resp = self.manager.list_endpoints_for_policy( policy=policy_id) expected_endpoint_ids = [ endpoint['id'] for endpoint in endpoints['endpoints']] actual_endpoint_ids = [endpoint.id for endpoint in endpoints_resp] self.assertEqual(expected_endpoint_ids, actual_endpoint_ids)