Add support for utf-8 error messages

* Both server and client should be fine with utf-8 error messages now
* Adding an interop test: special status message

5 files changed, 126 insertions, 23 deletions

diff --git a/src/python/grpcio/grpc/_common.py b/src/python/grpcio/grpc/_common.py
index d35f4566bd..42f3a4e614 100644
--- a/src/python/grpcio/grpc/_common.py
+++ b/src/python/grpcio/grpc/_common.py
@@ -66,18 +66,13 @@ def encode(s):
     if isinstance(s, bytes):
         return s
     else:
-        return s.encode('ascii')
+        return s.encode('utf8')
 
 
 def decode(b):
-    if isinstance(b, str):
-        return b
-    else:
-        try:
-            return b.decode('utf8')
-        except UnicodeDecodeError:
-            _LOGGER.exception('Invalid encoding on %s', b)
-            return b.decode('latin1')
+    if isinstance(b, bytes):
+        return b.decode('utf-8', 'replace')
+    return b
 
 
 def _transform(message, transformer, exception_message):
diff --git a/src/python/grpcio_tests/tests/interop/methods.py b/src/python/grpcio_tests/tests/interop/methods.py
index cda15a68a3..721dedf0b7 100644
--- a/src/python/grpcio_tests/tests/interop/methods.py
+++ b/src/python/grpcio_tests/tests/interop/methods.py
@@ -457,6 +457,22 @@ def _per_rpc_creds(stub, args):
                                                            response.username))
 
 
+def _special_status_message(stub, args):
+    details = b'\t\ntest with whitespace\r\nand Unicode BMP \xe2\x98\xba and non-BMP \xf0\x9f\x98\x88\t\n'.decode(
+        'utf-8')
+    code = 2
+    status = grpc.StatusCode.UNKNOWN  # code = 2
+
+    # Test with a UnaryCall
+    request = messages_pb2.SimpleRequest(
+        response_type=messages_pb2.COMPRESSABLE,
+        response_size=1,
+        payload=messages_pb2.Payload(body=b'\x00'),
+        response_status=messages_pb2.EchoStatus(code=code, message=details))
+    response_future = stub.UnaryCall.future(request)
+    _validate_status_code_and_details(response_future, status, details)
+
+
 @enum.unique
 class TestCase(enum.Enum):
     EMPTY_UNARY = 'empty_unary'
@@ -476,6 +492,7 @@ class TestCase(enum.Enum):
     JWT_TOKEN_CREDS = 'jwt_token_creds'
     PER_RPC_CREDS = 'per_rpc_creds'
     TIMEOUT_ON_SLEEPING_SERVER = 'timeout_on_sleeping_server'
+    SPECIAL_STATUS_MESSAGE = 'special_status_message'
 
     def test_interoperability(self, stub, args):
         if self is TestCase.EMPTY_UNARY:
@@ -512,6 +529,8 @@ class TestCase(enum.Enum):
             _jwt_token_creds(stub, args)
         elif self is TestCase.PER_RPC_CREDS:
             _per_rpc_creds(stub, args)
+        elif self is TestCase.SPECIAL_STATUS_MESSAGE:
+            _special_status_message(stub, args)
         else:
             raise NotImplementedError(
                 'Test case "%s" not implemented!' % self.name)
diff --git a/src/python/grpcio_tests/tests/tests.json b/src/python/grpcio_tests/tests/tests.json
index 5505369867..c5ea8c5fbb 100644
--- a/src/python/grpcio_tests/tests/tests.json
+++ b/src/python/grpcio_tests/tests/tests.json
@@ -42,6 +42,7 @@
   "unit._cython.cygrpc_test.SecureServerSecureClient",
   "unit._cython.cygrpc_test.TypeSmokeTest",
   "unit._empty_message_test.EmptyMessageTest",
+  "unit._error_message_encoding_test.ErrorMessageEncodingTest",
   "unit._exit_test.ExitTest",
   "unit._interceptor_test.InterceptorTest",
   "unit._invalid_metadata_test.InvalidMetadataTest",
