Refactors tests to limit use of current_app

celery/app/base.py

@@ -201,7 +201,7 @@ class Celery(object):
                 if opts.get('_force_evaluate'):
                     ret = self._task_from_fun(fun, **opts)
                 else:
-                    # return a proxy object that is only evaluated when first used
+                    # return a proxy object that only evaluates on first use
                     ret = PromiseProxy(self._task_from_fun, (fun, ), opts)
                     self._pending.append(ret)
                 if _filt:

celery/tests/app/test_amqp.py

@@ -58,19 +58,15 @@ class test_TaskConsumer(AppCase):
 
     def test_accept_content(self):
         with self.app.pool.acquire(block=True) as conn:
-            prev = self.app.conf.CELERY_ACCEPT_CONTENT
             self.app.conf.CELERY_ACCEPT_CONTENT = ['application/json']
-            try:
-                self.assertEqual(
-                    self.app.amqp.TaskConsumer(conn).accept,
-                    set(['application/json'])
-                )
-                self.assertEqual(
-                    self.app.amqp.TaskConsumer(conn, accept=['json']).accept,
-                    set(['application/json']),
-                )
-            finally:
-                self.app.conf.CELERY_ACCEPT_CONTENT = prev
+            self.assertEqual(
+                self.app.amqp.TaskConsumer(conn).accept,
+                set(['application/json'])
+            )
+            self.assertEqual(
+                self.app.amqp.TaskConsumer(conn, accept=['json']).accept,
+                set(['application/json']),
+            )
 
 
 class test_compat_TaskPublisher(AppCase):
@@ -92,63 +88,51 @@ class test_compat_TaskPublisher(AppCase):
 class test_PublisherPool(AppCase):
 
     def test_setup_nolimit(self):
-        L = self.app.conf.BROKER_POOL_LIMIT
         self.app.conf.BROKER_POOL_LIMIT = None
         try:
             delattr(self.app, '_pool')
         except AttributeError:
             pass
         self.app.amqp._producer_pool = None
-        try:
-            pool = self.app.amqp.producer_pool
-            self.assertEqual(pool.limit, self.app.pool.limit)
-            self.assertFalse(pool._resource.queue)
-
-            r1 = pool.acquire()
-            r2 = pool.acquire()
-            r1.release()
-            r2.release()
-            r1 = pool.acquire()
-            r2 = pool.acquire()
-        finally:
-            self.app.conf.BROKER_POOL_LIMIT = L
+        pool = self.app.amqp.producer_pool
+        self.assertEqual(pool.limit, self.app.pool.limit)
+        self.assertFalse(pool._resource.queue)
+
+        r1 = pool.acquire()
+        r2 = pool.acquire()
+        r1.release()
+        r2.release()
+        r1 = pool.acquire()
+        r2 = pool.acquire()
 
     def test_setup(self):
-        L = self.app.conf.BROKER_POOL_LIMIT
         self.app.conf.BROKER_POOL_LIMIT = 2
         try:
             delattr(self.app, '_pool')
         except AttributeError:
             pass
         self.app.amqp._producer_pool = None
-        try:
-            pool = self.app.amqp.producer_pool
-            self.assertEqual(pool.limit, self.app.pool.limit)
-            self.assertTrue(pool._resource.queue)
-
-            p1 = r1 = pool.acquire()
-            p2 = r2 = pool.acquire()
-            r1.release()
-            r2.release()
-            r1 = pool.acquire()
-            r2 = pool.acquire()
-            self.assertIs(p2, r1)
-            self.assertIs(p1, r2)
-            r1.release()
-            r2.release()
-        finally:
-            self.app.conf.BROKER_POOL_LIMIT = L
+        pool = self.app.amqp.producer_pool
+        self.assertEqual(pool.limit, self.app.pool.limit)
+        self.assertTrue(pool._resource.queue)
+
+        p1 = r1 = pool.acquire()
+        p2 = r2 = pool.acquire()
+        r1.release()
+        r2.release()
+        r1 = pool.acquire()
+        r2 = pool.acquire()
+        self.assertIs(p2, r1)
+        self.assertIs(p1, r2)
+        r1.release()
+        r2.release()
 
 
 class test_Queues(AppCase):
 
     def test_queues_format(self):
-        prev, self.app.amqp.queues._consume_from = (
-            self.app.amqp.queues._consume_from, {})
-        try:
-            self.assertEqual(self.app.amqp.queues.format(), '')
-        finally:
-            self.app.amqp.queues._consume_from = prev
+        self.app.amqp.queues._consume_from = {}
+        self.assertEqual(self.app.amqp.queues.format(), '')
 
     def test_with_defaults(self):
         self.assertEqual(Queues(None), {})

celery/tests/app/test_annotations.py

@@ -1,32 +1,35 @@
 from __future__ import absolute_import
 
 from celery.app.annotations import MapAnnotation, prepare
-from celery.task import task
 from celery.utils.imports import qualname
 
-from celery.tests.case import Case
+from celery.tests.case import AppCase
 
 
-@task()
-def add(x, y):
-    return x + y
+class MyAnnotation(object):
+    foo = 65
 
 
-@task()
-def mul(x, y):
-    return x * y
+class AnnotationCase(AppCase):
 
+    def setup(self):
+        @self.app.task()
+        def add(x, y):
+            return x + y
+        self.add = add
 
-class MyAnnotation(object):
-    foo = 65
+        @self.app.task()
+        def mul(x, y):
+            return x * y
+        self.mul = mul
 
 
-class test_MapAnnotation(Case):
+class test_MapAnnotation(AnnotationCase):
 
     def test_annotate(self):
-        x = MapAnnotation({add.name: {'foo': 1}})
-        self.assertDictEqual(x.annotate(add), {'foo': 1})
-        self.assertIsNone(x.annotate(mul))
+        x = MapAnnotation({self.add.name: {'foo': 1}})
+        self.assertDictEqual(x.annotate(self.add), {'foo': 1})
+        self.assertIsNone(x.annotate(self.mul))
 
     def test_annotate_any(self):
         x = MapAnnotation({'*': {'foo': 2}})
@@ -36,10 +39,10 @@ class test_MapAnnotation(Case):
         self.assertIsNone(x.annotate_any())
 
 
-class test_prepare(Case):
+class test_prepare(AnnotationCase):
 
     def test_dict_to_MapAnnotation(self):
-        x = prepare({add.name: {'foo': 3}})
+        x = prepare({self.add.name: {'foo': 3}})
         self.assertIsInstance(x[0], MapAnnotation)
 
     def test_returns_list(self):

celery/tests/app/test_app.py

