Fix DBReferenceError on MySQL and SQLite

It turns out the test was wrong as the error message used was not a good
one. Real world tests showed that, so I've updated the test and the code
to work.

Refactored tests to use real backends, fix error message for SQLite.

Co-Authored-By: Victor Sergeyev <vsergeyev@mirantis.com>

Change-Id: Ifa64478a9ac9ed0a3c317a8974615b770c85f313

2 files changed, 98 insertions, 73 deletions

diff --git a/oslo/db/sqlalchemy/exc_filters.py b/oslo/db/sqlalchemy/exc_filters.py
index c03c4b4..b3da401 100644
--- a/oslo/db/sqlalchemy/exc_filters.py
+++ b/oslo/db/sqlalchemy/exc_filters.py
@@ -169,7 +169,7 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
 
 
 @filters("sqlite", sqla_exc.IntegrityError,
-         r".*SQL error: foreign key constraint failed")
+         r"(?i).*foreign key constraint failed")
 @filters("postgresql", sqla_exc.IntegrityError,
          r".*on table \"(?P<table>[^\"]+)\" violates "
          "foreign key constraint \"(?P<constraint>[^\"]+)\"\s*\n"
@@ -177,13 +177,13 @@ def _sqlite_dupe_key_error(integrity_error, match, engine_name, is_disconnect):
          "is not present in table "
          "\"(?P<key_table>[^\"]+)\".")
 @filters("mysql", sqla_exc.IntegrityError,
-         r".* Cannot add or update a child row: "
-         "a foreign key constraint fails "
-         "\((?P<table>.+), CONSTRAINT (?P<constraint>.+) "
-         "FOREIGN KEY \((?P<key>.+)\) "
-         "REFERENCES (?P<key_table>.+) \(.+\)\)")
+         r".* 'Cannot add or update a child row: "
+         'a foreign key constraint fails \([`"].+[`"]\.[`"](?P<table>.+)[`"], '
+         'CONSTRAINT [`"](?P<constraint>.+)[`"] FOREIGN KEY '
+         '\([`"](?P<key>.+)[`"]\) REFERENCES [`"](?P<key_table>.+)[`"] ')
 def _foreign_key_error(integrity_error, match, engine_name, is_disconnect):
     """Filter for foreign key errors."""
+
     try:
         table = match.group("table")
     except IndexError:
diff --git a/tests/sqlalchemy/test_exc_filters.py b/tests/sqlalchemy/test_exc_filters.py
index b6b1375..d0ea38b 100644
--- a/tests/sqlalchemy/test_exc_filters.py
+++ b/tests/sqlalchemy/test_exc_filters.py
@@ -195,87 +195,112 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
         self.assertEqual("mysqldb has an attribute error", matched.args[0])
 
 
-class TestRaiseReferenceError(TestsExceptionFilter):
-    def test_postgresql(self):
-        e = self._run_test(
-            "postgresql",
-            "INSERT SOMETHING",
-            self.IntegrityError(
-                "insert or update on table "
-                "\"resource_entity\" "
-                "violates foreign key constraint "
-                "\"resource_entity_entity_id_fkey\"\n"
-                "DETAIL:  Key "
-                "(entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
-                "is not present in table \"entity\".\n"
-                "'INSERT INTO resource_entity (resource_id, entity_id, name) "
-                "VALUES (%(resource_id)s, "
-                "%(entity_id)s, %(name)s)' "
-                "{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
-                "'name': u'foo', "
-                "'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'}"),
+class TestReferenceErrorSQLite(test_base.DbTestCase):
+
+    def setUp(self):
+        super(TestReferenceErrorSQLite, self).setUp()
+
+        meta = sqla.MetaData(bind=self.engine)
+
+        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_2 = sqla.Table(
+            "resource_entity", meta,
+            sqla.Column("id", sqla.Integer, primary_key=True),
+            sqla.Column("foo_id", sqla.Integer,
+                        sqla.ForeignKey("resource_foo.id", name="foo_fkey")),
+            mysql_engine='InnoDB',
+            mysql_charset='utf8',
+        )
+        self.table_2.create()
+
+    def test_raise(self):
+        self.engine.execute("PRAGMA foreign_keys = ON;")
+
+        e = self.assertRaises(
             exception.DBReferenceError,
+            self.engine.execute,
+            self.table_2.insert({'id': 1, 'foo_id': 2})
         )
-        self.assertEqual("resource_entity", e.table)
-        self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
-        self.assertEqual("entity_id", e.key)
-        self.assertEqual("entity", e.key_table)
+
         self.assertEqual(
-            "(IntegrityError) insert or update on table "
-            "\"resource_entity\" violates foreign key constraint "
-            "\"resource_entity_entity_id_fkey\"\n"
-            "DETAIL:  Key (entity_id)=(74b5da71-5a9c-4f89-a8e9-4a2d856e6c29) "
-            "is not present in table \"entity\".\n"
-            "'INSERT INTO resource_entity (resource_id, entity_id, name) "
-            "VALUES (%(resource_id)s, %(entity_id)s, %(name)s)' "
-            "{'entity_id': '74b5da71-5a9c-4f89-a8e9-4a2d856e6c29', "
-            "'name': u'foo', "
-            "'resource_id': 'ffb12cb4-d955-4d96-a315-5f48ea161eef'} "
-            "'INSERT SOMETHING' ()",
-            str(e))
+            "(IntegrityError) FOREIGN KEY constraint failed u'INSERT INTO "
+            "resource_entity (id, foo_id) VALUES (?, ?)' (1, 2)".lower(),
+            str(e).lower())
+        self.assertIsNone(e.table)
+        self.assertIsNone(e.constraint)
+        self.assertIsNone(e.key)
+        self.assertIsNone(e.key_table)
 
