100% Coverage for celery.execute.trace

celery/execute/trace.py

@@ -9,6 +9,7 @@ from celery.datastructures import ExceptionInfo
 
 
 class TraceInfo(object):
+
     def __init__(self, status=states.PENDING, retval=None, exc_info=None):
         self.status = status
         self.retval = retval
@@ -39,7 +40,7 @@ class TraceInfo(object):
             if propagate:
                 raise
             return cls(states.FAILURE, retval=exc, exc_info=sys.exc_info())
-        except:
+        except:  # pragma: no cover
             # For Python2.4 where raising strings are still allowed.
             if propagate:
                 raise

celery/tests/test_execute_trace.py

@@ -0,0 +1,39 @@
+import operator
+
+from celery import states
+from celery.exceptions import RetryTaskError
+from celery.execute.trace import TraceInfo
+from celery.tests.utils import unittest
+
+trace = TraceInfo.trace
+
+
+def raises(exc):
+    raise exc
+
+
+class test_TraceInfo(unittest.TestCase):
+
+    def test_trace_successful(self):
+        info = trace(operator.add, (2, 2), {})
+        self.assertEqual(info.status, states.SUCCESS)
+        self.assertEqual(info.retval, 4)
+
+    def test_trace_SystemExit(self):
+        self.assertRaises(SystemExit, trace, raises, (SystemExit(), ), {})
+
+    def test_trace_RetryTaskError(self):
+        exc = RetryTaskError("foo", "bar")
+        info = trace(raises, (exc, ), {})
+        self.assertEqual(info.status, states.RETRY)
+        self.assertIs(info.retval, exc)
+
+    def test_trace_exception(self):
+        exc = KeyError("foo")
+        info = trace(raises, (exc, ), {})
+        self.assertEqual(info.status, states.FAILURE)
+        self.assertIs(info.retval, exc)
+
+    def test_trace_exception_propagate(self):
+        self.assertRaises(KeyError, trace, raises, (KeyError("foo"), ), {},
+                          propagate=True)

setup.cfg

@@ -11,7 +11,7 @@ cover3-exclude = celery
                  celery.bin.celeryd_detach
                  celery.bin.celeryctl
                  celery.bin.camqadm
-                 celery.task
+                 celery.execute
                  celery.platforms
                  celery.utils.patch
                  celery.utils.compat