diff --git a/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py b/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py
new file mode 100644
index 0000000000..6c551df3ec
--- /dev/null
+++ b/src/python/grpcio_tests/tests/unit/_error_message_encoding_test.py
@@ -0,0 +1,86 @@
+# Copyright 2018 gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests 'utf-8' encoded error message."""
+
+import unittest
+import weakref
+
+import grpc
+
+from tests.unit import test_common
+from tests.unit.framework.common import test_constants
+
+_UNICODE_ERROR_MESSAGES = [
+    b'\xe2\x80\x9d'.decode('utf-8'),
+    b'abc\x80\xd0\xaf'.decode('latin-1'),
+    b'\xc3\xa9'.decode('utf-8'),
+]
+
+_REQUEST = b'\x00\x00\x00'
+_RESPONSE = b'\x00\x00\x00'
+
+_UNARY_UNARY = '/test/UnaryUnary'
+
+
+class _MethodHandler(grpc.RpcMethodHandler):
+
+    def __init__(self, request_streaming=None, response_streaming=None):
+        self.request_streaming = request_streaming
+        self.response_streaming = response_streaming
+        self.request_deserializer = None
+        self.response_serializer = None
+        self.unary_stream = None
+        self.stream_unary = None
+        self.stream_stream = None
+
+    def unary_unary(self, request, servicer_context):
+        servicer_context.set_code(grpc.StatusCode.UNKNOWN)
+        servicer_context.set_details(request.decode('utf-8'))
+        return _RESPONSE
+
+
+class _GenericHandler(grpc.GenericRpcHandler):
+
+    def __init__(self, test):
+        self._test = test
+
+    def service(self, handler_call_details):
+        return _MethodHandler()
+
+
+class ErrorMessageEncodingTest(unittest.TestCase):
+
+    def setUp(self):
+        self._server = test_common.test_server()
+        self._server.add_generic_rpc_handlers((_GenericHandler(
+            weakref.proxy(self)),))
+        port = self._server.add_insecure_port('[::]:0')
+        self._server.start()
+        self._channel = grpc.insecure_channel('localhost:%d' % port)
+
+    def tearDown(self):
+        self._server.stop(0)
+
+    def testMessageEncoding(self):
+        for message in _UNICODE_ERROR_MESSAGES:
+            multi_callable = self._channel.unary_unary(_UNARY_UNARY)
+            with self.assertRaises(grpc.RpcError) as cm:
+                multi_callable(message.encode('utf-8'))
+
+            self.assertEqual(cm.exception.code(), grpc.StatusCode.UNKNOWN)
+            self.assertEqual(cm.exception.details(), message)
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)
diff --git a/tools/run_tests/run_interop_tests.py b/tools/run_tests/run_interop_tests.py
index 5722a88182..11f483e083 100755
--- a/tools/run_tests/run_interop_tests.py
+++ b/tools/run_tests/run_interop_tests.py
@@ -63,6 +63,8 @@ _SKIP_ADVANCED = [
     'unimplemented_service'
 ]
 
+_SKIP_SPECIAL_STATUS_MESSAGE = ['special_status_message']
+
 _TEST_TIMEOUT = 3 * 60
 
 # disable this test on core-based languages,
@@ -100,7 +102,7 @@ class CXXLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_DATA_FRAME_PADDING
+        return _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return []
@@ -129,7 +131,7 @@ class CSharpLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _SKIP_COMPRESSION
@@ -158,7 +160,7 @@ class CSharpCoreCLRLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _SKIP_COMPRESSION
@@ -188,10 +190,10 @@ class DartLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_COMPRESSION
+        return _SKIP_COMPRESSION + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
-        return _SKIP_COMPRESSION
+        return _SKIP_COMPRESSION + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def __str__(self):
         return 'dart'
@@ -248,7 +250,7 @@ class JavaOkHttpClient:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_DATA_FRAME_PADDING
+        return _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def __str__(self):
         return 'javaokhttp'
@@ -309,7 +311,7 @@ class Http2Server:
         return {}
 
     def unimplemented_test_cases(self):
-        return _TEST_CASES + _SKIP_DATA_FRAME_PADDING
+        return _TEST_CASES + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _TEST_CASES
@@ -339,7 +341,7 @@ class Http2Client:
         return {}
 
     def unimplemented_test_cases(self):
-        return _TEST_CASES
+        return _TEST_CASES + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _TEST_CASES
@@ -431,7 +433,7 @@ class PHPLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return []
@@ -456,7 +458,7 @@ class PHP7Language:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return []
@@ -491,7 +493,7 @@ class ObjcLanguage:
         # cmdline argument. Here we return all but one test cases as unimplemented,
         # and depend upon ObjC test's behavior that it runs all cases even when
         # we tell it to run just one.
-        return _TEST_CASES[1:] + _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _TEST_CASES[1:] + _SKIP_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _SKIP_COMPRESSION
@@ -526,7 +528,7 @@ class RubyLanguage:
         return {}
 
     def unimplemented_test_cases(self):
-        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING
+        return _SKIP_SERVER_COMPRESSION + _SKIP_DATA_FRAME_PADDING + _SKIP_SPECIAL_STATUS_MESSAGE
 
     def unimplemented_test_cases_server(self):
         return _SKIP_COMPRESSION
@@ -610,7 +612,7 @@ _TEST_CASES = [
     'custom_metadata', 'status_code_and_message', 'unimplemented_method',
     'client_compressed_unary', 'server_compressed_unary',
     'client_compressed_streaming', 'server_compressed_streaming',
-    'unimplemented_service'
+    'unimplemented_service', 'special_status_message'
 ]
 
 _AUTH_TEST_CASES = [
@@ -1315,7 +1317,7 @@ try:
             for language in languages:
                 for test_case in _TEST_CASES:
                     if not test_case in language.unimplemented_test_cases():
-                        if not test_case in _SKIP_ADVANCED + _SKIP_COMPRESSION:
+                        if not test_case in _SKIP_ADVANCED + _SKIP_COMPRESSION + _SKIP_SPECIAL_STATUS_MESSAGE:
                             tls_test_job = cloud_to_prod_jobspec(
                                 language,
                                 test_case,