Fixed #3586 made celery respect exception types when using serializers (#5074)

* Fixed #3586 made celery respect exception types when using seriaizers

* isort

* isort reorder

* isort reorder

celery/backends/base.py

@@ -9,6 +9,7 @@
 from __future__ import absolute_import, unicode_literals
 
 import datetime
+import inspect
 import sys
 import time
 from collections import namedtuple
@@ -34,7 +35,6 @@ from celery.utils.collections import BufferMap
 from celery.utils.functional import LRUCache, arity_greater
 from celery.utils.log import get_logger
 from celery.utils.serialization import (create_exception_cls,
-                                        ensure_serializable,
                                         get_pickleable_exception,
                                         get_pickled_exception)
 
@@ -236,9 +236,14 @@ class Backend(object):
         serializer = self.serializer if serializer is None else serializer
         if serializer in EXCEPTION_ABLE_CODECS:
             return get_pickleable_exception(exc)
+        # retrieve exception original module
+        exc_module = inspect.getmodule(type(exc))
+        if exc_module:
+            exc_module = exc_module.__name__
+
         return {'exc_type': type(exc).__name__,
-                'exc_message': ensure_serializable(exc.args, self.encode),
-                'exc_module': type(exc).__module__}
+                'exc_args': exc.args,
+                'exc_module': exc_module}
 
     def exception_to_python(self, exc):
         """Convert serialized exception to Python exception."""

celery/utils/serialization.py

@@ -8,11 +8,11 @@ import sys
 from base64 import b64decode as base64decode
 from base64 import b64encode as base64encode
 from functools import partial
+from importlib import import_module
 from inspect import getmro
 from itertools import takewhile
 
 from kombu.utils.encoding import bytes_to_str, str_to_bytes
-
 from celery.five import (bytes_if_py2, items, python_2_unicode_compatible,
                          reraise, string_t)
 
@@ -81,6 +81,14 @@ def itermro(cls, stop):
 
 def create_exception_cls(name, module, parent=None):
     """Dynamically create an exception class."""
+    try:
+        mod = import_module(module)
+        exc_cls = getattr(mod, name, None)
+        if exc_cls and isinstance(exc_cls, type(BaseException)):
+            return exc_cls
+    except ImportError:
+        pass
+    # we could not find the exception, fallback and create a type.
     if not parent:
         parent = Exception
     return subclass_exception(name, parent, module)

t/unit/backends/test_base.py

@@ -225,6 +225,10 @@ class DictBackend(BaseBackend):
         self._data.pop(group_id, None)
 
 
+class CustomTestError(Exception):
+    pass
+
+
 class test_BaseBackend_dict:
 
     def setup(self):
@@ -245,13 +249,26 @@ class test_BaseBackend_dict:
         self.b.delete_group('can-delete')
         assert 'can-delete' not in self.b._data
 
-    def test_prepare_exception_json(self):
-        x = DictBackend(self.app, serializer='json')
-        e = x.prepare_exception(KeyError('foo'))
-        assert 'exc_type' in e
+    @pytest.mark.parametrize(("serializer"), (("pickle", "json")))
+    def test_prepare_builtin_exception(self, serializer):
+        x = DictBackend(self.app, serializer=serializer)
+        e = x.prepare_exception(ValueError('foo'))
+        if not isinstance(e, BaseException):
+            # not using pickle
+            assert 'exc_type' in e
+        e = x.exception_to_python(e)
+        assert e.__class__ is ValueError
+        assert e.args == ("foo", )
+
+    @pytest.mark.parametrize(("serializer"), (("pickle", "json")))
+    def test_prepare_custom_exception(self, serializer):
+        x = DictBackend(self.app, serializer=serializer)
+        e = x.prepare_exception(CustomTestError('foo'))
+        if not isinstance(e, BaseException):
+            assert 'exc_type' in e
         e = x.exception_to_python(e)
-        assert e.__class__.__name__ == 'KeyError'
-        assert str(e).strip('u') == "'foo'"
+        assert e.__class__ is CustomTestError
+        assert e.args == ("foo", )
 
     def test_save_group(self):
         b = BaseBackend(self.app)