diff options
-rw-r--r-- | oslo_db/exception.py | 15 | ||||
-rw-r--r-- | oslo_db/sqlalchemy/exc_filters.py | 28 | ||||
-rw-r--r-- | oslo_db/tests/sqlalchemy/test_exc_filters.py | 68 |
3 files changed, 111 insertions, 0 deletions
diff --git a/oslo_db/exception.py b/oslo_db/exception.py index c5e0fa9..e67ea01 100644 --- a/oslo_db/exception.py +++ b/oslo_db/exception.py @@ -127,6 +127,21 @@ class DBReferenceError(DBError): super(DBReferenceError, self).__init__(inner_exception) +class DBNonExistentConstraint(DBError): + """Constraint does not exist. + + :param table: table name + :type table: str + :param constraint: constraint name + :type table: str + """ + + def __init__(self, table, constraint, inner_exception=None): + self.table = table + self.constraint = constraint + super(DBNonExistentConstraint, self).__init__(inner_exception) + + class DBDeadlock(DBError): """Database dead lock error. diff --git a/oslo_db/sqlalchemy/exc_filters.py b/oslo_db/sqlalchemy/exc_filters.py index 6ab082e..7b2d976 100644 --- a/oslo_db/sqlalchemy/exc_filters.py +++ b/oslo_db/sqlalchemy/exc_filters.py @@ -244,6 +244,34 @@ def _check_constraint_error( raise exception.DBConstraintError(table, check_name, integrity_error) +@filters("postgresql", sqla_exc.ProgrammingError, + r".* constraint \"(?P<constraint>.+)\" " + "of relation " + "\"(?P<relation>.+)\" does not exist") +@filters("mysql", sqla_exc.InternalError, + r".*1091,.*Can't DROP '(?P<constraint>.+)'; " + "check that column/key exists") +@filters("mysql", sqla_exc.InternalError, + r".*1025,.*Error on rename of '.+/(?P<relation>.+)' to ") +def _check_constraint_non_existing( + programming_error, match, engine_name, is_disconnect): + """Filter for constraint non existing errors.""" + + try: + relation = match.group("relation") + except IndexError: + relation = None + + try: + constraint = match.group("constraint") + except IndexError: + constraint = None + + raise exception.DBNonExistentConstraint(relation, + constraint, + programming_error) + + @filters("ibm_db_sa", sqla_exc.IntegrityError, r"^.*SQL0803N.*$") def _db2_dupe_key_error(integrity_error, match, engine_name, is_disconnect): """Filter for DB2 duplicate key errors. diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py index 4593c04..9dcf46b 100644 --- a/oslo_db/tests/sqlalchemy/test_exc_filters.py +++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py @@ -1,3 +1,5 @@ +# -*- encoding: utf-8 -*- +# # 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 @@ -20,6 +22,7 @@ from oslotest import base as oslo_test_base import six import sqlalchemy as sqla from sqlalchemy import event +import sqlalchemy.exc from sqlalchemy.orm import mapper from oslo_db import exception @@ -243,6 +246,71 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter): self.assertEqual("mysqldb has an attribute error", matched.args[0]) +class TestNonExistentConstraint( + _SQLAExceptionMatcher, + test_base.DbTestCase): + + def setUp(self): + super(TestNonExistentConstraint, self).setUp() + + meta = sqla.MetaData(bind=self.engine) + + self.table_1 = sqla.Table( + "resource_foo", meta, + sqla.Column("id", sqla.Integer, primary_key=True), + mysql_engine='InnoDB', + mysql_charset='utf8', + ) + self.table_1.create() + + +class TestNonExistentConstraintPostgreSQL( + TestNonExistentConstraint, + test_base.PostgreSQLOpportunisticTestCase): + + def test_raise(self): + matched = self.assertRaises( + exception.DBNonExistentConstraint, + self.engine.execute, + sqla.schema.DropConstraint( + sqla.ForeignKeyConstraint(["id"], ["baz.id"], + name="bar_fkey", + table=self.table_1)), + ) + self.assertInnerException( + matched, + "ProgrammingError", + "constraint \"bar_fkey\" of relation " + "\"resource_foo\" does not exist\n", + "ALTER TABLE resource_foo DROP CONSTRAINT bar_fkey", + ) + self.assertEqual("resource_foo", matched.table) + self.assertEqual("bar_fkey", matched.constraint) + + +class TestNonExistentConstraintMySQL( + TestNonExistentConstraint, + test_base.MySQLOpportunisticTestCase): + + def test_raise(self): + matched = self.assertRaises( + exception.DBNonExistentConstraint, + self.engine.execute, + sqla.schema.DropConstraint( + sqla.ForeignKeyConstraint(["id"], ["baz.id"], + name="bar_fkey", + table=self.table_1)), + ) + # NOTE(jd) Cannot check precisely with assertInnerException since MySQL + # error are not the same depending on its version… + self.assertIsInstance(matched.inner_exception, + sqlalchemy.exc.InternalError) + if matched.table is not None: + self.assertEqual("resource_foo", matched.table) + if matched.constraint is not None: + self.assertEqual("bar_fkey", matched.constraint) + + class TestReferenceErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase): def setUp(self): |