Merge "Assert exceptions based on API, not string messages"1.1.0

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
         )