Chain with one task should run as expected (celery#4730)

* bugfix: Chain with one task
* Fixes celery#4498

* add unit_test

celery/canvas.py

@@ -779,10 +779,9 @@ class chain(_chain):
     def __new__(cls, *tasks, **kwargs):
         # This forces `chain(X, Y, Z)` to work the same way as `X | Y | Z`
         if not kwargs and tasks:
-            if len(tasks) == 1 and is_list(tasks[0]):
-                # ensure chain(generator_expression) works.
-                tasks = tasks[0]
-            return reduce(operator.or_, tasks)
+            if len(tasks) != 1 or is_list(tasks[0]):
+                tasks = tasks[0] if len(tasks) == 1 else tasks
+                return reduce(operator.or_, tasks)
         return super(chain, cls).__new__(cls, *tasks, **kwargs)
 
 

t/integration/test_canvas.py

@@ -24,6 +24,11 @@ def test_simple_chain(self, manager):
         c = add.s(4, 4) | add.s(8) | add.s(16)
         assert c().get(timeout=TIMEOUT) == 32
 
+    @flaky
+    def test_single_chain(self, manager):
+        c = chain(add.s(3, 4))()
+        assert c.get(timeout=TIMEOUT) == 7
+
     @flaky
     def test_complex_chain(self, manager):
         c = (

t/unit/tasks/test_canvas.py

@@ -441,6 +441,11 @@ def test_apply(self):
         assert res.parent.parent.get() == 8
         assert res.parent.parent.parent is None
 
+    def test_single_expresion(self):
+        x = chain(self.add.s(1, 2)).apply()
+        assert x.get() == 3
+        assert x.parent is None
+
     def test_empty_chain_returns_none(self):
         assert chain(app=self.app)() is None
         assert chain(app=self.app).apply_async() is None