Merge "Use testtools as base test class."

This commit is contained in:
Jenkins 2013-08-31 03:21:12 +00:00 committed by Gerrit Code Review
commit 372a06236a
17 changed files with 147 additions and 93 deletions

View File

@ -24,10 +24,10 @@ This module can be removed when keystone.middleware.auth_token is removed.
"""
import unittest
import testtools
class TestAuthToken(unittest.TestCase):
class TestAuthToken(testtools.TestCase):
def test_import(self):
# a consuming service like nova would import oslo.config first
from oslo.config import cfg

View File

@ -14,9 +14,9 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime
import errno
import os
import re
import shutil
import socket
import StringIO
@ -28,7 +28,7 @@ import mox
import nose.exc
from paste import deploy
import stubout
import unittest2 as unittest
import testtools
from keystone.common import environment
environment.use_eventlet()
@ -203,7 +203,7 @@ class NoModule(object):
sys.meta_path.insert(0, finder)
class TestCase(NoModule, unittest.TestCase):
class TestCase(NoModule, testtools.TestCase):
def __init__(self, *args, **kw):
super(TestCase, self).__init__(*args, **kw)
self._paths = []
@ -378,14 +378,51 @@ class TestCase(NoModule, unittest.TestCase):
:param delta: Maximum allowable time delta, defined in seconds.
"""
self.assertAlmostEqual(a, b, delta=datetime.timedelta(seconds=delta))
msg = '%s != %s within %s delta' % (a, b, delta)
self.assertTrue(abs(a - b).seconds <= delta, msg)
def assertNotEmpty(self, l):
self.assertTrue(len(l))
def assertDictEqual(self, d1, d2, msg=None):
self.assert_(isinstance(d1, dict), 'First argument is not a dict')
self.assert_(isinstance(d2, dict), 'Second argument is not a dict')
self.assertEqual(d1, d2, msg)
def assertRaisesRegexp(self, expected_exception, expected_regexp,
callable_obj, *args, **kwargs):
"""Asserts that the message in a raised exception matches a regexp.
"""
try:
callable_obj(*args, **kwargs)
except expected_exception as exc_value:
if isinstance(expected_regexp, basestring):
expected_regexp = re.compile(expected_regexp)
if not expected_regexp.search(str(exc_value)):
raise self.failureException(
'"%s" does not match "%s"' %
(expected_regexp.pattern, str(exc_value)))
else:
if hasattr(expected_exception, '__name__'):
excName = expected_exception.__name__
else:
excName = str(expected_exception)
raise self.failureException, "%s not raised" % excName
def assertDictContainsSubset(self, expected, actual, msg=None):
"""Checks whether actual is a superset of expected."""
safe_repr = unittest.util.safe_repr
def safe_repr(obj, short=False):
_MAX_LENGTH = 80
try:
result = repr(obj)
except Exception:
result = object.__repr__(obj)
if not short or len(result) < _MAX_LENGTH:
return result
return result[:_MAX_LENGTH] + ' [truncated]...'
missing = []
mismatched = []
for key, value in expected.iteritems():

View File

@ -2634,16 +2634,22 @@ class TrustTests(object):
class CommonHelperTests(test.TestCase):
def test_format_helper_raises_malformed_on_missing_key(self):
with self.assertRaises(exception.MalformedEndpoint):
core.format_url("http://%(foo)s/%(bar)s", {"foo": "1"})
self.assertRaises(exception.MalformedEndpoint,
core.format_url,
"http://%(foo)s/%(bar)s",
{"foo": "1"})
def test_format_helper_raises_malformed_on_wrong_type(self):
with self.assertRaises(exception.MalformedEndpoint):
core.format_url("http://%foo%s", {"foo": "1"})
self.assertRaises(exception.MalformedEndpoint,
core.format_url,
"http://%foo%s",
{"foo": "1"})
def test_format_helper_raises_malformed_on_incomplete_format(self):
with self.assertRaises(exception.MalformedEndpoint):
core.format_url("http://%(foo)", {"foo": "1"})
self.assertRaises(exception.MalformedEndpoint,
core.format_url,
"http://%(foo)",
{"foo": "1"})
class CatalogTests(object):

View File