-    def test_mysql(self):
-        e = self._run_test(
-            "mysql",
-            "INSERT SOMETHING",
-            self.IntegrityError(
-                "Cannot add or update a child row: "
-                "a foreign key constraint fails "
-                "(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
-                "FOREIGN KEY (entity_id) "
-                "REFERENCES entity (entity_id))"
-            ),
+
+class TestReferenceErrorPostgreSQL(TestReferenceErrorSQLite,
+                                   test_base.PostgreSQLOpportunisticTestCase):
+    def test_raise(self):
+        e = self.assertRaises(
             exception.DBReferenceError,
+            self.engine.execute,
+            self.table_2.insert({'id': 1, 'foo_id': 2})
         )
+
+        self.assertIn(
+            "(IntegrityError) insert or update on table \"resource_entity\" "
+            "violates foreign key constraint \"foo_fkey\"\nDETAIL:  Key "
+            "(foo_id)=(2) is not present in table \"resource_foo\".\n"
+            " 'INSERT INTO resource_entity (id, foo_id) VALUES (%(id)s, "
+            "%(foo_id)s)'", str(e))
         self.assertEqual("resource_entity", e.table)
-        self.assertEqual("resource_entity_entity_id_fkey", e.constraint)
-        self.assertEqual("entity_id", e.key)
-        self.assertEqual("entity", e.key_table)
+        self.assertEqual("foo_fkey", e.constraint)
+        self.assertEqual("foo_id", e.key)
+        self.assertEqual("resource_foo", e.key_table)
+
+
+class TestReferenceErrorMySQL(TestReferenceErrorSQLite,
+                              test_base.MySQLOpportunisticTestCase):
+    def test_raise(self):
+        e = self.assertRaises(
+            exception.DBReferenceError,
+            self.engine.execute,
+            self.table_2.insert({'id': 1, 'foo_id': 2})
+        )
+
         self.assertEqual(
-            "(IntegrityError) Cannot add or update a child row: "
-            "a foreign key constraint fails "
-            "(resource_entity, CONSTRAINT resource_entity_entity_id_fkey "
-            "FOREIGN KEY (entity_id) REFERENCES entity (entity_id)) "
-            "'INSERT SOMETHING' ()",
+            "(IntegrityError) (1452, 'Cannot add or update a child row: a "
+            "foreign key constraint fails (`{0}`.`resource_entity`, "
+            "CONSTRAINT `foo_fkey` FOREIGN KEY (`foo_id`) REFERENCES "
+            "`resource_foo` (`id`))') 'INSERT INTO resource_entity (id, foo_id"
+            ") VALUES (%s, %s)' (1, 2)".format(self.engine.url.database),
             str(e))
+        self.assertEqual("resource_entity", e.table)
+        self.assertEqual("foo_fkey", e.constraint)
+        self.assertEqual("foo_id", e.key)
+        self.assertEqual("resource_foo", e.key_table)
 
-    def test_sqlite(self):
-        e = self._run_test(
-            "sqlite",
-            "INSERT SOMETHING",
-            self.IntegrityError(
-                "SQL error: foreign key constraint failed"
-            ),
+    def test_raise_ansi_quotes(self):
+        self.engine.execute("SET SESSION sql_mode = 'ANSI';")
+        e = self.assertRaises(
             exception.DBReferenceError,
+            self.engine.execute,
+            self.table_2.insert({'id': 1, 'foo_id': 2})
         )
-        self.assertIsNone(e.table)
-        self.assertIsNone(e.constraint)
-        self.assertIsNone(e.key)
-        self.assertIsNone(e.key_table)
+
         self.assertEqual(
-            "(IntegrityError) SQL error: foreign key "
-            "constraint failed 'INSERT SOMETHING' ()",
+            "(IntegrityError) (1452, 'Cannot add or update a child row: a "
+            'foreign key constraint fails ("{0}"."resource_entity", '
+            'CONSTRAINT "foo_fkey" FOREIGN KEY ("foo_id") REFERENCES '
+            '"resource_foo" ("id"))\') \'INSERT INTO resource_entity (id, '
+            "foo_id) VALUES (%s, %s)' (1, 2)".format(self.engine.url.database),
             str(e))
+        self.assertEqual("resource_entity", e.table)
+        self.assertEqual("foo_fkey", e.constraint)
+        self.assertEqual("foo_id", e.key)
+        self.assertEqual("resource_foo", e.key_table)
 
 
 class TestDuplicate(TestsExceptionFilter):