Assert exceptions based on API, not string messages

The tests/sqlalchemy/test_exc_filters.py tests were asserting
the exact string format of SQLAlchemy exceptions.
SQLAlchemy is improving its exception format in version 1.0,
thereby breaking these tests.  Instead of supporting multiple
string formats, change the assertions to compare specific
fields per the oslo.db.exceptions and sqlalchemy.exc APIs.

Change-Id: Ie98ac3ca82a1052c463ca9ca5d41e228935497b3

1 files changed, 129 insertions, 98 deletions

diff --git a/tests/sqlalchemy/test_exc_filters.py b/tests/sqlalchemy/test_exc_filters.py
index 68c583c..cb3d7ef 100644
--- a/tests/sqlalchemy/test_exc_filters.py
+++ b/tests/sqlalchemy/test_exc_filters.py
@@ -31,7 +31,29 @@ from tests import utils as test_utils
 _TABLE_NAME = '__tmp__test__tmp__'
 
 
-class TestsExceptionFilter(oslo_test_base.BaseTestCase):
+class _SQLAExceptionMatcher(object):
+    def assertInnerException(
+            self,
+            matched, exception_type, message, sql=None, params=None):
+
+        exc = matched.inner_exception
+        self.assertSQLAException(exc, exception_type, message, sql, params)
+
+    def assertSQLAException(
+            self,
+            exc, exception_type, message, sql=None, params=None):
+        if isinstance(exception_type, (type, tuple)):
+            self.assertTrue(issubclass(exc.__class__, exception_type))
+        else:
+            self.assertEqual(exc.__class__.__name__, exception_type)
+        self.assertEqual(str(exc.orig).lower(), message.lower())
+        if sql is not None:
+            self.assertEqual(exc.statement, sql)
+        if params is not None:
+            self.assertEqual(exc.params, params)
+
+
+class TestsExceptionFilter(_SQLAExceptionMatcher, oslo_test_base.BaseTestCase):
 
     class Error(Exception):
         """DBAPI base error.
@@ -127,10 +149,11 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
             self.ProgrammingError("Error 123, you made a mistake"),
             exception.DBError
         )
-        self.assertEqual(
-            "(ProgrammingError) Error 123, you made a "
-            "mistake 'select you_made_a_programming_error' ()",
-            matched.args[0])
+        self.assertInnerException(
+            matched,
+            "ProgrammingError",
+            "Error 123, you made a mistake",
+            'select you_made_a_programming_error', ())
 
     def test_generic_dbapi_disconnect(self):
         matched = self._run_test(
@@ -139,10 +162,10 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
             exception.DBConnectionError,
             is_disconnect=True
         )
-        self.assertEqual(
-            "(InterfaceError) connection lost "
-            "'select the_db_disconnected' ()",
-            matched.args[0])
+        self.assertInnerException(
+            matched,
+            "InterfaceError", "connection lost",
+            "select the_db_disconnected", ()),
 
     def test_operational_dbapi_disconnect(self):
         matched = self._run_test(
@@ -151,10 +174,10 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
             exception.DBConnectionError,
             is_disconnect=True
         )
-        self.assertEqual(
-            "(OperationalError) connection lost "
-            "'select the_db_disconnected' ()",
-            matched.args[0])
+        self.assertInnerException(
+            matched,
+            "OperationalError", "connection lost",
+            "select the_db_disconnected", ()),
 
     def test_operational_error_asis(self):
         """Test operational errors.
@@ -168,9 +191,10 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
             self.OperationalError("some op error"),
             sqla.exc.OperationalError
         )
-        self.assertEqual(
-            "(OperationalError) some op error",
-            matched.args[0])
+        self.assertSQLAException(
+            matched,
+            "OperationalError", "some op error"
+        )
 
     def test_unicode_encode(self):
         # intentionally generate a UnicodeEncodeError, as its
@@ -196,7 +220,7 @@ class TestFallthroughsAndNonDBAPI(TestsExceptionFilter):
         self.assertEqual("mysqldb has an attribute error", matched.args[0])
 
 
-class TestReferenceErrorSQLite(test_base.DbTestCase):
+class TestReferenceErrorSQLite(_SQLAExceptionMatcher, test_base.DbTestCase):
 
     def setUp(self):
         super(TestReferenceErrorSQLite, self).setUp()