@@ -20,7 +20,7 @@ from celery.utils.serialization import pickle
 
 from celery.tests import config
 from celery.tests.case import (
-    Case,
+    AppCase, Case,
     mask_modules,
     platform_pyimp,
     sys_platform,
@@ -64,10 +64,9 @@ class test_module(Case):
         self.assertTrue(_app.bugreport())
 
 
-class test_App(Case):
+class test_App(AppCase):
 
-    def setUp(self):
-        self.app = Celery(set_as_current=False)
+    def setup(self):
         self.app.conf.update(test_config)
 
     def test_task(self):

celery/tests/app/test_beat.py

@@ -150,19 +150,16 @@ class test_Scheduler(AppCase):
         self.assertIs(scheduler.data, custom)
 
     def test_apply_async_uses_registered_task_instances(self):
-        through_task = [False]
 
-        class MockTask(self.app.Task):
-
-            @classmethod
-            def apply_async(cls, *args, **kwargs):
-                through_task[0] = True
-
-        assert MockTask.name in MockTask._get_app().tasks
+        @self.app.task
+        def foo():
+            pass
+        foo.apply_async = Mock(name='foo.apply_async')
+        assert foo.name in foo._get_app().tasks
 
         scheduler = mScheduler(app=self.app)
-        scheduler.apply_async(scheduler.Entry(task=MockTask.name))
-        self.assertTrue(through_task[0])
+        scheduler.apply_async(scheduler.Entry(task=foo.name))
+        self.assertTrue(foo.apply_async.called)
 
     def test_apply_async_should_not_sync(self):
 

celery/tests/app/test_builtins.py

@@ -2,7 +2,7 @@ from __future__ import absolute_import
 
 from mock import Mock, patch
 
-from celery import group, shared_task, chord
+from celery import group, chord
 from celery.app import builtins
 from celery.canvas import Signature
 from celery.five import range
@@ -10,31 +10,31 @@ from celery._state import _task_stack
 from celery.tests.case import AppCase
 
 
-@shared_task()
-def add(x, y):
-    return x + y
+class BuiltinsCase(AppCase):
 
+    def setup(self):
+        @self.app.task(shared=False)
+        def xsum(x):
+            return sum(x)
+        self.xsum = xsum
 
-@shared_task()
-def xsum(x):
-    return sum(x)
+        @self.app.task(shared=False)
+        def add(x, y):
+            return x + y
+        self.add = add
 
 
-class test_backend_cleanup(AppCase):
+class test_backend_cleanup(BuiltinsCase):
 
     def test_run(self):
-        prev = self.app.backend
         self.app.backend.cleanup = Mock()
         self.app.backend.cleanup.__name__ = 'cleanup'
-        try:
-            cleanup_task = builtins.add_backend_cleanup_task(self.app)
-            cleanup_task()
-            self.assertTrue(self.app.backend.cleanup.called)
-        finally:
-            self.app.backend = prev
+        cleanup_task = builtins.add_backend_cleanup_task(self.app)
+        cleanup_task()
+        self.assertTrue(self.app.backend.cleanup.called)
 
 
-class test_map(AppCase):
+class test_map(BuiltinsCase):
 
     def test_run(self):
 
@@ -48,7 +48,7 @@ class test_map(AppCase):
         self.assertEqual(res, [4, 16, 64])
 
 
-class test_starmap(AppCase):
+class test_starmap(BuiltinsCase):
 
     def test_run(self):
 
@@ -62,7 +62,7 @@ class test_starmap(AppCase):
         self.assertEqual(res, [4, 16, 64])
 
 
-class test_chunks(AppCase):
+class test_chunks(BuiltinsCase):
 
     @patch('celery.canvas.chunks.apply_chunks')
     def test_run(self, apply_chunks):
@@ -77,32 +77,26 @@ class test_chunks(AppCase):
         self.assertTrue(apply_chunks.called)
 
 
-class test_group(AppCase):
+class test_group(BuiltinsCase):
 
     def setup(self):
-        self.prev = self.app.tasks.get('celery.group')
         self.task = builtins.add_group_task(self.app)()
-
-    def teardown(self):
-        self.app.tasks['celery.group'] = self.prev
+        super(test_group, self).setup()
 
     def test_apply_async_eager(self):
         self.task.apply = Mock()
         self.app.conf.CELERY_ALWAYS_EAGER = True
-        try:
-            self.task.apply_async()
-        finally:
-            self.app.conf.CELERY_ALWAYS_EAGER = False
+        self.task.apply_async()
         self.assertTrue(self.task.apply.called)
 
     def test_apply(self):
-        x = group([add.s(4, 4), add.s(8, 8)])
+        x = group([self.add.s(4, 4), self.add.s(8, 8)])
         x.name = self.task.name
         res = x.apply()
         self.assertEqual(res.get(), [8, 16])
 
     def test_apply_async(self):
-        x = group([add.s(4, 4), add.s(8, 8)])
+        x = group([self.add.s(4, 4), self.add.s(8, 8)])
         x.apply_async()
 
     def test_apply_empty(self):
@@ -113,33 +107,30 @@ class test_group(AppCase):
         self.assertFalse(res.results)
 
     def test_apply_async_with_parent(self):
-        _task_stack.push(add)
+        _task_stack.push(self.add)
         try:
-            add.push_request(called_directly=False)
+            self.add.push_request(called_directly=False)
             try:
-                assert not add.request.children
-                x = group([add.s(4, 4), add.s(8, 8)])
+                assert not self.add.request.children
+                x = group([self.add.s(4, 4), self.add.s(8, 8)])
                 res = x()
-                self.assertTrue(add.request.children)
-                self.assertIn(res, add.request.children)
-                self.assertEqual(len(add.request.children), 1)
+                self.assertTrue(self.add.request.children)
+                self.assertIn(res, self.add.request.children)
+                self.assertEqual(len(self.add.request.children), 1)
             finally:
-                add.pop_request()
+                self.add.pop_request()
         finally:
             _task_stack.pop()
 
 
-class test_chain(AppCase):
+class test_chain(BuiltinsCase):
 
     def setup(self):
-        self.prev = self.app.tasks.get('celery.chain')
+        BuiltinsCase.setup(self)
         self.task = builtins.add_chain_task(self.app)()
 
-    def teardown(self):
-        self.app.tasks['celery.chain'] = self.prev
-
     def test_apply_async(self):
-        c = add.s(2, 2) | add.s(4) | add.s(8)
+        c = self.add.s(2, 2) | self.add.s(4) | self.add.s(8)
         result = c.apply_async()
         self.assertTrue(result.parent)
         self.assertTrue(result.parent.parent)
@@ -147,17 +138,17 @@ class test_chain(AppCase):
 
     def test_group_to_chord(self):
         c = (
-            group(add.s(i, i) for i in range(5)) |
-            add.s(10) |
-            add.s(20) |
-            add.s(30)
+            group(self.add.s(i, i) for i in range(5)) |
+            self.add.s(10) |
+            self.add.s(20) |
+            self.add.s(30)
         )
         tasks, _ = c.type.prepare_steps((), c.tasks)
         self.assertIsInstance(tasks[0], chord)
         self.assertTrue(tasks[0].body.options['link'])
         self.assertTrue(tasks[0].body.options['link'][0].options['link'])
 
-        c2 = add.s(2, 2) | group(add.s(i, i) for i in range(10))
+        c2 = self.add.s(2, 2) | group(self.add.s(i, i) for i in range(10))
         tasks2, _ = c2.type.prepare_steps((), c2.tasks)
         self.assertIsInstance(tasks2[1], group)
 
@@ -169,7 +160,7 @@ class test_chain(AppCase):
                 return self
 
         def s(*args, **kwargs):
-            return static(add.name, args, kwargs)
+            return static(self.add, args, kwargs, type=self.add)
 
         c = s(2, 2) | s(4, 4) | s(8, 8)
         r1 = c.apply_async(task_id='some_id')
@@ -189,38 +180,32 @@ class test_chain(AppCase):
             self.assertListEqual(task.options['link_error'], [s('error')])
 
 
-class test_chord(AppCase):
+class test_chord(BuiltinsCase):
 
     def setup(self):
-        self.prev = self.app.tasks.get('celery.chord')
         self.task = builtins.add_chord_task(self.app)()
-
-    def teardown(self):
-        self.app.tasks['celery.chord'] = self.prev
+        super(test_chord, self).setup()
 
     def test_apply_async(self):
-        x = chord([add.s(i, i) for i in range(10)], body=xsum.s())
+        x = chord([self.add.s(i, i) for i in range(10)], body=self.xsum.s())
         r = x.apply_async()
         self.assertTrue(r)
         self.assertTrue(r.parent)
 
     def test_run_header_not_group(self):
-        self.task([add.s(i, i) for i in range(10)], xsum.s())
+        self.task([self.add.s(i, i) for i in range(10)], self.xsum.s())
 
     def test_forward_options(self):
-        body = xsum.s()
-        x = chord([add.s(i, i) for i in range(10)], body=body)
+        body = self.xsum.s()
+        x = chord([self.add.s(i, i) for i in range(10)], body=body)
         x.apply_async(group_id='some_group_id')
         self.assertEqual(body.options['group_id'], 'some_group_id')
-        x2 = chord([add.s(i, i) for i in range(10)], body=body)
+        x2 = chord([self.add.s(i, i) for i in range(10)], body=body)
         x2.apply_async(chord='some_chord_id')
         self.assertEqual(body.options['chord'], 'some_chord_id')
 
     def test_apply_eager(self):
         self.app.conf.CELERY_ALWAYS_EAGER = True
-        try:
-            x = chord([add.s(i, i) for i in range(10)], body=xsum.s())
-            r = x.apply_async()
-            self.assertEqual(r.get(), 90)
-        finally:
-            self.app.conf.CELERY_ALWAYS_EAGER = False
+        x = chord([self.add.s(i, i) for i in range(10)], body=self.xsum.s())
+        r = x.apply_async()
+        self.assertEqual(r.get(), 90)

celery/tests/app/test_control.py

@@ -7,16 +7,10 @@ from functools import wraps
 from kombu.pidbox import Mailbox
 
 from celery.app import control
-from celery.task import task
 from celery.utils import uuid
 from celery.tests.case import AppCase, Case
 
 
-@task()
-def mytask():
-    pass
-
-
 class MockMailbox(Mailbox):
     sent = []
 
@@ -165,6 +159,11 @@ class test_Broadcast(AppCase):
         self.control = Control(app=self.app)
         self.app.control = self.control
 
+        @self.app.task()
+        def mytask():
+            pass
+        self.mytask = mytask
+
     def tearDown(self):
         del(self.app.control)
 
@@ -191,12 +190,12 @@ class test_Broadcast(AppCase):
 
     @with_mock_broadcast
     def test_rate_limit(self):
-        self.control.rate_limit(mytask.name, '100/m')
+        self.control.rate_limit(self.mytask.name, '100/m')
         self.assertIn('rate_limit', MockMailbox.sent)
 
     @with_mock_broadcast
     def test_time_limit(self):
-        self.control.time_limit(mytask.name, soft=10, hard=20)
+        self.control.time_limit(self.mytask.name, soft=10, hard=20)
         self.assertIn('time_limit', MockMailbox.sent)
 
     @with_mock_broadcast

celery/tests/app/test_loaders.py

@@ -29,17 +29,18 @@ class DummyLoader(base.BaseLoader):
 class test_loaders(AppCase):
 
     def test_get_loader_cls(self):
-
         self.assertEqual(loaders.get_loader_cls('default'),
                          default.Loader)
 
     def test_current_loader(self):
+        self.app.set_current()  # XXX Compat test
         with self.assertWarnsRegex(
                 CPendingDeprecationWarning,
                 r'deprecation'):
             self.assertIs(loaders.current_loader(), self.app.loader)
 
     def test_load_settings(self):
+        self.app.set_current()  # XXX Compat test
         with self.assertWarnsRegex(
                 CPendingDeprecationWarning,
                 r'deprecation'):

celery/tests/app/test_routes.py

@@ -7,7 +7,6 @@ from kombu.utils.functional import maybe_evaluate
 
 from celery.app import routes
 from celery.exceptions import QueueNotFound
-from celery.task import task
 from celery.utils.functional import LRUCache
 
 from celery.tests.case import AppCase
@@ -17,11 +16,6 @@ def Router(app, *args, **kwargs):
     return routes.Router(*args, app=app, **kwargs)
 
 
-@task()
-def mytask():
-    pass
-
-
 def E(app, queues):
     def expand(answer):
         return Router(app, [], queues).expand_destination(answer)
@@ -60,15 +54,20 @@ class RouteCase(AppCase):
             'routing_key': self.app.conf.CELERY_DEFAULT_ROUTING_KEY,
         }
 
