exc_filters: catch and translate non existent constraint on drop

This patches add a new exception type that is raised when an operation on a non
existing constraint is executed.

Change-Id: Ib55a83b568bdb1c9139af27bd2fbf204508e15e4

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):