Leave bytestring exception messages as bytestrings

Fixed a regression introduced in version 1.2 where a refactor
of the :class:`.SQLAlchemyError` base exception class introduced an
inappropriate coercion of a plain string message into Unicode under
python 2k, which is not handled by the Python interpreter for characters
outside of the platform's encoding (typically ascii).  The
:class:`.SQLAlchemyError` class now passes a bytestring through under
Py2K for ``__str__()`` as is the behavior of exception objects in general
under Py2K, does a safe coercion to unicode utf-8 with
backslash fallback for ``__unicode__()``.  For Py3K the message is
typically unicode already, but if not is again safe-coerced with utf-8
with backslash fallback for the ``__str__()`` method.

Fixes: #4429
Change-Id: I2289da3f2c45c7d0041fa43d838958f7614defc3

2 files changed, 104 insertions, 1 deletions

diff --git a/test/base/test_utils.py b/test/base/test_utils.py
index 20b41101e..69af6e032 100644
--- a/test/base/test_utils.py
+++ b/test/base/test_utils.py
@@ -1,3 +1,5 @@
+#! coding: utf-8
+
 import copy
 import inspect
 import sys
@@ -2552,3 +2554,47 @@ class QuotedTokenParserTest(fixtures.TestBase):
 
     def test_quoted_single_w_dot_middle(self):
         self._test('"na.me"', ["na.me"])
+
+
+class BackslashReplaceTest(fixtures.TestBase):
+    def test_ascii_to_utf8(self):
+        eq_(
+            compat.decode_backslashreplace(util.b("hello world"), "utf-8"),
+            util.u("hello world"),
+        )
+
+    def test_utf8_to_utf8(self):
+        eq_(
+            compat.decode_backslashreplace(
+                util.u("some message méil").encode("utf-8"), "utf-8"
+            ),
+            util.u("some message méil"),
+        )
+
+    def test_latin1_to_utf8(self):
+        eq_(
+            compat.decode_backslashreplace(
+                util.u("some message méil").encode("latin-1"), "utf-8"
+            ),
+            util.u("some message m\\xe9il"),
+        )
+
+        eq_(
+            compat.decode_backslashreplace(
+                util.u("some message méil").encode("latin-1"), "latin-1"
+            ),
+            util.u("some message méil"),
+        )
+
+    def test_cp1251_to_utf8(self):
+        message = util.u("some message П").encode("cp1251")
+        eq_(message, b"some message \xcf")
+        eq_(
+            compat.decode_backslashreplace(message, "utf-8"),
+            util.u("some message \\xcf"),
+        )
+
+        eq_(
+            compat.decode_backslashreplace(message, "cp1251"),
+            util.u("some message П"),
+        )
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index d9d10a9a6..8613be5bc 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -406,7 +406,7 @@ class ExecuteTest(fixtures.TestBase):
                     obj,
                 )
 
-    def test_stmt_exception_non_ascii(self):
+    def test_stmt_exception_bytestring_raised(self):
         name = util.u("méil")
         with testing.db.connect() as conn:
             assert_raises_message(
@@ -427,6 +427,63 @@ class ExecuteTest(fixtures.TestBase):
                 {"uname_incorrect": "foo"},
             )
 
+    def test_stmt_exception_bytestring_utf8(self):
+        # uncommon case for Py3K, bytestring object passed
+        # as the error message
+        message = util.u("some message méil").encode("utf-8")
+
+        err = tsa.exc.SQLAlchemyError(message)
+        if util.py2k:
+            # string passes it through
+            eq_(str(err), message)
+
+            # unicode accessor decodes to utf-8
+            eq_(unicode(err), util.u("some message méil"))  # noqa
+        else:
+            eq_(str(err), util.u("some message méil"))
+
+    def test_stmt_exception_bytestring_latin1(self):
+        # uncommon case for Py3K, bytestring object passed
+        # as the error message
+        message = util.u("some message méil").encode("latin-1")
+
+        err = tsa.exc.SQLAlchemyError(message)
+        if util.py2k:
+            # string passes it through
+            eq_(str(err), message)
+
+            # unicode accessor decodes to utf-8
+            eq_(unicode(err), util.u("some message m\\xe9il"))  # noqa
+        else:
+            eq_(str(err), util.u("some message m\\xe9il"))
+
+    def test_stmt_exception_unicode_hook_unicode(self):
+        # uncommon case for Py2K, Unicode object passed
+        # as the error message
+        message = util.u("some message méil")
+
+        err = tsa.exc.SQLAlchemyError(message)
+        if util.py2k:
+            eq_(unicode(err), util.u("some message méil"))  # noqa
+        else:
+            eq_(str(err), util.u("some message méil"))
+
+    def test_stmt_exception_str_multi_args(self):
+        err = tsa.exc.SQLAlchemyError("some message", 206)
+        eq_(str(err), "('some message', 206)")
+
+    def test_stmt_exception_str_multi_args_bytestring(self):
+        message = util.u("some message méil").encode("utf-8")
+
+        err = tsa.exc.SQLAlchemyError(message, 206)
+        eq_(str(err), str((message, 206)))
+
+    def test_stmt_exception_str_multi_args_unicode(self):
+        message = util.u("some message méil")
+
+        err = tsa.exc.SQLAlchemyError(message, 206)
+        eq_(str(err), str((message, 206)))
+
     def test_stmt_exception_pickleable_no_dbapi(self):
         self._test_stmt_exception_pickleable(Exception("hello world"))