+        @self.app.task()
+        def mytask():
+            pass
+        self.mytask = mytask
+
 
 class test_MapRoute(RouteCase):
 
     def test_route_for_task_expanded_route(self):
         with _queues(self.app, foo=self.a_queue, bar=self.b_queue):
             expand = E(self.app, self.app.amqp.queues)
-            route = routes.MapRoute({mytask.name: {'queue': 'foo'}})
+            route = routes.MapRoute({self.mytask.name: {'queue': 'foo'}})
             self.assertEqual(
-                expand(route.route_for_task(mytask.name))['queue'].name,
+                expand(route.route_for_task(self.mytask.name))['queue'].name,
                 'foo',
             )
             self.assertIsNone(route.route_for_task('celery.awesome'))
@@ -76,10 +75,10 @@ class test_MapRoute(RouteCase):
     def test_route_for_task(self):
         with _queues(self.app, foo=self.a_queue, bar=self.b_queue):
             expand = E(self.app, self.app.amqp.queues)
-            route = routes.MapRoute({mytask.name: self.b_queue})
+            route = routes.MapRoute({self.mytask.name: self.b_queue})
             self.assertDictContainsSubset(
                 self.b_queue,
-                expand(route.route_for_task(mytask.name)),
+                expand(route.route_for_task(self.mytask.name)),
             )
             self.assertIsNone(route.route_for_task('celery.awesome'))
 
@@ -99,10 +98,10 @@ class test_lookup_route(RouteCase):
 
     def test_lookup_takes_first(self):
         with _queues(self.app, foo=self.a_queue, bar=self.b_queue):
-            R = routes.prepare(({mytask.name: {'queue': 'bar'}},
-                                {mytask.name: {'queue': 'foo'}}))
+            R = routes.prepare(({self.mytask.name: {'queue': 'bar'}},
+                                {self.mytask.name: {'queue': 'foo'}}))
             router = Router(self.app, R, self.app.amqp.queues)
-            self.assertEqual(router.route({}, mytask.name,
+            self.assertEqual(router.route({}, self.mytask.name,
                              args=[1, 2], kwargs={})['queue'].name, 'bar')
 
     def test_expands_queue_in_options(self):
@@ -118,7 +117,7 @@ class test_lookup_route(RouteCase):
                  'exchange': None,
                  'routing_key': None,
                  'immediate': False},
-                mytask.name,
+                self.mytask.name,
                 args=[1, 2], kwargs={},
             )
             self.assertEqual(route['queue'].name, 'testq')
@@ -137,10 +136,10 @@ class test_lookup_route(RouteCase):
                 self.app.conf.CELERY_DEFAULT_QUEUE: self.d_queue}):
             R = routes.prepare((
                 {'celery.xaza': {'queue': 'bar'}},
-                {mytask.name: {'queue': 'foo'}}
+                {self.mytask.name: {'queue': 'foo'}}
             ))
             router = Router(self.app, R, self.app.amqp.queues)
