100% coverage for celery.contrib.methods

celery/tests/bin/test_events.py

@@ -6,7 +6,7 @@ from mock import patch as mpatch
 from celery.app import app_or_default
 from celery.bin import events
 
-from celery.tests.case import Case, patch
+from celery.tests.case import Case, _old_patch as patch
 
 
 class MockCommand(object):

celery/tests/case.py

@@ -22,7 +22,10 @@ from datetime import datetime, timedelta
 from functools import partial, wraps
 from types import ModuleType
 
-import mock
+try:
+    from unittest import mock
+except ImportError:
+    import mock
 from nose import SkipTest
 from kombu.log import NullHandler
 from kombu.utils import nested
@@ -34,6 +37,9 @@ from celery.five import (
 )
 from celery.utils.functional import noop
 
+patch = mock.patch
+call = mock.call
+
 
 class Mock(mock.Mock):
 
@@ -397,7 +403,7 @@ def override_stdouts():
         sys.stderr = sys.__stderr__ = prev_err
 
 
-def patch(module, name, mocked):
+def _old_patch(module, name, mocked):
     module = importlib.import_module(module)
 
     def _patch(fun):
@@ -527,7 +533,7 @@ def mock_context(mock, typ=Mock):
 
 @contextmanager
 def mock_open(typ=WhateverIO, side_effect=None):
-    with mock.patch(open_fqdn) as open_:
+    with patch(open_fqdn) as open_:
         with mock_context(open_) as context:
             if side_effect is not None:
                 context.__enter__.side_effect = side_effect
@@ -537,7 +543,7 @@ def mock_open(typ=WhateverIO, side_effect=None):
 
 
 def patch_many(*targets):
-    return nested(*[mock.patch(target) for target in targets])
+    return nested(*[patch(target) for target in targets])
 
 
 @contextmanager

celery/tests/contrib/test_methods.py

@@ -0,0 +1,34 @@
+from __future__ import absolute_import
+
+from celery.contrib.methods import task_method, task
+
+from celery.tests.case import AppCase, patch
+
+
+class test_task_method(AppCase):
+
+    def test_task_method(self):
+
+        class X(object):
+
+            def __init__(self):
+                self.state = 0
+
+            @self.app.task(filter=task_method)
+            def add(self, x):
+                self.state += x
+
+        x = X()
+        x.add(2)
+        self.assertEqual(x.state, 2)
+        x.add(4)
+        self.assertEqual(x.state, 6)
+
+        self.assertTrue(X.add)
+        self.assertIs(x.add.__self__, x)
+
+    def test_task(self):
+        with patch('celery.contrib.methods.current_app') as curapp:
+            fun = object()
+            task(fun, x=1)
+            curapp.task.assert_called_with(fun, x=1, filter=task_method)