diff options
| -rw-r--r-- | oslo_db/exception.py | 12 | ||||
| -rw-r--r-- | oslo_db/sqlalchemy/exc_filters.py | 12 | ||||
| -rw-r--r-- | oslo_db/tests/sqlalchemy/test_exc_filters.py | 67 |
3 files changed, 91 insertions, 0 deletions
diff --git a/oslo_db/exception.py b/oslo_db/exception.py index e67ea01..9c0c6fb 100644 --- a/oslo_db/exception.py +++ b/oslo_db/exception.py @@ -142,6 +142,18 @@ class DBNonExistentConstraint(DBError): super(DBNonExistentConstraint, self).__init__(inner_exception) +class DBNonExistentTable(DBError): + """Table does not exist. + + :param table: table name + :type table: str + """ + + def __init__(self, table, inner_exception=None): + self.table = table + super(DBNonExistentTable, 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 7b2d976..7170219 100644 --- a/oslo_db/sqlalchemy/exc_filters.py +++ b/oslo_db/sqlalchemy/exc_filters.py @@ -272,6 +272,18 @@ def _check_constraint_non_existing( programming_error) +@filters("sqlite", sqla_exc.OperationalError, + r".* no such table: (?P<table>.+)") +@filters("mysql", sqla_exc.InternalError, + r".*1051,.*\"Unknown table '(.+\.)?(?P<table>.+)'\"") +@filters("postgresql", sqla_exc.ProgrammingError, + r".* table \"(?P<table>.+)\" does not exist") +def _check_table_non_existing( + programming_error, match, engine_name, is_disconnect): + """Filter for table non existing errors.""" + raise exception.DBNonExistentTable(match.group("table"), 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 9dcf46b..6f65048 100644 --- a/oslo_db/tests/sqlalchemy/test_exc_filters.py +++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py @@ -311,6 +311,73 @@ class TestNonExistentConstraintMySQL( self.assertEqual("bar_fkey", matched.constraint) +class TestNonExistentTable( + _SQLAExceptionMatcher, + test_base.DbTestCase): + + def setUp(self): + super(TestNonExistentTable, self).setUp() + + self.meta = sqla.MetaData(bind=self.engine) + + self.table_1 = sqla.Table( + "foo", self.meta, + sqla.Column("id", sqla.Integer, primary_key=True), + mysql_engine='InnoDB', + mysql_charset='utf8', + ) + + def test_raise(self): + matched = self.assertRaises( + exception.DBNonExistentTable, + self.engine.execute, + sqla.schema.DropTable(self.table_1), + ) + self.assertInnerException( + matched, + "OperationalError", + "no such table: foo", + "\nDROP TABLE foo", + ) + self.assertEqual("foo", matched.table) + + +class TestNonExistentTablePostgreSQL( + TestNonExistentTable, + test_base.PostgreSQLOpportunisticTestCase): + + def test_raise(self): + matched = self.assertRaises( + exception.DBNonExistentTable, + self.engine.execute, + sqla.schema.DropTable(self.table_1), + ) + self.assertInnerException( + matched, + "ProgrammingError", + "table \"foo\" does not exist\n", + "\nDROP TABLE foo", + ) + self.assertEqual("foo", matched.table) + + +class TestNonExistentTableMySQL( + TestNonExistentTable, + test_base.MySQLOpportunisticTestCase): + + def test_raise(self): + matched = self.assertRaises( + exception.DBNonExistentTable, + self.engine.execute, + sqla.schema.DropTable(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) + self.assertEqual("foo", matched.table) + + class TestReferenceErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase): def setUp(self): |