@@ -225,83 +249,98 @@ class TestReferenceErrorSQLite(test_base.DbTestCase):
     def test_raise(self):
         self.engine.execute("PRAGMA foreign_keys = ON;")
 
-        e = self.assertRaises(
+        matched = self.assertRaises(
             exception.DBReferenceError,
             self.engine.execute,
             self.table_2.insert({'id': 1, 'foo_id': 2})
         )
 
-        self.assertEqual(
-            "(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)
+        self.assertInnerException(
+            matched,
+            "IntegrityError",
+            "FOREIGN KEY constraint failed",
+            'INSERT INTO resource_entity (id, foo_id) VALUES (?, ?)',
+            (1, 2)
+        )
+
+        self.assertIsNone(matched.table)
+        self.assertIsNone(matched.constraint)
+        self.assertIsNone(matched.key)
+        self.assertIsNone(matched.key_table)
 
 
 class TestReferenceErrorPostgreSQL(TestReferenceErrorSQLite,
                                    test_base.PostgreSQLOpportunisticTestCase):
     def test_raise(self):
-        e = self.assertRaises(
+        params = {'id': 1, 'foo_id': 2}
+        matched = self.assertRaises(
             exception.DBReferenceError,
             self.engine.execute,
-            self.table_2.insert({'id': 1, 'foo_id': 2})
+            self.table_2.insert(params)
         )
-
-        self.assertIn(
-            "(IntegrityError) insert or update on table \"resource_entity\" "
+        self.assertInnerException(
+            matched,
+            "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("foo_fkey", e.constraint)
-        self.assertEqual("foo_id", e.key)
-        self.assertEqual("resource_foo", e.key_table)
+            "(foo_id)=(2) is not present in table \"resource_foo\".\n",
+            "INSERT INTO resource_entity (id, foo_id) VALUES (%(id)s, "
+            "%(foo_id)s)",
+            params,
+        )
+
+        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 TestReferenceErrorMySQL(TestReferenceErrorSQLite,
                               test_base.MySQLOpportunisticTestCase):
     def test_raise(self):
-        e = self.assertRaises(
+        matched = self.assertRaises(
             exception.DBReferenceError,
             self.engine.execute,
             self.table_2.insert({'id': 1, 'foo_id': 2})
         )
 
-        self.assertEqual(
-            "(IntegrityError) (1452, 'Cannot add or update a child row: a "
+        self.assertInnerException(
+            matched,
+            "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)
+            "`resource_foo` (`id`))')".format(self.engine.url.database),
+            "INSERT INTO resource_entity (id, foo_id) VALUES (%s, %s)",
+            (1, 2)
+        )
+        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)
 
     def test_raise_ansi_quotes(self):
         self.engine.execute("SET SESSION sql_mode = 'ANSI';")
-        e = self.assertRaises(
+        matched = self.assertRaises(
             exception.DBReferenceError,
             self.engine.execute,
             self.table_2.insert({'id': 1, 'foo_id': 2})
         )
 
-        self.assertEqual(
-            "(IntegrityError) (1452, 'Cannot add or update a child row: a "
+        self.assertInnerException(
+            matched,
+            "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)
+            '"resource_foo" ("id"))\')'.format(self.engine.url.database),
+            "INSERT INTO resource_entity (id, foo_id) VALUES (%s, %s)",
+            (1, 2)
+        )
+        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):
@@ -318,13 +357,18 @@ class TestDuplicate(TestsExceptionFilter):
         self.assertEqual(expected_value, matched.value)
 
     def _not_dupe_constraint_test(self, dialect_name, statement, message,
-                                  expected_cls, expected_message):
+                                  expected_cls):
         matched = self._run_test(
             dialect_name, statement,
             self.IntegrityError(message),
             expected_cls
         )
-        self.assertEqual(expected_message, matched.args[0])
+        self.assertInnerException(
+            matched,
+            "IntegrityError",
+            str(self.IntegrityError(message)),
+            statement
+        )
 
     def test_sqlite(self):
         self._run_dupe_constraint_test("sqlite", 'column a, b are not unique')
@@ -376,9 +420,7 @@ class TestDuplicate(TestsExceptionFilter):
         self._not_dupe_constraint_test(
             "nonexistent", "insert into table some_values",
             self.IntegrityError("constraint violation"),
-            exception.DBError,
-            "(IntegrityError) constraint violation "
-            "'insert into table some_values' ()"
+            exception.DBError
         )
 
     def test_ibm_db_sa(self):
@@ -400,56 +442,47 @@ class TestDuplicate(TestsExceptionFilter):
             'SQL0542N  The column named "NAME" cannot be a column of a '
             'primary key or unique key constraint because it can contain null '
             'values.',
-            exception.DBError,
-            '(IntegrityError) SQL0542N  The column named "NAME" cannot be a '
-            'column of a primary key or unique key constraint because it can '
-            'contain null values. \'ALTER TABLE instance_types ADD CONSTRAINT '
-            'uniq_name_x_deleted UNIQUE (name, deleted)\' ()'
+            exception.DBError
         )
 
 
 class TestDeadlock(TestsExceptionFilter):
+    statement = ('SELECT quota_usages.created_at AS '
+                 'quota_usages_created_at FROM quota_usages '
+                 'WHERE quota_usages.project_id = %(project_id_1)s '
+                 'AND quota_usages.deleted = %(deleted_1)s FOR UPDATE')
+    params = {
+        'project_id_1': '8891d4478bbf48ad992f050cdf55e9b5',
+        'deleted_1': 0
+    }
+
     def _run_deadlock_detect_test(
         self, dialect_name, message,
         orig_exception_cls=TestsExceptionFilter.OperationalError):
-        statement = ('SELECT quota_usages.created_at AS '
-                     'quota_usages_created_at FROM quota_usages \n'
-                     'WHERE quota_usages.project_id = %(project_id_1)s '
-                     'AND quota_usages.deleted = %(deleted_1)s FOR UPDATE')
-        params = {
-            'project_id_1': '8891d4478bbf48ad992f050cdf55e9b5',
-            'deleted_1': 0
-        }
         self._run_test(
-            dialect_name, statement,
+            dialect_name, self.statement,
             orig_exception_cls(message),
             exception.DBDeadlock,
-            params=params
+            params=self.params
         )
 
     def _not_deadlock_test(
         self, dialect_name, message,
-        expected_cls, expected_message,
+        expected_cls, expected_dbapi_cls,
         orig_exception_cls=TestsExceptionFilter.OperationalError):
-        statement = ('SELECT quota_usages.created_at AS '
-                     'quota_usages_created_at FROM quota_usages \n'
-                     'WHERE quota_usages.project_id = %%(project_id_1)s '
-                     'AND quota_usages.deleted = %%(deleted_1)s FOR UPDATE')
-        params = {
-            'project_id_1': '8891d4478bbf48ad992f050cdf55e9b5',
-            'deleted_1': 0
-        }
+
         matched = self._run_test(
-            dialect_name, statement,
+            dialect_name, self.statement,
             orig_exception_cls(message),
             expected_cls,
-            params=params
-        )
-        self.assertEqual(
-            expected_message % {'statement': statement, 'params': params},
-            str(matched)
+            params=self.params
         )
 
+        if isinstance(matched, exception.DBError):
+            matched = matched.inner_exception
+
+        self.assertEqual(matched.orig.__class__.__name__, expected_dbapi_cls)
+
     def test_mysql_mysqldb_deadlock(self):
         self._run_deadlock_detect_test(
             "mysql",
@@ -471,8 +504,7 @@ class TestDeadlock(TestsExceptionFilter):
             "mysql",
             "(1005, 'some other error')",
             sqla.exc.OperationalError,  # note OperationalErrors are sent thru
-            "(OperationalError) (1005, 'some other error') "
-            "%(statement)r %(params)r"
+            "OperationalError",
         )
 
     def test_postgresql_deadlock(self):
@@ -488,8 +520,7 @@ class TestDeadlock(TestsExceptionFilter):
             'relation "fake" does not exist',
             # can be either depending on #3075
             (exception.DBError, sqla.exc.OperationalError),
-            '(TransactionRollbackError) '
-            'relation "fake" does not exist %(statement)r %(params)r',
+            "TransactionRollbackError",
             orig_exception_cls=self.TransactionRollbackError
         )
 
@@ -508,7 +539,7 @@ class TestDeadlock(TestsExceptionFilter):
             "ibm_db_sa",
             "SQL01234B Some other error.",
             exception.DBError,
-            "(Error) SQL01234B Some other error. %(statement)r %(params)r",
+            "Error",
             orig_exception_cls=self.Error
         )