exc_filters: support for ForeignKey error on delete

The current code only supports this type of constraint on INSERT, this
adds support for DELETE.

Change-Id: I981eb5f7b71bd37a35a93bbd3fada01d74aba12d
Closes-Bug: #1439358

2 files changed, 80 insertions, 5 deletions

diff --git a/oslo_db/sqlalchemy/exc_filters.py b/oslo_db/sqlalchemy/exc_filters.py
index 777fda6..b3c94e2 100644
--- a/oslo_db/sqlalchemy/exc_filters.py
+++ b/oslo_db/sqlalchemy/exc_filters.py
@@ -187,12 +187,12 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
          r"(?i).*foreign key constraint failed")
 @filters("postgresql", sqla_exc.IntegrityError,
          r".*on table \"(?P<table>[^\"]+)\" violates "
-         "foreign key constraint \"(?P<constraint>[^\"]+)\"\s*\n"
+         "foreign key constraint \"(?P<constraint>[^\"]+)\".*\n"
          "DETAIL:  Key \((?P<key>.+)\)=\(.+\) "
-         "is not present in table "
+         "is (not present in|still referenced from) table "
          "\"(?P<key_table>[^\"]+)\".")
 @filters("mysql", sqla_exc.IntegrityError,
-         r".* u?'Cannot add or update a child row: "
+         r".* u?'Cannot (add|delete) or update a (child|parent) row: "
          'a foreign key constraint fails \([`"].+[`"]\.[`"](?P<table>.+)[`"], '
          'CONSTRAINT [`"](?P<constraint>.+)[`"] FOREIGN KEY '
          '\([`"](?P<key>.+)[`"]\) REFERENCES [`"](?P<key_table>.+)[`"] ')
diff --git a/oslo_db/tests/sqlalchemy/test_exc_filters.py b/oslo_db/tests/sqlalchemy/test_exc_filters.py
index aafdcfb..6121998 100644
--- a/oslo_db/tests/sqlalchemy/test_exc_filters.py
+++ b/oslo_db/tests/sqlalchemy/test_exc_filters.py
@@ -233,14 +233,14 @@ class TestReferenceErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase):
 
         meta = sqla.MetaData(bind=self.engine)
 
-        table_1 = sqla.Table(
+        self.table_1 = sqla.Table(
             "resource_foo", meta,
             sqla.Column("id", sqla.Integer, primary_key=True),
             sqla.Column("foo", sqla.Integer),
             mysql_engine='InnoDB',
             mysql_charset='utf8',
         )
-        table_1.create()
+        self.table_1.create()
 
         self.table_2 = sqla.Table(
             "resource_entity", meta,
@@ -274,6 +274,30 @@ class TestReferenceErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase):
         self.assertIsNone(matched.key)
         self.assertIsNone(matched.key_table)
 
+    def test_raise_delete(self):
+        self.engine.execute("PRAGMA foreign_keys = ON;")
+
+        with self.engine.connect() as conn:
+            conn.execute(self.table_1.insert({"id": 1234, "foo": 42}))
+            conn.execute(self.table_2.insert({"id": 4321, "foo_id": 1234}))
+        matched = self.assertRaises(
+            exception.DBReferenceError,
+            self.engine.execute,
+            self.table_1.delete()
+        )
+        self.assertInnerException(
+            matched,
+            "IntegrityError",
+            "foreign key constraint failed",
+            "DELETE FROM resource_foo",
+            (),
+        )
+
+        self.assertIsNone(matched.table)
+        self.assertIsNone(matched.constraint)
+        self.assertIsNone(matched.key)
+        self.assertIsNone(matched.key_table)
+
 
 class TestReferenceErrorPostgreSQL(TestReferenceErrorSQLite,
                                    test_base.PostgreSQLOpportunisticTestCase):
@@ -300,6 +324,31 @@ class TestReferenceErrorPostgreSQL(TestReferenceErrorSQLite,
         self.assertEqual("foo_id", matched.key)
         self.assertEqual("resource_foo", matched.key_table)
 
+    def test_raise_delete(self):
+        with self.engine.connect() as conn:
+            conn.execute(self.table_1.insert({"id": 1234, "foo": 42}))
+            conn.execute(self.table_2.insert({"id": 4321, "foo_id": 1234}))
+        matched = self.assertRaises(
+            exception.DBReferenceError,
+            self.engine.execute,
+            self.table_1.delete()
+        )
+        self.assertInnerException(
+            matched,
+            "IntegrityError",
+            "update or delete on table \"resource_foo\" violates foreign key "
+            "constraint \"foo_fkey\" on table \"resource_entity\"\n"
+            "DETAIL:  Key (id)=(1234) is still referenced from "
+            "table \"resource_entity\".\n",
+            "DELETE FROM resource_foo",
+            {},
+        )
+
+        self.assertEqual("resource_foo", matched.table)
+        self.assertEqual("foo_fkey", matched.constraint)
+        self.assertEqual("id", matched.key)
+        self.assertEqual("resource_entity", matched.key_table)
+
 
 class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
                               test_base.MySQLOpportunisticTestCase):
@@ -353,6 +402,32 @@ class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
         self.assertEqual("foo_id", matched.key)
         self.assertEqual("resource_foo", matched.key_table)
 
+    def test_raise_delete(self):
+        with self.engine.connect() as conn:
+            conn.execute(self.table_1.insert({"id": 1234, "foo": 42}))
+            conn.execute(self.table_2.insert({"id": 4321, "foo_id": 1234}))
+        matched = self.assertRaises(
+            exception.DBReferenceError,
+            self.engine.execute,
+            self.table_1.delete()
+        )
+        self.assertInnerException(
+            matched,
+            "IntegrityError",
+            "(1451, 'cannot delete or update a parent row: a foreign key "
+            "constraint fails (`{0}`.`resource_entity`, "
+            "constraint `foo_fkey` "
+            "foreign key (`foo_id`) references "
+            "`resource_foo` (`id`))')".format(self.engine.url.database),
+            "DELETE FROM resource_foo",
+            (),
+        )
+
+        self.assertEqual("resource_entity", matched.table)
+        self.assertEqual("foo_fkey", matched.constraint)
+        self.assertEqual("foo_id", matched.key)
+        self.assertEqual("resource_foo", matched.key_table)
+
 
 class TestDuplicate(TestsExceptionFilter):