openstack/deb-barbican
Code Issues Proposed changes

Handle SQL Integrity Error More Generically

Browse Source 
SQL command with IntegrityError when any number of
constraints fail to pass.  These include checking for
duplicates, checking for not null, or foreign key checks.

Barbican was reporting these all as duplicates which can confuse
debugging.  This patch reports the error more accurately as an
SQL constraint check failure.

Change-Id: I7043fe01f949326b1e38c8f320ece8418f36acc4
Co-Authored-By: Dave McCowan <dmccowan@cisco.com>
Closes-bug: #1489457
This commit is contained in:
Dave McCowan 2016-01-14 15:31:40 -06:00 committed by Samantha Blanco
parent d0ebc6c321
commit 9744f80186
6 changed files with 23 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
barbican/common/exception.py
View File

@ -167,8 +167,8 @@ class BadStoreUri(BarbicanException):
message = u._("The Store URI was malformed.")
class Duplicate(BarbicanException):
message = u._("An object with the same identifier already exists.")
class ConstraintCheck(BarbicanException):
message = u._("A defined SQL constraint check failed: %(error)s")
class StorageFull(BarbicanException):

12
barbican/model/repositories.py
View File

@ -21,6 +21,7 @@ quite intense for sqlalchemy, and maybe could be simplified.
"""
import logging
import re
import sys
import time
import uuid
@ -407,9 +408,10 @@ class BaseRepo(object):
try:
LOG.debug("Saving entity...")
entity.save(session=session)
except sqlalchemy.exc.IntegrityError:
except sqlalchemy.exc.IntegrityError as e:
LOG.exception(u._LE('Problem saving entity for create'))
_raise_entity_already_exists(self._do_entity_name())
error_msg = re.sub('[()]', '', str(e.orig.args))
raise exception.ConstraintCheck(error=error_msg)
LOG.debug('Elapsed repo '
'create secret:%s', (time.time() - start)) # DEBUG
@ -2512,9 +2514,3 @@ def _raise_no_entities_found(entity_name):
raise exception.NotFound(
u._("No entities of type {entity_name} found").format(
entity_name=entity_name))
def _raise_entity_already_exists(entity_name):
raise exception.Duplicate(
u._("Entity '{entity_name}' "
"already exists").format(entity_name=entity_name))

11
barbican/tests/model/repositories/test_repositories.py
View File

@ -144,17 +144,6 @@ class WhenInvokingExceptionMethods(utils.BaseTestCase):
"No entities of type test_entity found",
exception_result.message)
def test_should_raise_for_entity_already_exists(self):
exception_result = self.assertRaises(
exception.Duplicate,
repositories._raise_entity_already_exists,
self.entity_name)
self.assertEqual(
"Entity 'test_entity' already exists",
exception_result.message)
class WhenTestingBaseRepository(database_utils.RepositoryTestCase):

4
barbican/tests/model/repositories/test_repositories_certificate_authorities.py
View File

@ -441,7 +441,7 @@ class WhenTestingPreferredCARepo(database_utils.RepositoryTestCase):
session=session,
suppress_exception=False)
def test_should_raise_duplicate_entries(self):
def test_should_raise_constraint_check(self):
session = self.ca_repo.get_session()
ca = self._add_ca(self.parsed_ca, session)
@ -449,7 +449,7 @@ class WhenTestingPreferredCARepo(database_utils.RepositoryTestCase):
project = self._add_project("project_1", session)
self._add_preferred_ca(project.id, ca.id, session)
self.assertRaises(
exception.Duplicate,
exception.ConstraintCheck,
self._add_preferred_ca,
project.id,
ca2.id,

7
barbican/tests/model/repositories/test_repositories_consumers.py
View File

@ -55,7 +55,7 @@ class WhenTestingContainerConsumerRepository(utils.RepositoryTestCase):
container.id, project.external_id, session=session)
self.assertEqual(1, len(container2.consumers))
def test_should_raise_duplicate_create_same_composite_key_no_id(self):
def test_should_raise_constraint_create_same_composite_key_no_id(self):
session = self.repo.get_session()
project = models.Project()
@ -82,12 +82,13 @@ class WhenTestingContainerConsumerRepository(utils.RepositoryTestCase):
container.id, project.id, {'name': 'name', 'URL': 'www.foo.com'})
exception_result = self.assertRaises(
exception.Duplicate,
exception.ConstraintCheck,
self.repo.create_from,
consumer2,
session=session)
self.assertEqual(
"Entity 'ContainerConsumer' already exists",
u"A defined SQL constraint check failed: 'UNIQUE constraint "
"failed: container_consumer_metadata.data_hash',",
exception_result.message)
def test_should_raise_no_result_found_get_container_id(self):

21
barbican/tests/model/repositories/test_repositories_secret_stores.py
View File

@ -137,24 +137,24 @@ class WhenTestingSecretStoresRepo(database_utils.RepositoryTestCase):
self.assertEqual("middle_name", all_stores[3].name)
self.assertEqual(False, all_stores[3].global_default)
def test_should_raise_duplicate_for_same_plugin_names(self):
def test_should_raise_constraint_for_same_plugin_names(self):
"""Check for store and crypto plugin name combination uniqueness"""
name = 'second_name'
store_plugin = 'second_store'
crypto_plugin = 'second_crypto'
self._create_secret_store(name, store_plugin, crypto_plugin, False)
self.assertRaises(exception.Duplicate, self._create_secret_store,
self.assertRaises(exception.ConstraintCheck, self._create_secret_store,
"thrid_name", store_plugin, crypto_plugin, False)
def test_should_raise_duplicate_for_same_names(self):
def test_should_raise_constraint_for_same_names(self):
"""Check for secret store 'name' uniqueness"""
name = 'Db backend'
store_plugin = 'second_store'
crypto_plugin = 'second_crypto'
self._create_secret_store(name, store_plugin, crypto_plugin, False)
self.assertRaises(exception.Duplicate, self._create_secret_store,
self.assertRaises(exception.ConstraintCheck, self._create_secret_store,
name, "another_store", "another_crypto", False)
def test_do_entity_name(self):
@ -169,8 +169,8 @@ class WhenTestingSecretStoresRepo(database_utils.RepositoryTestCase):
try:
self._create_secret_store(name, store_plugin, crypto_plugin, False)
self.assertFail()
except exception.Duplicate as ex:
self.assertIn("SecretStores", ex.message)
except exception.ConstraintCheck as ex:
self.assertIn("UNIQUE constraint", ex.message)
class WhenTestingProjectSecretStoreRepo(database_utils.RepositoryTestCase):
@ -264,7 +264,7 @@ class WhenTestingProjectSecretStoreRepo(database_utils.RepositoryTestCase):
self.assertIsNone(proj_s_store)
def test_should_raise_duplicate_for_same_project_id(self):
def test_should_raise_constraint_for_same_project_id(self):
"""Check preferred secret store is set only once for project"""
project1 = self._create_project()
@ -284,7 +284,8 @@ class WhenTestingProjectSecretStoreRepo(database_utils.RepositoryTestCase):
s_store2 = self._create_secret_store(name, store_plugin,
crypto_plugin, False)
self.assertRaises(exception.Duplicate, self._create_project_store,
self.assertRaises(exception.ConstraintCheck,
self._create_project_store,
project1.id, s_store2.id)
def test_do_entity_name(self):
@ -311,8 +312,8 @@ class WhenTestingProjectSecretStoreRepo(database_utils.RepositoryTestCase):
crypto_plugin, False)
self._create_project_store(project1.id, s_store2.id)
self.assertFail()
except exception.Duplicate as ex:
self.assertIn("ProjectSecretStore", ex.message)
except exception.ConstraintCheck as ex:
self.assertIn("UNIQUE constraint", ex.message)
def test_get_secret_store_for_project(self):
project1 = self._create_project()