-            self.assertEqual(router.route({}, mytask.name,
+            self.assertEqual(router.route({}, self.mytask.name,
                              args=[1, 2], kwargs={})['queue'].name, 'foo')
             self.assertEqual(
                 router.route({}, 'celery.poza')['queue'].name,

celery/tests/app/test_utils.py

@@ -2,7 +2,6 @@ from __future__ import absolute_import
 
 from collections import Mapping, MutableMapping
 
-from celery import Celery
 from celery.app.utils import Settings, bugreport
 
 from celery.tests.case import AppCase, Case, Mock
@@ -24,9 +23,8 @@ class TestSettings(Case):
 class test_bugreport(AppCase):
 
     def test_no_conn_driver_info(self):
-        app = Celery(set_as_current=False)
-        app.connection = Mock()
-        conn = app.connection.return_value = Mock()
+        self.app.connection = Mock()
+        conn = self.app.connection.return_value = Mock()
         conn.transport = None
 
-        bugreport(app)
+        bugreport(self.app)

celery/tests/backends/test_backends.py

@@ -33,6 +33,7 @@ class test_backends(AppCase):
             backends.get_backend_cls('fasodaopjeqijwqe')
 
     def test_default_backend(self):
+        self.app.set_current()  # XXX compat test
         self.assertEqual(backends.default_backend, self.app.backend)
 
     def test_backend_by_url(self, url='redis://localhost/1'):

celery/tests/backends/test_cache.py

@@ -70,34 +70,30 @@ class test_CacheBackend(AppCase):
         gid, res = uuid(), [AsyncResult(uuid()) for _ in range(3)]
         tb.on_chord_apply(gid, {}, result=res)
 
-    @patch('celery.result.GroupResult')
-    def test_on_chord_part_return(self, setresult):
+    @patch('celery.result.GroupResult.restore')
+    def test_on_chord_part_return(self, restore):
         tb = CacheBackend(backend='memory://', app=self.app)
 
         deps = Mock()
         deps.__len__ = Mock()
         deps.__len__.return_value = 2
-        setresult.restore.return_value = deps
+        restore.return_value = deps
         task = Mock()
         task.name = 'foobarbaz'
-        try:
-            self.app.tasks['foobarbaz'] = task
-            task.request.chord = subtask(task)
-
-            gid, res = uuid(), [AsyncResult(uuid()) for _ in range(3)]
-            task.request.group = gid
-            tb.on_chord_apply(gid, {}, result=res)
+        self.app.tasks['foobarbaz'] = task
+        task.request.chord = subtask(task)
 
-            self.assertFalse(deps.join_native.called)
-            tb.on_chord_part_return(task)
-            self.assertFalse(deps.join_native.called)
+        gid, res = uuid(), [AsyncResult(uuid()) for _ in range(3)]
+        task.request.group = gid
+        tb.on_chord_apply(gid, {}, result=res)
 
-            tb.on_chord_part_return(task)
-            deps.join_native.assert_called_with(propagate=True)
-            deps.delete.assert_called_with()
+        self.assertFalse(deps.join_native.called)
+        tb.on_chord_part_return(task)
+        self.assertFalse(deps.join_native.called)
 
-        finally:
-            self.app.tasks.pop('foobarbaz')
+        tb.on_chord_part_return(task)
+        deps.join_native.assert_called_with(propagate=True)
+        deps.delete.assert_called_with()
 
     def test_mget(self):
         self.tb.set('foo', 1)

celery/tests/backends/test_redis.py

@@ -153,32 +153,29 @@ class test_RedisBackend(AppCase):
         b.expires = None
         b.set('foo', 'bar')
 
-    @patch('celery.result.GroupResult')
-    def test_on_chord_part_return(self, setresult):
+    @patch('celery.result.GroupResult.restore')
+    def test_on_chord_part_return(self, restore):
         b = self.MockBackend(app=self.app)
         deps = Mock()
         deps.__len__ = Mock()
         deps.__len__.return_value = 10
-        setresult.restore.return_value = deps
+        restore.return_value = deps
         b.client.incr.return_value = 1
         task = Mock()
         task.name = 'foobarbaz'
-        try:
-            self.app.tasks['foobarbaz'] = task
-            task.request.chord = subtask(task)
-            task.request.group = 'group_id'
+        self.app.tasks['foobarbaz'] = task
+        task.request.chord = subtask(task)
+        task.request.group = 'group_id'
 
-            b.on_chord_part_return(task)
-            self.assertTrue(b.client.incr.call_count)
+        b.on_chord_part_return(task)
+        self.assertTrue(b.client.incr.call_count)
 
-            b.client.incr.return_value = len(deps)
-            b.on_chord_part_return(task)
-            deps.join_native.assert_called_with(propagate=True)
-            deps.delete.assert_called_with()
+        b.client.incr.return_value = len(deps)
+        b.on_chord_part_return(task)
+        deps.join_native.assert_called_with(propagate=True)
+        deps.delete.assert_called_with()
 
-            self.assertTrue(b.client.expire.call_count)
-        finally:
-            self.app.tasks.pop('foobarbaz')
+        self.assertTrue(b.client.expire.call_count)
 
     def test_process_cleanup(self):
         self.Backend(app=self.app).process_cleanup()

celery/tests/case.py

@@ -205,11 +205,15 @@ class Case(unittest.TestCase):
 
 
 class AppCase(Case):
+    contained = True
 
     def setUp(self):
+        from celery import Celery
         from celery.app import current_app
         from celery.backends.cache import CacheBackend, DummyClient
-        app = self.app = self._current_app = current_app()
+        self._current_app = current_app()
+        app = self.app = (Celery(set_as_current=False)
+                          if self.contained else self._current_app)
         if isinstance(app.backend, CacheBackend):
             if isinstance(app.backend.client, DummyClient):
                 app.backend.client.cache.clear()

celery/tests/events/test_snapshot.py

@@ -112,24 +112,21 @@ class test_evcam(AppCase):
             return test_evcam.MockReceiver()
 
     def setup(self):
-        self.prev, self.app.events = self.app.events, self.MockEvents()
+        self.app.events = self.MockEvents()
         self.app.events.app = self.app
 
-    def teardown(self):
-        self.app.events = self.prev
-
     def test_evcam(self):
         with restore_logging():
-            evcam(Polaroid, timer=timer)
-            evcam(Polaroid, timer=timer, loglevel='CRITICAL')
+            evcam(Polaroid, timer=timer, app=self.app)
+            evcam(Polaroid, timer=timer, loglevel='CRITICAL', app=self.app)
             self.MockReceiver.raise_keyboard_interrupt = True
             try:
                 with self.assertRaises(SystemExit):
-                    evcam(Polaroid, timer=timer)
+                    evcam(Polaroid, timer=timer, app=self.app)
             finally:
                 self.MockReceiver.raise_keyboard_interrupt = False
 
     @patch('celery.platforms.create_pidlock')
     def test_evcam_pidfile(self, create_pidlock):
-        evcam(Polaroid, timer=timer, pidfile='/var/pid')
+        evcam(Polaroid, timer=timer, pidfile='/var/pid', app=self.app)
         create_pidlock.assert_called_with('/var/pid')

celery/tests/tasks/test_canvas.py

@@ -2,7 +2,6 @@ from __future__ import absolute_import
 
 from mock import Mock
 
-from celery import shared_task
 from celery.canvas import (
     Signature,
     chain,
@@ -26,22 +25,27 @@ SIG = Signature({'task': 'TASK',
                  'subtask_type': ''})
 
 
-@shared_task()
-def add(x, y):
-    return x + y
+class CanvasCase(AppCase):
 
+    def setup(self):
 
-@shared_task()
-def mul(x, y):
-    return x * y
+        @self.app.task(shared=False)
+        def add(x, y):
+            return x + y
+        self.add = add
 
+        @self.app.task(shared=False)
+        def mul(x, y):
+            return x * y
+        self.mul = mul
 
-@shared_task()
-def div(x, y):
-    return x / y
+        @self.app.task(shared=False)
+        def div(x, y):
+            return x / y
+        self.div = div
 
 
-class test_Signature(AppCase):
+class test_Signature(CanvasCase):
 
     def test_getitem_property_class(self):
         self.assertTrue(Signature.task)
@@ -90,27 +94,27 @@ class test_Signature(AppCase):
         self.assertEqual(len(x.options['link_error']), 1)
 
     def test_flatten_links(self):
-        tasks = [add.s(2, 2), mul.s(4), div.s(2)]
+        tasks = [self.add.s(2, 2), self.mul.s(4), self.div.s(2)]
         tasks[0].link(tasks[1])
         tasks[1].link(tasks[2])
         self.assertEqual(tasks[0].flatten_links(), tasks)
 
     def test_OR(self):
-        x = add.s(2, 2) | mul.s(4)
+        x = self.add.s(2, 2) | self.mul.s(4)
         self.assertIsInstance(x, chain)
-        y = add.s(4, 4) | div.s(2)
+        y = self.add.s(4, 4) | self.div.s(2)
         z = x | y
         self.assertIsInstance(y, chain)
         self.assertIsInstance(z, chain)
         self.assertEqual(len(z.tasks), 4)
         with self.assertRaises(TypeError):
             x | 10
-        ax = add.s(2, 2) | (add.s(4) | add.s(8))
+        ax = self.add.s(2, 2) | (self.add.s(4) | self.add.s(8))
         self.assertIsInstance(ax, chain)
         self.assertEqual(len(ax.tasks), 3, 'consolidates chain to chain')
 
     def test_INVERT(self):
-        x = add.s(2, 2)
+        x = self.add.s(2, 2)
         x.apply_async = Mock()
         x.apply_async.return_value = Mock()
         x.apply_async.return_value.get = Mock()
@@ -119,14 +123,14 @@ class test_Signature(AppCase):
         self.assertTrue(x.apply_async.called)
 
     def test_merge_immutable(self):
-        x = add.si(2, 2, foo=1)
+        x = self.add.si(2, 2, foo=1)
         args, kwargs, options = x._merge((4, ), {'bar': 2}, {'task_id': 3})
         self.assertTupleEqual(args, (2, 2))
         self.assertDictEqual(kwargs, {'foo': 1})
         self.assertDictEqual(options, {'task_id': 3})
 
     def test_set_immutable(self):
-        x = add.s(2, 2)
+        x = self.add.s(2, 2)
         self.assertFalse(x.immutable)
         x.set(immutable=True)
         self.assertTrue(x.immutable)
@@ -134,7 +138,7 @@ class test_Signature(AppCase):
         self.assertFalse(x.immutable)
 
     def test_election(self):
-        x = add.s(2, 2)
+        x = self.add.s(2, 2)
         x.freeze('foo')
         prev, x.type.app.control = x.type.app.control, Mock()
         try:
@@ -153,27 +157,27 @@ class test_Signature(AppCase):
         self.assertTrue(s._apply_async)
 
 
-class test_xmap_xstarmap(AppCase):
+class test_xmap_xstarmap(CanvasCase):
 
     def test_apply(self):
         for type, attr in [(xmap, 'map'), (xstarmap, 'starmap')]:
             args = [(i, i) for i in range(10)]
-            s = getattr(add, attr)(args)
+            s = getattr(self.add, attr)(args)
             s.type = Mock()
 
             s.apply_async(foo=1)
             s.type.apply_async.assert_called_with(
-                (), {'task': add.s(), 'it': args}, foo=1,
+                (), {'task': self.add.s(), 'it': args}, foo=1,
             )
 
             self.assertEqual(type.from_dict(dict(s)), s)
             self.assertTrue(repr(s))
 
 
-class test_chunks(AppCase):
+class test_chunks(CanvasCase):
 
     def test_chunks(self):
-        x = add.chunks(range(100), 10)
+        x = self.add.chunks(range(100), 10)
         self.assertEqual(chunks.from_dict(dict(x)), x)
 
         self.assertTrue(x.group())
@@ -195,26 +199,28 @@ class test_chunks(AppCase):
             self.app.conf.CELERY_ALWAYS_EAGER = False
 
 
-class test_chain(AppCase):
+class test_chain(CanvasCase):
 
     def test_repr(self):
-        x = add.s(2, 2) | add.s(2)
-        self.assertEqual(repr(x), '%s(2, 2) | %s(2)' % (add.name, add.name))
+        x = self.add.s(2, 2) | self.add.s(2)
+        self.assertEqual(
+            repr(x), '%s(2, 2) | %s(2)' % (self.add.name, self.add.name),
+        )
 
     def test_reverse(self):
-        x = add.s(2, 2) | add.s(2)
+        x = self.add.s(2, 2) | self.add.s(2)
         self.assertIsInstance(subtask(x), chain)
         self.assertIsInstance(subtask(dict(x)), chain)
 
     def test_always_eager(self):
         self.app.conf.CELERY_ALWAYS_EAGER = True
         try:
-            self.assertEqual(~(add.s(4, 4) | add.s(8)), 16)
+            self.assertEqual(~(self.add.s(4, 4) | self.add.s(8)), 16)
         finally:
             self.app.conf.CELERY_ALWAYS_EAGER = False
 
     def test_apply(self):
-        x = chain(add.s(4, 4), add.s(8), add.s(10))
+        x = chain(self.add.s(4, 4), self.add.s(8), self.add.s(10))
         res = x.apply()
         self.assertIsInstance(res, EagerResult)
         self.assertEqual(res.get(), 26)
@@ -228,40 +234,42 @@ class test_chain(AppCase):
         self.assertFalse(x())
 
     def test_call_with_tasks(self):
-        x = add.s(2, 2) | add.s(4)
+        x = self.add.s(2, 2) | self.add.s(4)
         x.apply_async = Mock()
         x(2, 2, foo=1)
         x.apply_async.assert_called_with((2, 2), {'foo': 1})
 
     def test_from_dict_no_args__with_args(self):
-        x = dict(add.s(2, 2) | add.s(4))
+        x = dict(self.add.s(2, 2) | self.add.s(4))
         x['args'] = None
         self.assertIsInstance(chain.from_dict(x), chain)
         x['args'] = (2, )
         self.assertIsInstance(chain.from_dict(x), chain)
 
     def test_accepts_generator_argument(self):
-        x = chain(add.s(i) for i in range(10))
-        self.assertTrue(x.tasks[0].type, add)
+        x = chain(self.add.s(i) for i in range(10))
+        self.assertTrue(x.tasks[0].type, self.add)
         self.assertTrue(x.type)
 
 
-class test_group(AppCase):
+class test_group(CanvasCase):
 
     def test_repr(self):
-        x = group([add.s(2, 2), add.s(4, 4)])
+        x = group([self.add.s(2, 2), self.add.s(4, 4)])
         self.assertEqual(repr(x), repr(x.tasks))
 
     def test_reverse(self):
-        x = group([add.s(2, 2), add.s(4, 4)])
+        x = group([self.add.s(2, 2), self.add.s(4, 4)])
         self.assertIsInstance(subtask(x), group)
         self.assertIsInstance(subtask(dict(x)), group)
 
     def test_maybe_group_sig(self):
-        self.assertListEqual(_maybe_group(add.s(2, 2)), [add.s(2, 2)])
+        self.assertListEqual(
+            _maybe_group(self.add.s(2, 2)), [self.add.s(2, 2)],
+        )
 
     def test_from_dict(self):
-        x = group([add.s(2, 2), add.s(4, 4)])
+        x = group([self.add.s(2, 2), self.add.s(4, 4)])
         x['args'] = (2, 2)
         self.assertTrue(group.from_dict(dict(x)))
         x['args'] = None
@@ -272,25 +280,25 @@ class test_group(AppCase):
         self.assertIsNone(x())
 
     def test_skew(self):
-        g = group([add.s(i, i) for i in range(10)])
+        g = group([self.add.s(i, i) for i in range(10)])
         g.skew(start=1, stop=10, step=1)
         for i, task in enumerate(g.tasks):
             self.assertEqual(task.options['countdown'], i + 1)
 
     def test_iter(self):
-        g = group([add.s(i, i) for i in range(10)])
+        g = group([self.add.s(i, i) for i in range(10)])
         self.assertListEqual(list(iter(g)), g.tasks)
 
 
-class test_chord(AppCase):
+class test_chord(CanvasCase):
 
     def test_reverse(self):
-        x = chord([add.s(2, 2), add.s(4, 4)], body=mul.s(4))
+        x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
         self.assertIsInstance(subtask(x), chord)
         self.assertIsInstance(subtask(dict(x)), chord)
 
     def test_clone_clones_body(self):
-        x = chord([add.s(2, 2), add.s(4, 4)], body=mul.s(4))
+        x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
         y = x.clone()
         self.assertIsNot(x.kwargs['body'], y.kwargs['body'])
         y.kwargs.pop('body')
@@ -298,12 +306,12 @@ class test_chord(AppCase):
         self.assertIsNone(z.kwargs.get('body'))
 
     def test_links_to_body(self):
-        x = chord([add.s(2, 2), add.s(4, 4)], body=mul.s(4))
-        x.link(div.s(2))
+        x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
+        x.link(self.div.s(2))
         self.assertFalse(x.options.get('link'))
         self.assertTrue(x.kwargs['body'].options['link'])
 
-        x.link_error(div.s(2))
+        x.link_error(self.div.s(2))
         self.assertFalse(x.options.get('link_error'))
         self.assertTrue(x.kwargs['body'].options['link_error'])
 
@@ -311,20 +319,20 @@ class test_chord(AppCase):
         self.assertTrue(x.body)
 
     def test_repr(self):
-        x = chord([add.s(2, 2), add.s(4, 4)], body=mul.s(4))
+        x = chord([self.add.s(2, 2), self.add.s(4, 4)], body=self.mul.s(4))
         self.assertTrue(repr(x))
         x.kwargs['body'] = None
         self.assertIn('without body', repr(x))
 
 
-class test_maybe_subtask(AppCase):
+class test_maybe_subtask(CanvasCase):
 
     def test_is_None(self):
         self.assertIsNone(maybe_subtask(None))
 
     def test_is_dict(self):
-        self.assertIsInstance(maybe_subtask(dict(add.s())), Signature)
+        self.assertIsInstance(maybe_subtask(dict(self.add.s())), Signature)
 
     def test_when_sig(self):
-        s = add.s()
+        s = self.add.s()
         self.assertIs(maybe_subtask(s), s)

celery/tests/tasks/test_chord.py

@@ -1,6 +1,5 @@
 from __future__ import absolute_import
 
-from mock import patch
 from contextlib import contextmanager
 
 from celery import group
@@ -18,7 +17,7 @@ class ChordCase(AppCase):
 
     def setup(self):
 
-        @self.app.task
+        @self.app.task(shared=False)
         def add(x, y):
             return x + y
         self.add = add
@@ -64,8 +63,7 @@ def patch_unlock_retry(app):
 
 class test_unlock_chord_task(ChordCase):
 
-    @patch('celery.result.GroupResult')
-    def test_unlock_ready(self, GroupResult):
+    def test_unlock_ready(self):
 
         class AlwaysReady(TSR):
             is_ready = True
@@ -79,6 +77,7 @@ class test_unlock_chord_task(ChordCase):
             self.assertFalse(retry.call_count)
 
     def test_callback_fails(self):
+
         class AlwaysReady(TSR):
             is_ready = True
             value = [2, 4, 8, 6]
@@ -130,38 +129,37 @@ class test_unlock_chord_task(ChordCase):
 
     @contextmanager
     def _chord_context(self, ResultCls, setup=None, **kwargs):
-        with patch('celery.result.GroupResult'):
-
-            @self.app.task()
-            def callback(*args, **kwargs):
-                pass
-
-            pts, result.GroupResult = result.GroupResult, ResultCls
-            callback.apply_async = Mock()
-            callback_s = callback.s()
-            callback_s.id = 'callback_id'
-            fail_current = self.app.backend.fail_from_current_stack = Mock()
-            try:
-                with patch_unlock_retry(self.app) as (unlock, retry):
-                    subtask, canvas.maybe_subtask = (
-                        canvas.maybe_subtask, passthru,
+        @self.app.task(shared=False)
+        def callback(*args, **kwargs):
+            pass
+        self.app.finalize()
+
+        pts, result.GroupResult = result.GroupResult, ResultCls
+        callback.apply_async = Mock()
+        callback_s = callback.s()
+        callback_s.id = 'callback_id'
+        fail_current = self.app.backend.fail_from_current_stack = Mock()
+        try:
+            with patch_unlock_retry(self.app) as (unlock, retry):
+                subtask, canvas.maybe_subtask = (
+                    canvas.maybe_subtask, passthru,
+                )
+                if setup:
+                    setup(callback)
+                try:
+                    assert self.app.tasks['celery.chord_unlock'] is unlock
+                    unlock(
+                        'group_id', callback_s,
+                        result=[AsyncResult(r) for r in ['1', 2, 3]],
+                        GroupResult=ResultCls, **kwargs
                     )
-                    if setup:
-                        setup(callback)
-                    try:
-                        unlock(
-                            'group_id', callback_s,
-                            result=[AsyncResult(r) for r in ['1', 2, 3]],
-                            GroupResult=ResultCls, **kwargs
-                        )
-                    finally:
-                        canvas.maybe_subtask = subtask
-                    yield callback_s, retry, fail_current
-            finally:
-                result.GroupResult = pts
-
-    @patch('celery.result.GroupResult')
-    def test_when_not_ready(self, GroupResult):
+                finally:
+                    canvas.maybe_subtask = subtask
+                yield callback_s, retry, fail_current
+        finally:
+            result.GroupResult = pts
+
+    def test_when_not_ready(self):
         class NeverReady(TSR):
             is_ready = False
 

celery/tests/tasks/test_http.py

@@ -140,6 +140,7 @@ class test_HttpDispatch(AppCase):
 
 
 class test_URL(AppCase):
+    contained = False
 
     def test_URL_get_async(self):
         with eager_tasks(self.app):

celery/tests/tasks/test_tasks.py

@@ -9,12 +9,9 @@ from pickle import loads, dumps
 
 from kombu import Queue
 
+from celery import Task
+
 from celery.task import (
-    current,
-    task,
-    Task,
-    BaseTask,
-    TaskSet,
     periodic_task,
     PeriodicTask
 )
@@ -26,7 +23,7 @@ from celery.schedules import crontab, crontab_parser, ParseException
 from celery.utils import uuid
 from celery.utils.timeutils import parse_iso8601, timedelta_seconds
 
-from celery.tests.case import AppCase, eager_tasks, WhateverIO
+from celery.tests.case import AppCase
 
 
 def return_True(*args, **kwargs):
@@ -34,213 +31,217 @@ def return_True(*args, **kwargs):
     return True
 
 
-return_True_task = task()(return_True)
-
-
 def raise_exception(self, **kwargs):
     raise Exception('%s error' % self.__class__)
 
 
 class MockApplyTask(Task):
+    abstract = True
     applied = 0
 
     def run(self, x, y):
         return x * y
 
-    @classmethod
     def apply_async(self, *args, **kwargs):
         self.applied += 1
 
 
-@task(name='c.unittest.increment_counter_task', count=0)
-def increment_counter(increment_by=1):
-    increment_counter.count += increment_by or 1
-    return increment_counter.count
-
-
-@task(name='c.unittest.raising_task')
-def raising():
-    raise KeyError('foo')
+class TasksCase(AppCase):
 
+    def setup(self):
 
-@task(max_retries=3, iterations=0)
-def retry_task(arg1, arg2, kwarg=1, max_retries=None, care=True):
-    current.iterations += 1
-    rmax = current.max_retries if max_retries is None else max_retries
-
-    assert repr(current.request)
-    retries = current.request.retries
-    if care and retries >= rmax:
-        return arg1
-    else:
-        raise current.retry(countdown=0, max_retries=rmax)
+        self.return_True_task = self.app.task(shared=False)(return_True)
 
+        @self.app.task(bind=True, count=0, shared=False)
+        def increment_counter(self, increment_by=1):
+            self.count += increment_by or 1
+            return self.count
+        self.increment_counter = increment_counter
 
-@task(max_retries=3, iterations=0, accept_magic_kwargs=True)
-def retry_task_noargs(**kwargs):
-    current.iterations += 1
+        @self.app.task(shared=False)
+        def raising():
+            raise KeyError('foo')
+        self.raising = raising
 
-    retries = kwargs['task_retries']
-    if retries >= 3:
-        return 42
-    else:
-        raise current.retry(countdown=0)
+        @self.app.task(bind=True, max_retries=3, iterations=0, shared=False)
+        def retry_task(self, arg1, arg2, kwarg=1, max_retries=None, care=True):
+            self.iterations += 1
+            rmax = self.max_retries if max_retries is None else max_retries
 
+            assert repr(self.request)
+            retries = self.request.retries
+            if care and retries >= rmax:
+                return arg1
+            else:
+                raise self.retry(countdown=0, max_retries=rmax)
+        self.retry_task = retry_task
 
-@task(max_retries=3, iterations=0, base=MockApplyTask,
-      accept_magic_kwargs=True)
-def retry_task_mockapply(arg1, arg2, kwarg=1, **kwargs):
-    current.iterations += 1
+        @self.app.task(bind=True, max_retries=3, iterations=0, shared=False)
+        def retry_task_noargs(self, **kwargs):
+            self.iterations += 1
 
-    retries = kwargs['task_retries']
-    if retries >= 3:
-        return arg1
-    else:
-        kwargs.update(kwarg=kwarg)
-    raise current.retry(countdown=0)
+            if self.request.retries >= 3:
+                return 42
+            else:
+                raise self.retry(countdown=0)
+        self.retry_task_noargs = retry_task_noargs
+
+        @self.app.task(bind=True, max_retries=3, iterations=0,
+                       base=MockApplyTask, shared=False)
+        def retry_task_mockapply(self, arg1, arg2, kwarg=1):
+            self.iterations += 1
+
+            retries = self.request.retries
+            if retries >= 3:
+                return arg1
+            raise self.retry(countdown=0)
+        self.retry_task_mockapply = retry_task_mockapply
+
+        @self.app.task(bind=True, max_retries=3, iterations=0, shared=False)
+        def retry_task_customexc(self, arg1, arg2, kwarg=1, **kwargs):
+            self.iterations += 1
+
+            retries = self.request.retries
+            if retries >= 3:
+                return arg1 + kwarg
+            else:
+                try:
+                    raise MyCustomException('Elaine Marie Benes')
+                except MyCustomException as exc:
+                    kwargs.update(kwarg=kwarg)
+                    raise self.retry(countdown=0, exc=exc)
+        self.retry_task_customexc = retry_task_customexc
 
 
 class MyCustomException(Exception):
     """Random custom exception."""
 
 
-@task(max_retries=3, iterations=0, accept_magic_kwargs=True)
-def retry_task_customexc(arg1, arg2, kwarg=1, **kwargs):
-    current.iterations += 1
-
-    retries = kwargs['task_retries']
-    if retries >= 3:
-        return arg1 + kwarg
-    else:
-        try:
-            raise MyCustomException('Elaine Marie Benes')
-        except MyCustomException as exc:
-            kwargs.update(kwarg=kwarg)
-            raise current.retry(countdown=0, exc=exc)
-
-
-class test_task_retries(AppCase):
+class test_task_retries(TasksCase):
 
     def test_retry(self):
-        retry_task.__class__.max_retries = 3
-        retry_task.iterations = 0
-        retry_task.apply([0xFF, 0xFFFF])
-        self.assertEqual(retry_task.iterations, 4)
+        self.retry_task.max_retries = 3
+        self.retry_task.iterations = 0
+        self.retry_task.apply([0xFF, 0xFFFF])
+        self.assertEqual(self.retry_task.iterations, 4)
 
-        retry_task.__class__.max_retries = 3
-        retry_task.iterations = 0
-        retry_task.apply([0xFF, 0xFFFF], {'max_retries': 10})
-        self.assertEqual(retry_task.iterations, 11)
+        self.retry_task.max_retries = 3
+        self.retry_task.iterations = 0
+        self.retry_task.apply([0xFF, 0xFFFF], {'max_retries': 10})
+        self.assertEqual(self.retry_task.iterations, 11)
 
     def test_retry_no_args(self):
-        assert retry_task_noargs.accept_magic_kwargs
-        retry_task_noargs.__class__.max_retries = 3
-        retry_task_noargs.iterations = 0
-        retry_task_noargs.apply()
-        self.assertEqual(retry_task_noargs.iterations, 4)
+        self.retry_task_noargs.max_retries = 3
+        self.retry_task_noargs.iterations = 0
+        self.retry_task_noargs.apply(propagate=True).get()
+        self.assertEqual(self.retry_task_noargs.iterations, 4)
 
     def test_retry_kwargs_can_be_empty(self):
-        retry_task_mockapply.push_request()
+        self.retry_task_mockapply.push_request()
         try:
             with self.assertRaises(RetryTaskError):
-                retry_task_mockapply.retry(args=[4, 4], kwargs=None)
+                self.retry_task_mockapply.retry(args=[4, 4], kwargs=None)
         finally:
-            retry_task_mockapply.pop_request()
+            self.retry_task_mockapply.pop_request()
 
     def test_retry_not_eager(self):
-        retry_task_mockapply.push_request()
+        self.retry_task_mockapply.push_request()
         try:
-            retry_task_mockapply.request.called_directly = False
+            self.retry_task_mockapply.request.called_directly = False
             exc = Exception('baz')
             try:
-                retry_task_mockapply.retry(
+                self.retry_task_mockapply.retry(
                     args=[4, 4], kwargs={'task_retries': 0},
                     exc=exc, throw=False,
                 )
-                self.assertTrue(retry_task_mockapply.__class__.applied)
+                self.assertTrue(self.retry_task_mockapply.applied)
             finally:
-                retry_task_mockapply.__class__.applied = 0
+                self.retry_task_mockapply.applied = 0
 
             try:
                 with self.assertRaises(RetryTaskError):
-                    retry_task_mockapply.retry(
+                    self.retry_task_mockapply.retry(
                         args=[4, 4], kwargs={'task_retries': 0},
                         exc=exc, throw=True)
-                self.assertTrue(retry_task_mockapply.__class__.applied)
+                self.assertTrue(self.retry_task_mockapply.applied)
             finally:
-                retry_task_mockapply.__class__.applied = 0
+                self.retry_task_mockapply.applied = 0
         finally:
-            retry_task_mockapply.pop_request()
+            self.retry_task_mockapply.pop_request()
 
     def test_retry_with_kwargs(self):
-        retry_task_customexc.__class__.max_retries = 3
-        retry_task_customexc.iterations = 0
-        retry_task_customexc.apply([0xFF, 0xFFFF], {'kwarg': 0xF})
-        self.assertEqual(retry_task_customexc.iterations, 4)
+        self.retry_task_customexc.max_retries = 3
+        self.retry_task_customexc.iterations = 0
+        self.retry_task_customexc.apply([0xFF, 0xFFFF], {'kwarg': 0xF})
+        self.assertEqual(self.retry_task_customexc.iterations, 4)
 
     def test_retry_with_custom_exception(self):
-        retry_task_customexc.__class__.max_retries = 2
-        retry_task_customexc.iterations = 0
-        result = retry_task_customexc.apply([0xFF, 0xFFFF], {'kwarg': 0xF})
+        self.retry_task_customexc.max_retries = 2
+        self.retry_task_customexc.iterations = 0
+        result = self.retry_task_customexc.apply(
+            [0xFF, 0xFFFF], {'kwarg': 0xF},
+        )
         with self.assertRaises(MyCustomException):
             result.get()
-        self.assertEqual(retry_task_customexc.iterations, 3)
+        self.assertEqual(self.retry_task_customexc.iterations, 3)
 
     def test_max_retries_exceeded(self):
-        retry_task.__class__.max_retries = 2
-        retry_task.iterations = 0
-        result = retry_task.apply([0xFF, 0xFFFF], {'care': False})
-        with self.assertRaises(retry_task.MaxRetriesExceededError):
+        self.retry_task.max_retries = 2
+        self.retry_task.iterations = 0
+        result = self.retry_task.apply([0xFF, 0xFFFF], {'care': False})
+        with self.assertRaises(self.retry_task.MaxRetriesExceededError):
             result.get()
-        self.assertEqual(retry_task.iterations, 3)
+        self.assertEqual(self.retry_task.iterations, 3)
 
-        retry_task.__class__.max_retries = 1
-        retry_task.iterations = 0
-        result = retry_task.apply([0xFF, 0xFFFF], {'care': False})
-        with self.assertRaises(retry_task.MaxRetriesExceededError):
+        self.retry_task.max_retries = 1
+        self.retry_task.iterations = 0
+        result = self.retry_task.apply([0xFF, 0xFFFF], {'care': False})
+        with self.assertRaises(self.retry_task.MaxRetriesExceededError):
             result.get()
-        self.assertEqual(retry_task.iterations, 2)
+        self.assertEqual(self.retry_task.iterations, 2)
 
 
-class test_canvas_utils(AppCase):
+class test_canvas_utils(TasksCase):
 
     def test_si(self):
-        self.assertTrue(retry_task.si())
-        self.assertTrue(retry_task.si().immutable)
+        self.assertTrue(self.retry_task.si())
+        self.assertTrue(self.retry_task.si().immutable)
 
     def test_chunks(self):
-        self.assertTrue(retry_task.chunks(range(100), 10))
+        self.assertTrue(self.retry_task.chunks(range(100), 10))
 
     def test_map(self):
-        self.assertTrue(retry_task.map(range(100)))
+        self.assertTrue(self.retry_task.map(range(100)))
 
     def test_starmap(self):
-        self.assertTrue(retry_task.starmap(range(100)))
+        self.assertTrue(self.retry_task.starmap(range(100)))
 
     def test_on_success(self):
-        retry_task.on_success(1, 1, (), {})
+        self.retry_task.on_success(1, 1, (), {})
 
 
-class test_tasks(AppCase):
+class test_tasks(TasksCase):
 
     def now(self):
         return self.app.now()
 
     def test_unpickle_task(self):
+        self.app.set_current()
         import pickle
 
-        @task
+        @self.app.task(shared=True)
         def xxx():
             pass
         self.assertIs(pickle.loads(pickle.dumps(xxx)), xxx.app.tasks[xxx.name])
 
-    def createTask(self, name):
-        return task(__module__=self.__module__, name=name)(return_True)
+    def create_task(self, name):
+        return self.app.task(__module__=self.__module__,
+                             shared=False, name=name)(return_True)
 
     def test_AsyncResult(self):
         task_id = uuid()
-        result = retry_task.AsyncResult(task_id)
-        self.assertEqual(result.backend, retry_task.backend)
+        result = self.retry_task.AsyncResult(task_id)
+        self.assertEqual(result.backend, self.retry_task.backend)
         self.assertEqual(result.id, task_id)
 
     def assertNextTaskDataEqual(self, consumer, presult, task_name,
@@ -271,82 +272,80 @@ class test_tasks(AppCase):
 
     def test_task_kwargs_must_be_dictionary(self):
         with self.assertRaises(ValueError):
-            increment_counter.apply_async([], 'str')
+            self.increment_counter.apply_async([], 'str')
 
     def test_task_args_must_be_list(self):
         with self.assertRaises(ValueError):
-            increment_counter.apply_async('str', {})
+            self.increment_counter.apply_async('str', {})
 
     def test_regular_task(self):
-        T1 = self.createTask('c.unittest.t.t1')
-        self.assertIsInstance(T1, BaseTask)
+        T1 = self.create_task('c.unittest.t.t1')
+        self.assertIsInstance(T1, Task)
         self.assertTrue(T1.run())
         self.assertTrue(isinstance(T1, Callable), 'Task class is callable()')
         self.assertTrue(T1(), 'Task class runs run() when called')
 
-        consumer = T1.get_consumer()
-        with self.assertRaises(NotImplementedError):
-            consumer.receive('foo', 'foo')
-        consumer.purge()
-        self.assertIsNone(consumer.queues[0].get())
-        T1.get_consumer(queues=[Queue('foo')])
-
-        # Without arguments.
-        presult = T1.delay()
-        self.assertNextTaskDataEqual(consumer, presult, T1.name)
-
-        # With arguments.
-        presult2 = T1.apply_async(kwargs=dict(name='George Costanza'))
-        self.assertNextTaskDataEqual(
-            consumer, presult2, T1.name, name='George Costanza',
-        )
-
-        # send_task
-        sresult = send_task(T1.name, kwargs=dict(name='Elaine M. Benes'))
-        self.assertNextTaskDataEqual(
-            consumer, sresult, T1.name, name='Elaine M. Benes',
-        )
-
-        # With eta.
-        presult2 = T1.apply_async(
-            kwargs=dict(name='George Costanza'),
-            eta=self.now() + timedelta(days=1),
-            expires=self.now() + timedelta(days=2),
-        )
-        self.assertNextTaskDataEqual(
-            consumer, presult2, T1.name,
-            name='George Costanza', test_eta=True, test_expires=True,
-        )
-
-        # With countdown.
-        presult2 = T1.apply_async(kwargs=dict(name='George Costanza'),
-                                  countdown=10, expires=12)
-        self.assertNextTaskDataEqual(
-            consumer, presult2, T1.name,
-            name='George Costanza', test_eta=True, test_expires=True,
-        )
-
-        # Discarding all tasks.
-        consumer.purge()
-        T1.apply_async()
-        self.assertEqual(consumer.purge(), 1)
-        self.assertIsNone(consumer.queues[0].get())
-
-        self.assertFalse(presult.successful())
-        T1.backend.mark_as_done(presult.id, result=None)
-        self.assertTrue(presult.successful())
-
-        publisher = T1.get_publisher()
-        self.assertTrue(publisher.exchange)
+        with self.app.connection_or_acquire() as conn:
+            consumer = self.app.amqp.TaskConsumer(conn)
+            with self.assertRaises(NotImplementedError):
+                consumer.receive('foo', 'foo')
+            consumer.purge()
+            self.assertIsNone(consumer.queues[0].get())
+            self.app.amqp.TaskConsumer(conn, queues=[Queue('foo')])
+
+            # Without arguments.
+            presult = T1.delay()
+            self.assertNextTaskDataEqual(consumer, presult, T1.name)
+
+            # With arguments.
+            presult2 = T1.apply_async(kwargs=dict(name='George Costanza'))
+            self.assertNextTaskDataEqual(
+                consumer, presult2, T1.name, name='George Costanza',
+            )
+
+            # send_task
+            sresult = send_task(T1.name, kwargs=dict(name='Elaine M. Benes'))
+            self.assertNextTaskDataEqual(
+                consumer, sresult, T1.name, name='Elaine M. Benes',
+            )
+
+            # With eta.
+            presult2 = T1.apply_async(
+                kwargs=dict(name='George Costanza'),
+                eta=self.now() + timedelta(days=1),
+                expires=self.now() + timedelta(days=2),
+            )
+            self.assertNextTaskDataEqual(
+                consumer, presult2, T1.name,
+                name='George Costanza', test_eta=True, test_expires=True,
+            )
+
+            # With countdown.
+            presult2 = T1.apply_async(kwargs=dict(name='George Costanza'),
+                                      countdown=10, expires=12)
+            self.assertNextTaskDataEqual(
+                consumer, presult2, T1.name,
+                name='George Costanza', test_eta=True, test_expires=True,
+            )
+
+            # Discarding all tasks.
+            consumer.purge()
+            T1.apply_async()
+            self.assertEqual(consumer.purge(), 1)
+            self.assertIsNone(consumer.queues[0].get())
+
+            self.assertFalse(presult.successful())
+            T1.backend.mark_as_done(presult.id, result=None)
+            self.assertTrue(presult.successful())
 
     def test_repr_v2_compat(self):
-        task = type(self.createTask('c.unittest.v2c'))
+        task = type(self.create_task('c.unittest.v2c')._get_current_object())
         task.__v2_compat__ = True
         self.assertIn('v2 compatible', repr(task))
 
     def test_apply_with_self(self):
 
-        @task(__self__=42)
+        @self.app.task(__self__=42, shared=False)
         def tawself(self):
             return self
 
@@ -355,7 +354,7 @@ class test_tasks(AppCase):
         self.assertEqual(tawself(), 42)
 
     def test_context_get(self):
-        task = self.createTask('c.unittest.t.c.g')
+        task = self.create_task('c.unittest.t.c.g')
         task.push_request()
         try:
             request = task.request
@@ -367,7 +366,7 @@ class test_tasks(AppCase):
             task.pop_request()
 
     def test_task_class_repr(self):
-        task = self.createTask('c.unittest.t.repr')
+        task = self.create_task('c.unittest.t.repr')
         self.assertIn('class Task of', repr(task.app.Task))
         prev, task.app.Task._app = task.app.Task._app, None
         try:
@@ -376,8 +375,8 @@ class test_tasks(AppCase):
             task.app.Task._app = prev
 
     def test_bind_no_magic_kwargs(self):
-        task = self.createTask('c.unittest.t.magic_kwargs')
-        task.__class__.accept_magic_kwargs = None
+        task = self.create_task('c.unittest.t.magic_kwargs')
+        task.accept_magic_kwargs = None
         task.bind(task.app)
 
     def test_annotate(self):
@@ -387,40 +386,23 @@ class test_tasks(AppCase):
             self.assertEqual(Task.FOO, 'BAR')
 
     def test_after_return(self):
-        task = self.createTask('c.unittest.t.after_return')
+        task = self.create_task('c.unittest.t.after_return')
         task.push_request()
         try:
-            task.request.chord = return_True_task.s()
+            task.request.chord = self.return_True_task.s()
             task.after_return('SUCCESS', 1.0, 'foobar', (), {}, None)
             task.request.clear()
         finally:
             task.pop_request()
 
     def test_send_task_sent_event(self):
-        T1 = self.createTask('c.unittest.t.t1')
-        app = T1.app
-        with app.connection() as conn:
-            app.conf.CELERY_SEND_TASK_SENT_EVENT = True
-            del(app.amqp.__dict__['TaskProducer'])
-            try:
-                self.assertTrue(app.amqp.TaskProducer(conn).send_sent_event)
-            finally:
-                app.conf.CELERY_SEND_TASK_SENT_EVENT = False
-                del(app.amqp.__dict__['TaskProducer'])
-
-    def test_get_publisher(self):
-        connection = self.app.connection()
-        p = increment_counter.get_publisher(connection, auto_declare=False,
-                                            exchange='foo')
-        self.assertEqual(p.exchange.name, 'foo')
-        p = increment_counter.get_publisher(connection, auto_declare=False,
-                                            exchange='foo',
-                                            exchange_type='fanout')
-        self.assertEqual(p.exchange.type, 'fanout')
+        with self.app.connection() as conn:
+            self.app.conf.CELERY_SEND_TASK_SENT_EVENT = True
+            self.assertTrue(self.app.amqp.TaskProducer(conn).send_sent_event)
 
     def test_update_state(self):
 
-        @task
+        @self.app.task(shared=False)
         def yyy():
             pass
 
@@ -440,7 +422,7 @@ class test_tasks(AppCase):
 
     def test_repr(self):
 
-        @task
+        @self.app.task(shared=False)
         def task_test_repr():
             pass
 
@@ -448,113 +430,42 @@ class test_tasks(AppCase):
 
     def test_has___name__(self):
 
-        @task
+        @self.app.task(shared=False)
         def yyy2():
             pass
 
         self.assertTrue(yyy2.__name__)
 
-    def test_get_logger(self):
-        t1 = self.createTask('c.unittest.t.t1')
-        t1.push_request()
-        try:
-            logfh = WhateverIO()
-            logger = t1.get_logger(logfile=logfh, loglevel=0)
-            self.assertTrue(logger)
 
-            t1.request.loglevel = 3
-            logger = t1.get_logger(logfile=logfh, loglevel=None)
-            self.assertTrue(logger)
-        finally:
-            t1.pop_request()
-
-
-class test_TaskSet(AppCase):
-
-    def test_function_taskset(self):
-        with eager_tasks(self.app):
-            subtasks = [return_True_task.s(i) for i in range(1, 6)]
-            ts = TaskSet(subtasks)
-            res = ts.apply_async()
-            self.assertListEqual(res.join(), [True, True, True, True, True])
-
-    def test_counter_taskset(self):
-        increment_counter.count = 0
-        ts = TaskSet(tasks=[
-            increment_counter.s(),
-            increment_counter.s(increment_by=2),
-            increment_counter.s(increment_by=3),
-            increment_counter.s(increment_by=4),
-            increment_counter.s(increment_by=5),
-            increment_counter.s(increment_by=6),
-            increment_counter.s(increment_by=7),
-            increment_counter.s(increment_by=8),
-            increment_counter.s(increment_by=9),
-        ])
-        self.assertEqual(ts.total, 9)
-
-        consumer = increment_counter.get_consumer()
-        consumer.purge()
-        consumer.close()
-        taskset_res = ts.apply_async()
-        subtasks = taskset_res.subtasks
-        taskset_id = taskset_res.taskset_id
-        consumer = increment_counter.get_consumer()
-        for subtask in subtasks:
-            m = consumer.queues[0].get(accept=['pickle']).payload
-            self.assertDictContainsSubset({'taskset': taskset_id,
-                                           'task': increment_counter.name,
-                                           'id': subtask.id}, m)
-            increment_counter(
-                increment_by=m.get('kwargs', {}).get('increment_by'))
-        self.assertEqual(increment_counter.count, sum(range(1, 10)))
-
-    def test_named_taskset(self):
-        prefix = 'test_named_taskset-'
-        ts = TaskSet([return_True_task.subtask([1])])
-        res = ts.apply(taskset_id=prefix + uuid())
-        self.assertTrue(res.taskset_id.startswith(prefix))
-
-
-class test_apply_task(AppCase):
+class test_apply_task(TasksCase):
 
     def test_apply_throw(self):
         with self.assertRaises(KeyError):
-            raising.apply(throw=True)
-
-    def test_apply_no_magic_kwargs(self):
-        increment_counter.accept_magic_kwargs = False
-        try:
-            increment_counter.apply()
-        finally:
-            increment_counter.accept_magic_kwargs = True
+            self.raising.apply(throw=True)
 
     def test_apply_with_CELERY_EAGER_PROPAGATES_EXCEPTIONS(self):
-        raising.app.conf.CELERY_EAGER_PROPAGATES_EXCEPTIONS = True
-        try:
-            with self.assertRaises(KeyError):
-                raising.apply()
-        finally:
-            raising.app.conf.CELERY_EAGER_PROPAGATES_EXCEPTIONS = False
+        self.app.conf.CELERY_EAGER_PROPAGATES_EXCEPTIONS = True
+        with self.assertRaises(KeyError):
+            self.raising.apply()
 
     def test_apply(self):
-        increment_counter.count = 0
+        self.increment_counter.count = 0
 
-        e = increment_counter.apply()
+        e = self.increment_counter.apply()
         self.assertIsInstance(e, EagerResult)
         self.assertEqual(e.get(), 1)
 
-        e = increment_counter.apply(args=[1])
+        e = self.increment_counter.apply(args=[1])
         self.assertEqual(e.get(), 2)
 
-        e = increment_counter.apply(kwargs={'increment_by': 4})
+        e = self.increment_counter.apply(kwargs={'increment_by': 4})
         self.assertEqual(e.get(), 6)
 
         self.assertTrue(e.successful())
         self.assertTrue(e.ready())
         self.assertTrue(repr(e).startswith('<EagerResult:'))
 
-        f = raising.apply()
+        f = self.raising.apply()
         self.assertTrue(f.ready())
         self.assertFalse(f.successful())
         self.assertTrue(f.traceback)

celery/tests/worker/test_control.py

@@ -11,7 +11,6 @@ from mock import Mock, patch, call
 
 from celery.datastructures import AttributeDict
 from celery.five import Queue as FastQueue
-from celery.task import task
 from celery.utils import uuid
 from celery.utils.timer2 import Timer
 from celery.worker import WorkController as _WC
@@ -28,11 +27,6 @@ from celery.tests.case import AppCase
 hostname = socket.gethostname()
 
 
-@task(rate_limit=200)                   # for extra info in dump_tasks
-def mytask():
-    pass
-
-
 class WorkController(object):
     autoscaler = None
 
@@ -124,6 +118,11 @@ class test_ControlPanel(AppCase):
     def setup(self):
         self.panel = self.create_panel(consumer=Consumer(self.app))
 
+        @self.app.task(rate_limit=200, shared=False)
+        def mytask():
+            pass
+        self.mytask = mytask
+
     def create_state(self, **kwargs):
         kwargs.setdefault('app', self.app)
         return AttributeDict(kwargs)
@@ -199,24 +198,28 @@ class test_ControlPanel(AppCase):
 
     def test_time_limit(self):
         panel = self.create_panel(consumer=Mock())
-        th, ts = mytask.time_limit, mytask.soft_time_limit
+        th, ts = self.mytask.time_limit, self.mytask.soft_time_limit
         try:
             r = panel.handle('time_limit', arguments=dict(
-                task_name=mytask.name, hard=30, soft=10))
-            self.assertEqual((mytask.time_limit, mytask.soft_time_limit),
-                             (30, 10))
+                task_name=self.mytask.name, hard=30, soft=10))
+            self.assertEqual(
+                (self.mytask.time_limit, self.mytask.soft_time_limit),
+                (30, 10),
+            )
             self.assertIn('ok', r)
             r = panel.handle('time_limit', arguments=dict(
-                task_name=mytask.name, hard=None, soft=None))
-            self.assertEqual((mytask.time_limit, mytask.soft_time_limit),
-                             (None, None))
+                task_name=self.mytask.name, hard=None, soft=None))
+            self.assertEqual(
+                (self.mytask.time_limit, self.mytask.soft_time_limit),
+                (None, None),
+            )
             self.assertIn('ok', r)
 
             r = panel.handle('time_limit', arguments=dict(
                 task_name='248e8afya9s8dh921eh928', hard=30))
             self.assertIn('error', r)
         finally:
-            mytask.time_limit, mytask.soft_time_limit = th, ts
+            self.time_limit, self.soft_time_limit = th, ts
 
     def test_active_queues(self):
         import kombu
@@ -249,7 +252,7 @@ class test_ControlPanel(AppCase):
         self.panel.handle('report')
 
     def test_active(self):
-        r = TaskRequest(mytask.name, 'do re mi', (), {}, app=self.app)
+        r = TaskRequest(self.mytask.name, 'do re mi', (), {}, app=self.app)
         worker_state.active_requests.add(r)
         try:
             self.assertTrue(self.panel.handle('dump_active'))
@@ -331,7 +334,7 @@ class test_ControlPanel(AppCase):
         consumer = Consumer(self.app)
         panel = self.create_panel(consumer=consumer)
         self.assertFalse(panel.handle('dump_schedule'))
-        r = TaskRequest(mytask.name, 'CAFEBABE', (), {}, app=self.app)
+        r = TaskRequest(self.mytask.name, 'CAFEBABE', (), {}, app=self.app)
         consumer.timer.schedule.enter(
             consumer.timer.Entry(lambda x: x, (r, )),
             datetime.now() + timedelta(seconds=10))
@@ -343,14 +346,14 @@ class test_ControlPanel(AppCase):
     def test_dump_reserved(self):
         consumer = Consumer(self.app)
         worker_state.reserved_requests.add(
-            TaskRequest(mytask.name, uuid(), args=(2, 2), kwargs={},
+            TaskRequest(self.mytask.name, uuid(), args=(2, 2), kwargs={},
                         app=self.app),
         )
         try:
             panel = self.create_panel(consumer=consumer)
             response = panel.handle('dump_reserved', {'safe': True})
             self.assertDictContainsSubset(
-                {'name': mytask.name,
+                {'name': self.mytask.name,
                  'args': (2, 2),
                  'kwargs': {},
                  'hostname': socket.gethostname()},
@@ -377,7 +380,7 @@ class test_ControlPanel(AppCase):
         consumer = xConsumer()
         panel = self.create_panel(app=self.app, consumer=consumer)
 
-        task = self.app.tasks[mytask.name]
+        task = self.app.tasks[self.mytask.name]
         old_rate_limit = task.rate_limit
         try:
             panel.handle('rate_limit', arguments=dict(task_name=task.name,
@@ -406,7 +409,7 @@ class test_ControlPanel(AppCase):
         m = {'method': 'revoke',
              'destination': hostname,
              'arguments': {'task_id': tid,
-                           'task_name': mytask.name}}
+                           'task_name': self.mytask.name}}
         self.panel.handle_message(m, None)
         self.assertIn(tid, revoked)