@ -584,25 +584,36 @@ class LDAPIdentity(test.TestCase, BaseLDAPIdentity):
def test_domain_crud(self):
domain = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex,
'enabled': True, 'description': uuid.uuid4().hex}
with self.assertRaises(exception.Forbidden):
self.identity_api.create_domain(domain['id'], domain)
with self.assertRaises(exception.Conflict):
self.identity_api.create_domain(
CONF.identity.default_domain_id, domain)
with self.assertRaises(exception.DomainNotFound):
self.identity_api.get_domain(domain['id'])
with self.assertRaises(exception.DomainNotFound):
domain['description'] = uuid.uuid4().hex
self.identity_api.update_domain(domain['id'], domain)
with self.assertRaises(exception.Forbidden):
self.identity_api.update_domain(
CONF.identity.default_domain_id, domain)
with self.assertRaises(exception.DomainNotFound):
self.identity_api.get_domain(domain['id'])
with self.assertRaises(exception.DomainNotFound):
self.identity_api.delete_domain(domain['id'])
with self.assertRaises(exception.Forbidden):
self.identity_api.delete_domain(CONF.identity.default_domain_id)
self.assertRaises(exception.Forbidden,
self.identity_api.create_domain,
domain['id'],
domain)
self.assertRaises(exception.Conflict,
self.identity_api.create_domain,
CONF.identity.default_domain_id,
domain)
self.assertRaises(exception.DomainNotFound,
self.identity_api.get_domain,
domain['id'])
domain['description'] = uuid.uuid4().hex
self.assertRaises(exception.DomainNotFound,
self.identity_api.update_domain,
domain['id'],
domain)
self.assertRaises(exception.Forbidden,
self.identity_api.update_domain,
CONF.identity.default_domain_id,
domain)
self.assertRaises(exception.DomainNotFound,
self.identity_api.get_domain,
domain['id'])
self.assertRaises(exception.DomainNotFound,
self.identity_api.delete_domain,
domain['id'])
self.assertRaises(exception.Forbidden,
self.identity_api.delete_domain,
CONF.identity.default_domain_id)
self.assertRaises(exception.DomainNotFound,
self.identity_api.get_domain,
domain['id'])
@ -766,6 +777,7 @@ class LdapIdentitySqlAssignment(sql.Base, test.TestCase, BaseLDAPIdentity):
test.testsdir('backend_ldap_sql.conf')])
def setUp(self):
super(LdapIdentitySqlAssignment, self).setUp()
self._set_config()
self.clear_database()
self.load_backends()
@ -779,6 +791,7 @@ class LdapIdentitySqlAssignment(sql.Base, test.TestCase, BaseLDAPIdentity):
sql.ModelBase.metadata.drop_all(bind=self.engine)
self.engine.dispose()
sql.set_global_engine(None)
super(LdapIdentitySqlAssignment, self).tearDown()
def test_domain_crud(self):
pass

View File

@ -138,8 +138,8 @@ class MemcacheToken(test.TestCase, test_backend.TokenTests):
self.token_api.list_tokens(user_id)
def test_flush_expired_token(self):
with self.assertRaises(exception.NotImplemented):
self.token_api.flush_expired_tokens()
self.assertRaises(exception.NotImplemented,
self.token_api.flush_expired_tokens)
def test_cleanup_user_index_on_create(self):
valid_token_id = uuid.uuid4().hex

View File

@ -357,8 +357,10 @@ class SqlCatalog(SqlTests, test_backend.CatalogTests):
}
self.catalog_api.create_endpoint(endpoint['id'], endpoint.copy())
with self.assertRaises(exception.MalformedEndpoint):
self.catalog_api.get_catalog('fake-user', 'fake-tenant')
self.assertRaises(exception.MalformedEndpoint,
self.catalog_api.get_catalog,
'fake-user',
'fake-tenant')
def test_get_catalog_with_empty_public_url(self):
service = {
@ -403,8 +405,10 @@ class SqlCatalog(SqlTests, test_backend.CatalogTests):
'url': uuid.uuid4().hex,
}
with self.assertRaises(exception.StringLengthExceeded):
self.catalog_api.create_endpoint(endpoint['id'], endpoint.copy())
self.assertRaises(exception.StringLengthExceeded,
self.catalog_api.create_endpoint,
endpoint['id'],
endpoint.copy())
class SqlPolicy(SqlTests, test_backend.PolicyTests):

View File

