From c4784d77f0e45e8993e34847c5f50704b50219ce Mon Sep 17 00:00:00 2001 From: Samuel de Medeiros Queiroz Date: Wed, 12 Aug 2015 09:40:37 -0300 Subject: [PATCH] Create unit tests for endpoint policy drivers This change adds unit tests for the current version of the endpoint policy abstract driver. It also clarifies the return type of get_policy_association in the endpoint policy abstract driver definition. Change-Id: Idde4b139ef4d1b048901f5a0a1fd9df3b6836994 --- keystone/endpoint_policy/backends/base.py | 5 +- .../tests/unit/endpoint_policy/__init__.py | 0 .../unit/endpoint_policy/backends/__init__.py | 0 .../endpoint_policy/backends/test_base.py | 152 ++++++++++++++++++ .../unit/endpoint_policy/backends/test_sql.py | 43 +++++ 5 files changed, 198 insertions(+), 2 deletions(-) create mode 100644 keystone/tests/unit/endpoint_policy/__init__.py create mode 100644 keystone/tests/unit/endpoint_policy/backends/__init__.py create mode 100644 keystone/tests/unit/endpoint_policy/backends/test_base.py create mode 100644 keystone/tests/unit/endpoint_policy/backends/test_sql.py diff --git a/keystone/endpoint_policy/backends/base.py b/keystone/endpoint_policy/backends/base.py index de7d48a67..f2195b0cd 100644 --- a/keystone/endpoint_policy/backends/base.py +++ b/keystone/endpoint_policy/backends/base.py @@ -47,7 +47,7 @@ class EndpointPolicyDriverV8(object): @abc.abstractmethod def check_policy_association(self, policy_id, endpoint_id=None, service_id=None, region_id=None): - """Check existence a policy association. + """Check existence of a policy association. :param policy_id: identity of policy that is being associated :type policy_id: string @@ -98,7 +98,8 @@ class EndpointPolicyDriverV8(object): :type region_id: string :raises keystone.exception.PolicyAssociationNotFound: If there is no match for the specified association. - :returns: dict containing policy_id + :returns: dict containing policy_id (value is a tuple containing only + the policy_id) """ raise exception.NotImplemented() # pragma: no cover diff --git a/keystone/tests/unit/endpoint_policy/__init__.py b/keystone/tests/unit/endpoint_policy/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/keystone/tests/unit/endpoint_policy/backends/__init__.py b/keystone/tests/unit/endpoint_policy/backends/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/keystone/tests/unit/endpoint_policy/backends/test_base.py b/keystone/tests/unit/endpoint_policy/backends/test_base.py new file mode 100644 index 000000000..8d00fb145 --- /dev/null +++ b/keystone/tests/unit/endpoint_policy/backends/test_base.py @@ -0,0 +1,152 @@ +# 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 keystone import exception + + +class DriverTestCase(object): + """Test cases to validate the endpoint policy driver behavior.""" + + @property + def driver(self): + raise exception.NotImplemented() + + def create_association(self, **kwargs): + association = {'policy_id': uuid.uuid4().hex, + 'endpoint_id': None, + 'service_id': None, + 'region_id': None} + association.update(kwargs) + self.driver.create_policy_association(**association) + return association + + def test_create_policy_association(self): + association = self.create_association(endpoint_id=uuid.uuid4().hex) + self.driver.check_policy_association(**association) + + association = self.create_association(service_id=uuid.uuid4().hex, + region_id=uuid.uuid4().hex) + self.driver.check_policy_association(**association) + + association = self.create_association(service_id=uuid.uuid4().hex) + self.driver.check_policy_association(**association) + + def test_recreate_policy_association(self): + # Creating a policy association to a target that already has a policy + # associated to it, will cause the original policy to be overridden + original_association = self.create_association( + service_id=uuid.uuid4().hex) + override_association = original_association.copy() + override_association.update({'policy_id': uuid.uuid4().hex}) + + self.driver.create_policy_association(**override_association) + + self.driver.check_policy_association(**override_association) + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **original_association) + + def test_check_policy_association(self): + association = self.create_association(service_id=uuid.uuid4().hex, + region_id=uuid.uuid4().hex) + self.driver.check_policy_association(**association) + + # An association is uniquely identified by its target. Omitting any + # attribute (region_id in this case) will result in a different check + association.pop('region_id') + + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) + + def test_delete_policy_association(self): + association = self.create_association(endpoint_id=uuid.uuid4().hex) + self.driver.delete_policy_association(**association) + + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) + + def test_get_policy_association(self): + association = self.create_association(service_id=uuid.uuid4().hex) + + # Extract the policy_id from the association and query it by the target + policy_id = association.pop('policy_id') + + association_ref = self.driver.get_policy_association(**association) + self.assertEqual({'policy_id': (policy_id,)}, association_ref) + + def test_list_associations_for_policy(self): + policy_id = uuid.uuid4().hex + first = self.create_association(endpoint_id=uuid.uuid4().hex, + policy_id=policy_id) + second = self.create_association(service_id=uuid.uuid4().hex, + policy_id=policy_id) + + associations_ref = self.driver.list_associations_for_policy(policy_id) + + self.assertIn(first, associations_ref) + self.assertIn(second, associations_ref) + + def test_delete_association_by_endpoint(self): + endpoint_id = uuid.uuid4().hex + associations = [self.create_association(endpoint_id=endpoint_id), + self.create_association(endpoint_id=endpoint_id)] + + self.driver.delete_association_by_endpoint(endpoint_id) + + for association in associations: + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) + + def test_delete_association_by_service(self): + service_id = uuid.uuid4().hex + associations = [self.create_association(service_id=service_id), + self.create_association(service_id=service_id)] + + self.driver.delete_association_by_service(service_id) + + for association in associations: + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) + + def test_delete_association_by_region(self): + region_id = uuid.uuid4().hex + first = self.create_association(service_id=uuid.uuid4().hex, + region_id=region_id) + second = self.create_association(service_id=uuid.uuid4().hex, + region_id=region_id) + + self.driver.delete_association_by_region(region_id) + + for association in [first, second]: + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) + + def test_delete_association_by_policy(self): + policy_id = uuid.uuid4().hex + first = self.create_association(endpoint_id=uuid.uuid4().hex, + policy_id=policy_id) + second = self.create_association(service_id=uuid.uuid4().hex, + policy_id=policy_id) + + self.driver.delete_association_by_policy(policy_id) + + for association in [first, second]: + self.assertRaises(exception.PolicyAssociationNotFound, + self.driver.check_policy_association, + **association) diff --git a/keystone/tests/unit/endpoint_policy/backends/test_sql.py b/keystone/tests/unit/endpoint_policy/backends/test_sql.py new file mode 100644 index 000000000..d236ad913 --- /dev/null +++ b/keystone/tests/unit/endpoint_policy/backends/test_sql.py @@ -0,0 +1,43 @@ +# 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 keystone.common import sql +from keystone.endpoint_policy.backends import sql as sql_driver +from keystone.tests import unit +from keystone.tests.unit.backend import core_sql +from keystone.tests.unit.endpoint_policy.backends import test_base +from keystone.tests.unit.ksfixtures import database + + +class SQLModelTestCase(core_sql.BaseBackendSqlModels): + """Test cases to validate the table structure.""" + + def test_policy_association_model(self): + cols = (('id', sql.String, 64), + ('policy_id', sql.String, 64), + ('endpoint_id', sql.String, 64), + ('service_id', sql.String, 64), + ('region_id', sql.String, 64)) + + self.assertExpectedSchema('policy_association', cols) + + +class SQLDriverTestCase(test_base.DriverTestCase, unit.TestCase): + + def setUp(self): + super(SQLDriverTestCase, self).setUp() + self.useFixture(database.Database()) + self._driver = sql_driver.EndpointPolicy() + + @property + def driver(self): + return self._driver