@ -63,5 +63,7 @@ class TestTemplatedCatalog(test.TestCase, test_backend.CatalogTests):
(self.catalog_api.driver.templates
['RegionOne']['compute']['adminURL']) = \
'http://localhost:$(compute_port)s/v1.1/$(tenant)s'
with self.assertRaises(exception.MalformedEndpoint):
self.catalog_api.get_catalog('fake-user', 'fake-tenant')
self.assertRaises(exception.MalformedEndpoint,
self.catalog_api.get_catalog,
'fake-user',
'fake-tenant')

View File

@ -15,7 +15,7 @@
# under the License.
import inspect
import unittest2 as unittest
import testtools
from keystone import assignment
from keystone import catalog
@ -27,7 +27,7 @@ from keystone import policy
from keystone import token
class TestDrivers(unittest.TestCase):
class TestDrivers(testtools.TestCase):
"""Asserts that drivers are written as expected.
Public methods on drivers should raise keystone.exception.NotImplemented,
@ -44,8 +44,7 @@ class TestDrivers(unittest.TestCase):
args = inspect.getargspec(f).args
args.remove('self')
kwargs = dict(zip(args, [None] * len(args)))
with self.assertRaises(exception.NotImplemented):
f(**kwargs)
self.assertRaises(exception.NotImplemented, f, **kwargs)
def assertInterfaceNotImplemented(self, interface):
"""Public methods on an interface class should not be implemented."""

View File

@ -28,12 +28,6 @@ CONF = config.CONF
class ExceptionTestCase(test.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
def assertValidJsonRendering(self, e):
resp = wsgi.render_exception(e)
self.assertEqual(resp.status_int, e.code)

View File

@ -14,13 +14,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest2 as unittest
import testtools
import uuid
from keystone.common import dependency
class TestDependencyInjection(unittest.TestCase):
class TestDependencyInjection(testtools.TestCase):
def tearDown(self):
dependency.reset()
super(TestDependencyInjection, self).tearDown()
@ -167,10 +167,12 @@ class TestDependencyInjection(unittest.TestCase):
class Consumer(object):
pass
with self.assertRaises(dependency.UnresolvableDependencyException):
def for_test():
Consumer()
dependency.resolve_future_dependencies()
self.assertRaises(dependency.UnresolvableDependencyException, for_test)
def test_circular_dependency(self):
p1_name = uuid.uuid4().hex
p2_name = uuid.uuid4().hex

View File

@ -47,6 +47,7 @@ class TestNoAdminTokenAuth(test.TestCase):
def tearDown(self):
self.admin_app = None
os.remove(test.tmpdir('no_admin_token_auth-paste.ini'))
super(TestNoAdminTokenAuth, self).tearDown()
def test_request_no_admin_token_auth(self):
# This test verifies that if the admin_token_auth middleware isn't

View File

@ -175,7 +175,7 @@ class DefaultPolicyTestCase(test.TestCase):
common_policy.set_rules(these_rules)
def tearDown(self):
super(DefaultPolicyTestCase, self).setUp()
super(DefaultPolicyTestCase, self).tearDown()
rules.reset()
def test_policy_called(self):

View File

@ -14,7 +14,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import unittest2 as unittest
import testtools
import webob
from keystone.middleware import s3_token
@ -53,7 +53,7 @@ class FakeHTTPConnection(object):
pass
class S3TokenMiddlewareTestBase(unittest.TestCase):
class S3TokenMiddlewareTestBase(testtools.TestCase):
def setUp(self):
super(S3TokenMiddlewareTestBase, self).setUp()
@ -194,7 +194,7 @@ class S3TokenMiddlewareTestBad(S3TokenMiddlewareTestBase):
self.assertEqual(resp.status_int, s3_invalid_req.status_int)
class S3TokenMiddlewareTestUtil(unittest.TestCase):
class S3TokenMiddlewareTestUtil(testtools.TestCase):
def test_split_path_failed(self):
self.assertRaises(ValueError, s3_token.split_path, '')
self.assertRaises(ValueError, s3_token.split_path, '/')

View File

@ -61,6 +61,7 @@ class RestfulTestCase(test_content_types.RestfulTestCase):
load_sample_data should be set to false.
"""
super(RestfulTestCase, self).setUp()
self.config(self.config_files())
self.setup_database()
@ -146,15 +147,12 @@ class RestfulTestCase(test_content_types.RestfulTestCase):
self.public_server = None
self.admin_server = None
self.teardown_database()
# NOTE(morganfainberg): The only way to reconfigure the
# CacheRegion object on each setUp() call is to remove the
# .backend property.
del cache.REGION.backend
# need to reset the plug-ins
auth.controllers.AUTH_METHODS = {}
#drop the policy rules
CONF.reset()
rules.reset()
super(RestfulTestCase, self).tearDown()
def new_ref(self):
"""Populates a ref with attributes common to all API entities."""

View File

@ -16,7 +16,6 @@
import uuid
from keystone import config
from keystone import exception
import test_v3
@ -1107,17 +1106,12 @@ class IdentityTestCase(test_v3.RestfulTestCase):
link_url=gp1_url)
class IdentityIneritanceTestCase(test_v3.RestfulTestCase):
class IdentityInheritanceTestCase(test_v3.RestfulTestCase):
"""Test inheritance crud and its effects."""
def setUp(self):
self.orig_extension_enablement = config.CONF.os_inherit.enabled
self.opt_in_group('os_inherit', enabled=True)
super(IdentityIneritanceTestCase, self).setUp()
def tearDown(self):
super(IdentityIneritanceTestCase, self).tearDown()
self.opt_in_group('os_inherit', enabled=self.orig_extension_enablement)
super(IdentityInheritanceTestCase, self).setUp()
def test_crud_user_inherited_domain_role_grants(self):
role_list = []
@ -1154,28 +1148,6 @@ class IdentityIneritanceTestCase(test_v3.RestfulTestCase):
self.assertValidRoleListResponse(r, expected_length=0)
self.assertIn(collection_url, r.result['links']['self'])
def test_crud_inherited_role_grants_failed_if_disabled(self):
# Disable the extension and check no API calls can be issued
self.opt_in_group('os_inherit', enabled=False)
super(IdentityIneritanceTestCase, self).setUp()
role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
self.assignment_api.create_role(role['id'], role)
base_collection_url = (
'/OS-INHERIT/domains/%(domain_id)s/users/%(user_id)s/roles' % {
'domain_id': self.domain_id,
'user_id': self.user['id']})
member_url = '%(collection_url)s/%(role_id)s/inherited_to_projects' % {
'collection_url': base_collection_url,
'role_id': role['id']}
collection_url = base_collection_url + '/inherited_to_projects'
self.put(member_url, expected_status=404)
self.head(member_url, expected_status=404)
self.get(collection_url, expected_status=404)
self.delete(member_url, expected_status=404)
def test_list_role_assignments_for_inherited_domain_grants(self):
"""Call ``GET /role_assignments with inherited domain grants``.
@ -1555,3 +1527,29 @@ class IdentityIneritanceTestCase(test_v3.RestfulTestCase):
role_id=role_list[4]['id'], inherited_to_projects=True)
self.assertRoleAssignmentInListResponse(r, ud_entity, link_url=ud_url)
self.assertRoleAssignmentInListResponse(r, gd_entity, link_url=gd_url)
class IdentityInheritanceDisabledTestCase(test_v3.RestfulTestCase):
"""Test inheritance crud and its effects."""
def setUp(self):
self.opt_in_group('os_inherit', enabled=False)
super(IdentityInheritanceDisabledTestCase, self).setUp()
def test_crud_inherited_role_grants_failed_if_disabled(self):
role = {'id': uuid.uuid4().hex, 'name': uuid.uuid4().hex}
self.assignment_api.create_role(role['id'], role)
base_collection_url = (
'/OS-INHERIT/domains/%(domain_id)s/users/%(user_id)s/roles' % {
'domain_id': self.domain_id,
'user_id': self.user['id']})
member_url = '%(collection_url)s/%(role_id)s/inherited_to_projects' % {
'collection_url': base_collection_url,
'role_id': role['id']}
collection_url = base_collection_url + '/inherited_to_projects'
self.put(member_url, expected_status=404)
self.head(member_url, expected_status=404)
self.get(collection_url, expected_status=404)
self.delete(member_url, expected_status=404)

View File

@ -61,6 +61,7 @@ class OAuth1Tests(test_v3.RestfulTestCase):
def tearDown(self):
os.remove(OAUTH_PASTE_FILE)
super(OAuth1Tests, self).tearDown()
def _generate_paste_config(self):
# Generate a file, based on keystone-paste.ini,

View File

@ -22,8 +22,7 @@ openstack.nose_plugin
nosehtmloutput
# required to build documentation
Sphinx>=1.1.2
# backport of unittest lib in python 2.7
unittest2
testtools>=0.9.32
# test wsgi apps without starting an http server
webtest